mark event-loop glue + defensive guards as no-cover
This commit is contained in:
parent
710276ffd8
commit
ba623b0862
4 changed files with 22 additions and 12 deletions
|
|
@ -34,7 +34,7 @@ def build_source(
|
||||||
source_id=cfg.id,
|
source_id=cfg.id,
|
||||||
)
|
)
|
||||||
if isinstance(cfg, HTTPSourceConfig):
|
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")
|
raise ValueError("HTTPSourceConfig.id is required")
|
||||||
return HTTPSource(source_id=cfg.id, urls=cfg.urls, headers=cfg.headers)
|
return HTTPSource(source_id=cfg.id, urls=cfg.urls, headers=cfg.headers)
|
||||||
if isinstance(cfg, S3SourceConfig):
|
if isinstance(cfg, S3SourceConfig):
|
||||||
|
|
@ -47,7 +47,7 @@ def build_source(
|
||||||
source_id=cfg.id,
|
source_id=cfg.id,
|
||||||
)
|
)
|
||||||
if isinstance(cfg, WebDAVSourceConfig):
|
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")
|
raise ValueError("WebDAVSourceConfig.id is required")
|
||||||
return WebDAVSource(
|
return WebDAVSource(
|
||||||
source_id=cfg.id,
|
source_id=cfg.id,
|
||||||
|
|
@ -59,4 +59,6 @@ def build_source(
|
||||||
include_patterns=cfg.include_patterns or None,
|
include_patterns=cfg.include_patterns or None,
|
||||||
supported_extensions=supported_extensions,
|
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__}"
|
||||||
|
)
|
||||||
|
|
|
||||||
|
|
@ -59,9 +59,10 @@ class FSPoller(BasePoller):
|
||||||
sweep_task.cancel()
|
sweep_task.cancel()
|
||||||
await asyncio.gather(watch_task, sweep_task, return_exceptions=True)
|
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
|
"""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():
|
while not self._stop.is_set():
|
||||||
try:
|
try:
|
||||||
await asyncio.wait_for(
|
await asyncio.wait_for(
|
||||||
|
|
@ -72,9 +73,15 @@ class FSPoller(BasePoller):
|
||||||
pass
|
pass
|
||||||
await self._sweep_once()
|
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
|
"""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:
|
try:
|
||||||
async for changes in awatch(
|
async for changes in awatch(
|
||||||
self._fs_source.root,
|
self._fs_source.root,
|
||||||
|
|
|
||||||
|
|
@ -33,9 +33,10 @@ class PeriodicPoller(BasePoller):
|
||||||
default_max_attempts=default_max_attempts,
|
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
|
# 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()
|
await self._sweep_once()
|
||||||
while not self._stop.is_set():
|
while not self._stop.is_set():
|
||||||
try:
|
try:
|
||||||
|
|
|
||||||
|
|
@ -83,7 +83,7 @@ class WorkerPool:
|
||||||
while not self._stop.is_set():
|
while not self._stop.is_set():
|
||||||
try:
|
try:
|
||||||
job = await self._jobs.claim_next(worker_id)
|
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)
|
logger.exception("claim_next failed in %s", worker_id)
|
||||||
await self._sleep_or_stop(self._poll_idle_s)
|
await self._sleep_or_stop(self._poll_idle_s)
|
||||||
continue
|
continue
|
||||||
|
|
@ -104,7 +104,7 @@ class WorkerPool:
|
||||||
reset = await self._jobs.reap_stale(self._claim_timeout_s)
|
reset = await self._jobs.reap_stale(self._claim_timeout_s)
|
||||||
if reset:
|
if reset:
|
||||||
logger.info("Reaper reset %d stale claim(s)", 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")
|
logger.exception("reaper failed")
|
||||||
|
|
||||||
async def _sleep_or_stop(self, seconds: float) -> None:
|
async def _sleep_or_stop(self, seconds: float) -> None:
|
||||||
|
|
@ -140,7 +140,7 @@ class WorkerPool:
|
||||||
e,
|
e,
|
||||||
)
|
)
|
||||||
return
|
return
|
||||||
except Exception as e:
|
except Exception as e: # pragma: no cover - pipeline classifier net
|
||||||
# Defensive: pipeline classifier should have caught everything.
|
# Defensive: pipeline classifier should have caught everything.
|
||||||
await self._jobs.mark_dead(job.id, f"unclassified: {e!r}")
|
await self._jobs.mark_dead(job.id, f"unclassified: {e!r}")
|
||||||
logger.exception("Unclassified error in job %s", job.id)
|
logger.exception("Unclassified error in job %s", job.id)
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue