from __future__ import annotations import base64 import contextlib import io import json import os import shutil import tempfile import threading import unittest from http.server import BaseHTTPRequestHandler, HTTPServer from pathlib import Path from typing import Any, Dict, List from urllib.parse import parse_qs, urlparse from unittest.mock import patch from src.cli import main TEST_TOKEN = "test-token" FIXTURE_INBOX = Path(__file__).resolve().parent / "fixtures" / "inbox" _MINIMAL_PNG_BASE64 = ( # 1x1 PNG, deterministic and small; avoids relying on external image tooling/libs. "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO5Fnc8AAAAASUVORK5CYII=" ) STUB_DEAL_FIELDS = [ {"key": "cf_event_date", "name": "Event date"}, {"key": "cf_venue", "name": "Venue"}, {"key": "cf_lat", "name": "Lat"}, {"key": "cf_lon", "name": "Lon"}, ] STUB_DEALS_PAGES = { 0: [ { "id": 100, "title": "In range", "cf_event_date": "2026-01-15", "cf_venue": "Venue A", "cf_lat": "52.1", "cf_lon": "4.9", } ], 2: [], } class _PipedriveStubHandler(BaseHTTPRequestHandler): deal_fields: List[Dict[str, Any]] = [] deals_pages: Dict[int, List[Dict[str, Any]]] = {} def do_GET(self) -> None: # noqa: N802 - required by http.server parsed = urlparse(self.path) qs = parse_qs(parsed.query) token = (qs.get("api_token") or [None])[0] if token != TEST_TOKEN: self._send_json(401, {"success": False, "error": "unauthorized"}) return if parsed.path.endswith("/dealFields"): self._send_json( 200, { "success": True, "data": list(self.deal_fields), "additional_data": {"pagination": {"more_items_in_collection": False}}, }, ) return if parsed.path.endswith("/deals"): start = int((qs.get("start") or ["0"])[0]) data = self.deals_pages.get(start, []) more = start == 0 and 2 in self.deals_pages payload: Dict[str, Any] = { "success": True, "data": list(data), "additional_data": { "pagination": { "more_items_in_collection": more, "next_start": 2 if more else None, } }, } self._send_json(200, payload) return self._send_json(404, {"success": False, "error": "not found"}) def _send_json(self, status: int, payload: Any) -> None: body = json.dumps(payload, ensure_ascii=True).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def log_message(self, format: str, *args: object) -> None: # noqa: A002 return @contextlib.contextmanager def _run_stub_server(handler_cls: type[BaseHTTPRequestHandler]) -> Any: server = HTTPServer(("127.0.0.1", 0), handler_cls) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() host, port = server.server_address try: yield f"http://{host}:{port}/v1" finally: server.shutdown() server.server_close() thread.join(timeout=5) def _write_mapping_file(path: Path) -> None: path.write_text( json.dumps( { "schema_version": 1, "fields": { "event_date": {"field_name": "Event date"}, "venue_name": {"field_name": "Venue"}, "lat": {"field_name": "Lat"}, "lon": {"field_name": "Lon"}, }, } ), encoding="utf-8", ) def _copy_fixture_inbox(dest: Path) -> None: dest.mkdir(parents=True, exist_ok=True) for item in FIXTURE_INBOX.iterdir(): if item.is_file(): if item.suffix.lower() in {".jpg", ".jpeg", ".png"}: continue shutil.copy2(item, dest / item.name) # Ensure there is always at least one valid image in the inbox for hermetic CLI E2E. (dest / "a.jpg").write_bytes(base64.b64decode(_MINIMAL_PNG_BASE64)) def _first_image_path(inbox: Path) -> Path: for ext in (".jpg", ".jpeg", ".png"): matches = sorted(inbox.glob(f"*{ext}")) if matches: return matches[0] raise AssertionError("No fixture images found to run CLI E2E test.") def _run_cli(*, env: Dict[str, str], argv: List[str]) -> tuple[int, str]: stdout = io.StringIO() with patch.dict(os.environ, env, clear=True), contextlib.redirect_stdout(stdout): code = main(argv) return code, stdout.getvalue() class TestCliE2E(unittest.TestCase): def _assert_cli_ok(self, *, env: Dict[str, str], argv: List[str]) -> str: code, out = _run_cli(env=env, argv=argv) self.assertEqual(code, 0, msg=f"CLI failed for argv={argv!r} (code={code})") return out def _make_env(self, *, base_url: str, mapping_path: Path) -> Dict[str, str]: return { "PIPE_DRIVE_API_TOKEN": TEST_TOKEN, "PIPEDRIVE_BASE_URL": base_url, "PIPEDRIVE_ALLOW_INSECURE_HTTP": "1", "PIPEDRIVE_FIELD_MAPPING_PATH": str(mapping_path), } def _step_ingest(self, *, env: Dict[str, str], inbox: Path, out_dir: Path) -> None: self._assert_cli_ok(env=env, argv=["ingest", "--inbox", str(inbox), "--out", str(out_dir)]) self.assertTrue((out_dir / "manifest.json").exists()) def _step_exif(self, *, env: Dict[str, str], sample_image: Path) -> None: out = self._assert_cli_ok(env=env, argv=["exif", "--path", str(sample_image)]) self.assertTrue(out.strip()) def _step_match(self, *, env: Dict[str, str], inbox: Path) -> None: out = self._assert_cli_ok( env=env, argv=[ "match", "--inbox", str(inbox), "--from", "2026-01-01", "--to", "2026-01-31", ], ) lines = [json.loads(line) for line in out.splitlines() if line.strip()] self.assertTrue(lines) def _step_export(self, *, env: Dict[str, str], inbox: Path, out_dir: Path, sample_image: Path) -> None: self._assert_cli_ok(env=env, argv=["export", "gbp", "--inbox", str(inbox), "--out", str(out_dir)]) rel = sample_image.relative_to(inbox) jpg_path = out_dir / "gbp" / rel.parent / f"{rel.stem}.jpg" png_path = out_dir / "gbp" / rel.parent / f"{rel.stem}.png" self.assertTrue(jpg_path.exists()) self.assertTrue(png_path.exists()) def _step_run(self, *, env: Dict[str, str], inbox: Path, out_dir: Path, run_id: str) -> Path: self._assert_cli_ok( env=env, argv=[ "run", "--inbox", str(inbox), "--out", str(out_dir), "--run-id", run_id, "--no-match", ], ) audit_dir = out_dir / "audit" / run_id self.assertTrue((audit_dir / "summary.json").exists()) self.assertTrue((audit_dir / "audit.jsonl").exists()) return audit_dir def _step_report(self, *, env: Dict[str, str], out_dir: Path, run_id: str, audit_dir: Path) -> None: self._assert_cli_ok(env=env, argv=["report", "--out", str(out_dir), "--run-id", run_id]) self.assertTrue((audit_dir / "audit.json").exists()) self.assertTrue((audit_dir / "audit.csv").exists()) def _run_workflow(self, *, base_url: str, tmp: str) -> None: inbox = Path(tmp) / "inbox" out_dir = Path(tmp) / "out" _copy_fixture_inbox(inbox) mapping_path = Path(tmp) / "mapping.json" _write_mapping_file(mapping_path) env = self._make_env(base_url=base_url, mapping_path=mapping_path) sample_image = _first_image_path(inbox) self._step_ingest(env=env, inbox=inbox, out_dir=out_dir) self._step_exif(env=env, sample_image=sample_image) self._step_match(env=env, inbox=inbox) self._step_export(env=env, inbox=inbox, out_dir=out_dir, sample_image=sample_image) run_id = "test-run" audit_dir = self._step_run(env=env, inbox=inbox, out_dir=out_dir, run_id=run_id) self._step_report(env=env, out_dir=out_dir, run_id=run_id, audit_dir=audit_dir) def test_cli_e2e_workflow(self) -> None: _PipedriveStubHandler.deal_fields = list(STUB_DEAL_FIELDS) _PipedriveStubHandler.deals_pages = dict(STUB_DEALS_PAGES) with _run_stub_server(_PipedriveStubHandler) as base_url: with tempfile.TemporaryDirectory() as tmp: self._run_workflow(base_url=base_url, tmp=tmp)