| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374 |
- from __future__ import annotations
- from uuid import uuid4
- import scripts.run_acquisition as cli
- from acquisition.runner import RunBatchResult
- class DummyTransaction:
- def __init__(self, conn):
- self.conn = conn
- def __enter__(self):
- return self.conn
- def __exit__(self, exc_type, exc, tb):
- return False
- def test_run_acquisition_cli_is_thin_delegate(monkeypatch, capsys):
- captured = {}
- batch_id = uuid4()
- settings = object()
- db_config = object()
- conn = object()
- monkeypatch.setattr(cli.Settings, "from_env", lambda env_file: settings)
- monkeypatch.setattr(cli.CreationDbConfig, "from_env", lambda env_file: db_config)
- monkeypatch.setattr(cli, "transaction", lambda config: DummyTransaction(conn))
- monkeypatch.setattr(cli, "PostgresAcquisitionRepository", lambda c: ("repo", c))
- def fake_run_batch(repo, **kwargs):
- captured["repo"] = repo
- captured.update(kwargs)
- return RunBatchResult(
- run_id=uuid4(),
- total_jobs=1,
- done=1,
- partial=0,
- failed=0,
- skipped=0,
- )
- monkeypatch.setattr(cli, "run_batch", fake_run_batch)
- rc = cli.main(
- [
- "--batch-id",
- str(batch_id),
- "--platform",
- "weixin",
- "--search-limit",
- "2",
- "--display-limit",
- "1",
- "--no-classify",
- "--run-key",
- "rk",
- "--env-file",
- ".env.test",
- ]
- )
- out = capsys.readouterr().out
- assert rc == 0
- assert '"done": 1' in out
- assert captured["repo"] == ("repo", conn)
- assert captured["batch_id"] == batch_id
- assert captured["settings"] is settings
- assert captured["platforms"] == ("weixin",)
- assert captured["search_limit"] == 2
- assert captured["display_limit"] == 1
- assert captured["classify"] is False
- assert captured["run_key"] == "rk"
|
