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 

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

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

18 

19 

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

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

22 

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

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

25 points from drifting and accidentally giving replayed rows a larger 

26 prompt/storage budget than fresh requests. 

27 """ 

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

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

30 return None 

31 

32 

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

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

35 

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

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

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

39 rejects booleans (and floats) instead. 

40 """ 

41 max_results = data.get("max_results") 

42 if max_results is not None and ( 

43 type(max_results) is not int 

44 or not MAX_RESULTS_MIN <= max_results <= MAX_RESULTS_MAX 

45 ): 

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

47 

48 time_period = data.get("time_period") 

49 if time_period is not None and ( 

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

51 ): 

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

53 

54 return None