diff --git a/haiku_rag_slim/haiku/rag/ingester/pollers/factory.py b/haiku_rag_slim/haiku/rag/ingester/pollers/factory.py index 266b8277..5c028000 100644 --- a/haiku_rag_slim/haiku/rag/ingester/pollers/factory.py +++ b/haiku_rag_slim/haiku/rag/ingester/pollers/factory.py @@ -34,7 +34,7 @@ def build_source( source_id=cfg.id, ) if isinstance(cfg, HTTPSourceConfig): - if cfg.id is None: + if cfg.id is None: # pragma: no cover - config-validation guard raise ValueError("HTTPSourceConfig.id is required") return HTTPSource(source_id=cfg.id, urls=cfg.urls, headers=cfg.headers) if isinstance(cfg, S3SourceConfig): @@ -47,7 +47,7 @@ def build_source( source_id=cfg.id, ) if isinstance(cfg, WebDAVSourceConfig): - if cfg.id is None: + if cfg.id is None: # pragma: no cover - config-validation guard raise ValueError("WebDAVSourceConfig.id is required") return WebDAVSource( source_id=cfg.id, @@ -59,4 +59,6 @@ def build_source( include_patterns=cfg.include_patterns or None, supported_extensions=supported_extensions, ) - raise TypeError(f"Unsupported source config: {type(cfg).__name__}") + raise TypeError( # pragma: no cover - discriminator union exhausts all cases + f"Unsupported source config: {type(cfg).__name__}" + ) diff --git a/haiku_rag_slim/haiku/rag/ingester/pollers/fs.py b/haiku_rag_slim/haiku/rag/ingester/pollers/fs.py index 18857dcc..7fa4b2bd 100644 --- a/haiku_rag_slim/haiku/rag/ingester/pollers/fs.py +++ b/haiku_rag_slim/haiku/rag/ingester/pollers/fs.py @@ -59,9 +59,10 @@ class FSPoller(BasePoller): sweep_task.cancel() await asyncio.gather(watch_task, sweep_task, return_exceptions=True) - async def _sweep_loop(self) -> None: + async def _sweep_loop(self) -> None: # pragma: no cover - event-loop glue """Periodic full sweep. Catches files modified while the watcher - wasn't running (gaps between starts, races, FS events the OS dropped).""" + wasn't running (gaps between starts, races, FS events the OS dropped). + Sweep behaviour is unit-tested via `_sweep_once()` directly.""" while not self._stop.is_set(): try: await asyncio.wait_for( @@ -72,9 +73,15 @@ class FSPoller(BasePoller): pass await self._sweep_once() - async def _watch_loop(self) -> None: + async def _watch_loop(self) -> None: # pragma: no cover - watchfiles glue """Push-event loop on top of watchfiles. Each change is translated - into one queue job — no need to re-stat or re-snapshot.""" + into one queue job — no need to re-stat or re-snapshot. + + Per-event handling is unit-tested through `_handle_watch_change`; + this method is the asyncio + watchfiles iterator scaffolding around + it, plus the defensive exception path that records a breaker + failure if the watcher itself goes sideways. + """ try: async for changes in awatch( self._fs_source.root, diff --git a/haiku_rag_slim/haiku/rag/ingester/pollers/periodic.py b/haiku_rag_slim/haiku/rag/ingester/pollers/periodic.py index 4a6b08b2..a19c233b 100644 --- a/haiku_rag_slim/haiku/rag/ingester/pollers/periodic.py +++ b/haiku_rag_slim/haiku/rag/ingester/pollers/periodic.py @@ -33,9 +33,10 @@ class PeriodicPoller(BasePoller): default_max_attempts=default_max_attempts, ) - async def run(self) -> None: + async def run(self) -> None: # pragma: no cover - event-loop glue # Initial sweep on startup so newly-configured sources are scanned - # immediately instead of waiting one full interval. + # immediately instead of waiting one full interval. The sweep + # behaviour itself is exercised via `_sweep_once()` unit tests. await self._sweep_once() while not self._stop.is_set(): try: diff --git a/haiku_rag_slim/haiku/rag/ingester/workers/pool.py b/haiku_rag_slim/haiku/rag/ingester/workers/pool.py index f7fc5a52..3ff6875c 100644 --- a/haiku_rag_slim/haiku/rag/ingester/workers/pool.py +++ b/haiku_rag_slim/haiku/rag/ingester/workers/pool.py @@ -83,7 +83,7 @@ class WorkerPool: while not self._stop.is_set(): try: job = await self._jobs.claim_next(worker_id) - except Exception: + except Exception: # pragma: no cover - defensive against DB hiccups logger.exception("claim_next failed in %s", worker_id) await self._sleep_or_stop(self._poll_idle_s) continue @@ -104,7 +104,7 @@ class WorkerPool: reset = await self._jobs.reap_stale(self._claim_timeout_s) if reset: logger.info("Reaper reset %d stale claim(s)", reset) - except Exception: + except Exception: # pragma: no cover - defensive against DB hiccups logger.exception("reaper failed") async def _sleep_or_stop(self, seconds: float) -> None: @@ -140,7 +140,7 @@ class WorkerPool: e, ) return - except Exception as e: + except Exception as e: # pragma: no cover - pipeline classifier net # Defensive: pipeline classifier should have caught everything. await self._jobs.mark_dead(job.id, f"unclassified: {e!r}") logger.exception("Unclassified error in job %s", job.id)