Coverage for src/local_deep_research/web/routes/research_validation.py: 100%

1"""Validation for user-supplied search overrides. 

2 

3Shared by the research-start route and the queue dispatcher so an override 

4is validated identically whether it arrives on a fresh request or is replayed 

5from a persisted queued row. 

6""" 

7 

8from collections.abc import Mapping 

9from typing import Final, TypeAlias 

10 

11MAX_RESULTS_MIN: Final = 1 

12MAX_RESULTS_MAX: Final = 50 

13MAX_QUERY_LENGTH: Final = 10_000 

14ALLOWED_TIME_PERIODS: Final = frozenset({"d", "w", "m", "y", "all"}) 

15 

16ITERATIONS_MIN: Final = 1 

17ITERATIONS_MAX: Final = 100 

18QUESTIONS_PER_ITERATION_MIN: Final = 1 

19QUESTIONS_PER_ITERATION_MAX: Final = 20 

20 

21JsonScalar: TypeAlias = str | int | float | bool | None 

22JsonValue: TypeAlias = JsonScalar | list["JsonValue"] | dict[str, "JsonValue"] 

23 

24 

25def validate_research_query_length(query: object) -> str | None: 

26 """Return the shared query-length error for an over-cap string. 

27 

28 Research requests can reach the worker either immediately or after a 

29 persisted queue replay. Keeping this check here prevents the two entry 

30 points from drifting and accidentally giving replayed rows a larger 

31 prompt/storage budget than fresh requests. 

32 """ 

33 if isinstance(query, str) and len(query) > MAX_QUERY_LENGTH: 

34 return f"Query exceeds maximum length of {MAX_QUERY_LENGTH} characters" 

35 return None 

36 

37 

38def validate_search_overrides(data: Mapping[str, JsonValue]) -> str | None: 

39 """Return an error message for an invalid override, or None if valid. 

40 

41 ``type(x) is not int`` is deliberate rather than ``isinstance``: ``bool`` 

42 is a subclass of ``int``, so ``isinstance(True, int)`` is True and would 

43 let ``max_results: true`` silently coerce to 1. The exact-type check 

44 rejects booleans (and floats) instead. 

45 """ 

46 max_results = data.get("max_results") 

47 if max_results is not None and ( 

48 type(max_results) is not int 

49 or not MAX_RESULTS_MIN <= max_results <= MAX_RESULTS_MAX 

50 ): 

51 return "max_results must be an integer between 1 and 50" 

52 

53 time_period = data.get("time_period") 

54 if time_period is not None and ( 

55 type(time_period) is not str or time_period not in ALLOWED_TIME_PERIODS 

56 ): 

57 return "time_period must be one of: d, w, m, y, all" 

58 

59 iterations = data.get("iterations") 

60 if iterations is not None and ( 

61 type(iterations) is not int 

62 or not ITERATIONS_MIN <= iterations <= ITERATIONS_MAX 

63 ): 

64 return f"iterations must be an integer between {ITERATIONS_MIN} and {ITERATIONS_MAX}" 

65 

66 questions = data.get("questions_per_iteration") 

67 if questions is not None and ( 

68 type(questions) is not int 

69 or not QUESTIONS_PER_ITERATION_MIN 

70 <= questions 

71 <= QUESTIONS_PER_ITERATION_MAX 

72 ): 

73 return ( 

74 f"questions_per_iteration must be an integer between " 

75 f"{QUESTIONS_PER_ITERATION_MIN} and {QUESTIONS_PER_ITERATION_MAX}" 

76 ) 

77 

78 return None