(
cached_push: Dict[str, Any],
*,
days: int,
limit_per_source: int,
target_date: str,
sources: Dict[str, Any],
)
| 1138 | |
| 1139 | |
| 1140 | def _normalized_cache_list(value: Any) -> List[str]: |
| 1141 | return sorted(_string_list(value)) |
| 1142 | |
| 1143 | |
| 1144 | def _cache_bool(value: Any) -> bool: |
| 1145 | if isinstance(value, bool): |
| 1146 | return value |
| 1147 | return str(value or "").strip().lower() in {"1", "true", "yes", "on"} |
| 1148 | |
| 1149 | |
| 1150 | def _daily_cache_matches_settings( |
| 1151 | cached_push: Dict[str, Any], |
| 1152 | *, |
| 1153 | days: int, |
| 1154 | limit_per_source: int, |
| 1155 | target_date: str, |
| 1156 | sources: Dict[str, Any], |
| 1157 | ) -> bool: |
| 1158 | metadata = dict(cached_push.get("metadata") or {}) |
| 1159 | try: |
| 1160 | cached_daily_limit = int(_to_float(metadata.get("daily_limit"), -1)) |
| 1161 | except (TypeError, ValueError): |
| 1162 | cached_daily_limit = -1 |
| 1163 | if cached_daily_limit != int(limit_per_source): |
| 1164 | return False |
| 1165 | if int(_to_float(metadata.get("limit_per_source"), -1)) != int(limit_per_source): |
| 1166 | return False |
| 1167 | if int(_to_float(metadata.get("relevance_threshold"), -1)) != _configured_relevance_threshold(): |
| 1168 | return False |
| 1169 | if int(_to_float(metadata.get("fetch_days"), -1)) != max(1, int(days or 1)): |
| 1170 | return False |
| 1171 | requested_target_date = str(target_date or "").strip() |
| 1172 | cached_target_date = str(metadata.get("target_date") or "").strip() |
| 1173 | if requested_target_date and cached_target_date != requested_target_date: |
| 1174 | return False |
| 1175 | if cached_target_date and cached_target_date != requested_target_date: |
| 1176 | return False |
| 1177 |
no test coverage detected