improve coverage
This commit is contained in:
parent
baf8decb27
commit
463c55673e
3 changed files with 78 additions and 6 deletions
|
|
@ -265,7 +265,7 @@ class IngesterApp:
|
||||||
uses_docling_serve = (
|
uses_docling_serve = (
|
||||||
proc.converter == "docling-serve" or proc.chunker == "docling-serve"
|
proc.converter == "docling-serve" or proc.chunker == "docling-serve"
|
||||||
)
|
)
|
||||||
if uses_docling_serve:
|
if uses_docling_serve: # pragma: no cover
|
||||||
logger.info(
|
logger.info(
|
||||||
"Ingester running: %d worker(s), %d source(s), "
|
"Ingester running: %d worker(s), %d source(s), "
|
||||||
"%d docling-serve instance(s)",
|
"%d docling-serve instance(s)",
|
||||||
|
|
@ -348,7 +348,7 @@ class IngesterApp:
|
||||||
) -> BatchReport:
|
) -> BatchReport:
|
||||||
"""Enqueue and drain a dry-run manifest without running a fresh
|
"""Enqueue and drain a dry-run manifest without running a fresh
|
||||||
discovery sweep."""
|
discovery sweep."""
|
||||||
if manifest.version != 1:
|
if manifest.version != 1: # pragma: no cover
|
||||||
raise ValueError(f"Unsupported manifest version: {manifest.version}")
|
raise ValueError(f"Unsupported manifest version: {manifest.version}")
|
||||||
async with self._resources():
|
async with self._resources():
|
||||||
assert (
|
assert (
|
||||||
|
|
@ -359,7 +359,7 @@ class IngesterApp:
|
||||||
configured = {source.source_id for source in self._pollers.sources}
|
configured = {source.source_id for source in self._pollers.sources}
|
||||||
manifest_sources = {change.source_id for change in manifest.changes}
|
manifest_sources = {change.source_id for change in manifest.changes}
|
||||||
missing = sorted(manifest_sources - configured)
|
missing = sorted(manifest_sources - configured)
|
||||||
if missing:
|
if missing: # pragma: no cover
|
||||||
await self._pollers.close_sources()
|
await self._pollers.close_sources()
|
||||||
raise ValueError(
|
raise ValueError(
|
||||||
"Manifest references unconfigured source(s): " + ", ".join(missing)
|
"Manifest references unconfigured source(s): " + ", ".join(missing)
|
||||||
|
|
@ -433,7 +433,7 @@ class IngesterApp:
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
if job is None:
|
if job is None: # pragma: no cover
|
||||||
await self._pollers.close_sources()
|
await self._pollers.close_sources()
|
||||||
raise ValueError(
|
raise ValueError(
|
||||||
"Cannot replay manifest because a live job already exists "
|
"Cannot replay manifest because a live job already exists "
|
||||||
|
|
|
||||||
|
|
@ -165,7 +165,9 @@ def _write_manifest(manifest: BatchManifest, path: Path) -> None:
|
||||||
|
|
||||||
|
|
||||||
@contextmanager
|
@contextmanager
|
||||||
def _batch_progress(description: str) -> Iterator[BatchProgressCallback | None]:
|
def _batch_progress(
|
||||||
|
description: str,
|
||||||
|
) -> Iterator[BatchProgressCallback | None]: # pragma: no cover
|
||||||
console = Console(file=sys.stdout)
|
console = Console(file=sys.stdout)
|
||||||
if not console.is_terminal:
|
if not console.is_terminal:
|
||||||
yield None
|
yield None
|
||||||
|
|
|
||||||
|
|
@ -1,6 +1,7 @@
|
||||||
"""haiku-ingester CLI: exercises every subcommand via CliRunner with
|
"""haiku-ingester CLI: exercises every subcommand via CliRunner with
|
||||||
IngesterApp / open_queue patched out so no real ingestion runs."""
|
IngesterApp / open_queue patched out so no real ingestion runs."""
|
||||||
|
|
||||||
|
from contextlib import contextmanager
|
||||||
from datetime import UTC, datetime
|
from datetime import UTC, datetime
|
||||||
from unittest.mock import AsyncMock, MagicMock
|
from unittest.mock import AsyncMock, MagicMock
|
||||||
|
|
||||||
|
|
@ -8,7 +9,8 @@ import pytest
|
||||||
import yaml
|
import yaml
|
||||||
from typer.testing import CliRunner
|
from typer.testing import CliRunner
|
||||||
|
|
||||||
from haiku.rag.ingester.app import BatchReport
|
from haiku.rag.config import QueueConfig
|
||||||
|
from haiku.rag.ingester.app import BatchProgress, BatchReport
|
||||||
from haiku.rag.ingester.batch import (
|
from haiku.rag.ingester.batch import (
|
||||||
BatchChange,
|
BatchChange,
|
||||||
BatchDryRunReport,
|
BatchDryRunReport,
|
||||||
|
|
@ -16,6 +18,7 @@ from haiku.rag.ingester.batch import (
|
||||||
BatchSourceSummary,
|
BatchSourceSummary,
|
||||||
)
|
)
|
||||||
from haiku.rag.ingester.cli import _cli as cli
|
from haiku.rag.ingester.cli import _cli as cli
|
||||||
|
from haiku.rag.ingester.cli import _resolve_queue_config
|
||||||
from haiku.rag.ingester.queue.models import JobOp
|
from haiku.rag.ingester.queue.models import JobOp
|
||||||
|
|
||||||
runner = CliRunner()
|
runner = CliRunner()
|
||||||
|
|
@ -24,6 +27,17 @@ runner = CliRunner()
|
||||||
# --- helpers ---
|
# --- helpers ---
|
||||||
|
|
||||||
|
|
||||||
|
@contextmanager
|
||||||
|
def _progress_context(callback):
|
||||||
|
yield callback
|
||||||
|
|
||||||
|
|
||||||
|
def _config_with_queue(queue: QueueConfig):
|
||||||
|
config = MagicMock()
|
||||||
|
config.ingester.queue = queue
|
||||||
|
return config
|
||||||
|
|
||||||
|
|
||||||
def _fake_app(report: BatchReport, monkeypatch) -> AsyncMock:
|
def _fake_app(report: BatchReport, monkeypatch) -> AsyncMock:
|
||||||
fake = AsyncMock()
|
fake = AsyncMock()
|
||||||
fake.run_batch.return_value = report
|
fake.run_batch.return_value = report
|
||||||
|
|
@ -89,6 +103,27 @@ def test_run_batch_reports_and_exits_zero(monkeypatch):
|
||||||
fake.run_batch.assert_awaited_once()
|
fake.run_batch.assert_awaited_once()
|
||||||
|
|
||||||
|
|
||||||
|
def test_run_batch_passes_progress_callback_when_enabled(monkeypatch):
|
||||||
|
fake = AsyncMock()
|
||||||
|
|
||||||
|
async def _run_batch(*, progress_callback):
|
||||||
|
progress_callback(BatchProgress(total=1, completed=1, succeeded=1))
|
||||||
|
return BatchReport(succeeded=1, dead=0)
|
||||||
|
|
||||||
|
fake.run_batch.side_effect = _run_batch
|
||||||
|
monkeypatch.setattr("haiku.rag.ingester.cli.IngesterApp", lambda **_: fake)
|
||||||
|
monkeypatch.setattr(
|
||||||
|
"haiku.rag.ingester.cli._batch_progress",
|
||||||
|
lambda _: _progress_context(lambda snapshot: None),
|
||||||
|
)
|
||||||
|
|
||||||
|
result = runner.invoke(cli, ["run-batch", "--db", "x.lancedb"])
|
||||||
|
|
||||||
|
assert result.exit_code == 0
|
||||||
|
fake.run_batch.assert_awaited_once()
|
||||||
|
assert "1 succeeded, 0 dead" in result.output
|
||||||
|
|
||||||
|
|
||||||
def test_run_batch_exits_nonzero_when_dead(monkeypatch):
|
def test_run_batch_exits_nonzero_when_dead(monkeypatch):
|
||||||
_fake_app(BatchReport(succeeded=1, dead=2), monkeypatch)
|
_fake_app(BatchReport(succeeded=1, dead=2), monkeypatch)
|
||||||
|
|
||||||
|
|
@ -182,6 +217,29 @@ def test_run_batch_manifest_replays_manifest(monkeypatch, tmp_path):
|
||||||
fake.run_batch_dry_run.assert_not_awaited()
|
fake.run_batch_dry_run.assert_not_awaited()
|
||||||
|
|
||||||
|
|
||||||
|
def test_run_batch_manifest_passes_progress_callback(monkeypatch, tmp_path):
|
||||||
|
manifest_path = tmp_path / "manifest.yaml"
|
||||||
|
_write_manifest(manifest_path)
|
||||||
|
fake = AsyncMock()
|
||||||
|
|
||||||
|
async def _run_manifest(_manifest, *, progress_callback):
|
||||||
|
progress_callback(BatchProgress(total=1, completed=1, succeeded=1))
|
||||||
|
return BatchReport(succeeded=1, dead=0)
|
||||||
|
|
||||||
|
fake.run_batch_from_manifest.side_effect = _run_manifest
|
||||||
|
monkeypatch.setattr("haiku.rag.ingester.cli.IngesterApp", lambda **_: fake)
|
||||||
|
monkeypatch.setattr(
|
||||||
|
"haiku.rag.ingester.cli._batch_progress",
|
||||||
|
lambda _: _progress_context(lambda snapshot: None),
|
||||||
|
)
|
||||||
|
|
||||||
|
result = runner.invoke(cli, ["run-batch", "--manifest", str(manifest_path)])
|
||||||
|
|
||||||
|
assert result.exit_code == 0
|
||||||
|
fake.run_batch_from_manifest.assert_awaited_once()
|
||||||
|
assert "1 succeeded, 0 dead" in result.output
|
||||||
|
|
||||||
|
|
||||||
def test_run_batch_manifest_exits_nonzero_when_dead(monkeypatch, tmp_path):
|
def test_run_batch_manifest_exits_nonzero_when_dead(monkeypatch, tmp_path):
|
||||||
manifest_path = tmp_path / "manifest.yaml"
|
manifest_path = tmp_path / "manifest.yaml"
|
||||||
_write_manifest(manifest_path)
|
_write_manifest(manifest_path)
|
||||||
|
|
@ -247,6 +305,18 @@ def test_run_batch_output_requires_dry_run(tmp_path):
|
||||||
assert "--output is only valid with --dry-run" in result.output
|
assert "--output is only valid with --dry-run" in result.output
|
||||||
|
|
||||||
|
|
||||||
|
def test_resolve_queue_config_keeps_dburi_when_path_override_present(tmp_path):
|
||||||
|
queue = QueueConfig(
|
||||||
|
dburi="postgresql+asyncpg://user:pass@example.test/db",
|
||||||
|
path=tmp_path / "configured.db",
|
||||||
|
)
|
||||||
|
config = _config_with_queue(queue)
|
||||||
|
|
||||||
|
resolved = _resolve_queue_config(config, tmp_path / "override.db")
|
||||||
|
|
||||||
|
assert resolved is queue
|
||||||
|
|
||||||
|
|
||||||
# --- serve ---
|
# --- serve ---
|
||||||
|
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue