from datetime import datetime, timezone from pathlib import Path from textwrap import dedent import yaml from typer.testing import CliRunner from job_research.apec.adapter import ApecSearchResult from job_research.cli import app def _fixed_now() -> datetime: return datetime(2026, 6, 1, 10, 0, 0, tzinfo=timezone.utc) def _apec_detail_html( *, title: str = "Role From YAML", company: str = "Example Corp", location: str = "Paris - 75", contract: str = "CDI", description: str = "Build pipelines", profile: str = "Python / SQL", source_job_id: str = "178554452W", published_at: str = "20/04/2026", updated_at: str = "02/06/2026", ) -> str: return dedent( f"""

{title}

Ref. Apec : {source_job_id}
  • {company}
  • 1 {contract}
  • {location}

Publiée le {published_at} Actualisée le {updated_at}

{company}

Salaire

A partir de 70 k€ brut annuel

Prise de poste

Dès que possible

Expérience

Minimum 7 ans

Descriptif du poste

{description}

Profil recherché

{profile}

Compétences attendues

Ignored

Entreprise

Ignored

Ignored recruiter info
""" ).strip() def test_fetch_apec_reads_profile_and_writes_run_artifacts(tmp_path, monkeypatch) -> None: data_root = tmp_path / "data" data_root.mkdir() (data_root / "candidate-profile.yaml").write_text( dedent( """ target_roles: - Role From YAML """ ).strip(), encoding="utf-8", ) html = _apec_detail_html() class FakeApecAdapter: instances: list["FakeApecAdapter"] = [] def __init__(self, max_listings: int = 50) -> None: self.max_listings = max_listings self.search_calls: list[list[str]] = [] self.fetch_calls: list[str] = [] FakeApecAdapter.instances.append(self) def search(self, queries: list[str]) -> list[ApecSearchResult]: self.search_calls.append(list(queries)) return [ApecSearchResult(url="https://example.test/job/123", source_job_id="job-123")] def fetch_listing_html(self, url: str) -> str: self.fetch_calls.append(url) return html monkeypatch.setattr("job_research.cli.ApecAdapter", FakeApecAdapter) monkeypatch.setattr( "job_research.cli._utc_now", _fixed_now, ) result = CliRunner().invoke(app, ["fetch-apec", "--data-root", str(data_root)]) assert result.exit_code == 0 assert "query=1" in result.stdout assert "fetched=1" in result.stdout assert "normalized=1" in result.stdout assert "deduplicated=1" in result.stdout assert "failed=0" in result.stdout adapter = FakeApecAdapter.instances[0] assert adapter.max_listings == 50 assert adapter.search_calls == [["Role From YAML"]] assert adapter.fetch_calls == ["https://example.test/job/123"] runs_root = data_root / "apec" / "runs" run_dirs = list(runs_root.iterdir()) assert len(run_dirs) == 1 run_dir = run_dirs[0] snapshot_files = list((run_dir / "snapshots").glob("*.html")) assert len(snapshot_files) == 1 assert snapshot_files[0].name == "job-123.html" assert snapshot_files[0].read_text(encoding="utf-8") == html listings_payload = yaml.safe_load((run_dir / "listings.yaml").read_text(encoding="utf-8")) assert listings_payload == [ { "source": "apec", "source_job_id": "job-123", "url": "https://example.test/job/123", "title": "Role From YAML", "company": "Example Corp", "location": "Paris - 75", "contract_type": "CDI", "description_text": "Build pipelines", "published_at": "2026-04-20", "fetched_at": "2026-06-01T10:00:00Z", "warnings": [], } ] run_meta_payload = yaml.safe_load((run_dir / "run-meta.yaml").read_text(encoding="utf-8")) assert run_meta_payload == { "run_id": "2026-06-01T10-00-00Z", "run_started_at": "2026-06-01T10:00:00Z", "derived_queries": ["Role From YAML"], "fetched_count": 1, "normalized_count": 1, "deduplicated_count": 1, "failed_count": 0, "listing_errors": [], } def test_fetch_apec_records_partial_failures_without_losing_successful_listings( tmp_path, monkeypatch, ) -> None: data_root = tmp_path / "data" data_root.mkdir() (data_root / "candidate-profile.yaml").write_text( dedent( """ target_roles: - Role From YAML """ ).strip(), encoding="utf-8", ) html = _apec_detail_html() good_result = ApecSearchResult(url="https://example.test/job/123", source_job_id="job-123") bad_result = ApecSearchResult(url="https://example.test/job/456", source_job_id="job-456") class MixedApecAdapter: instances: list["MixedApecAdapter"] = [] def __init__(self, max_listings: int = 50) -> None: self.max_listings = max_listings self.search_calls: list[list[str]] = [] self.fetch_calls: list[str] = [] MixedApecAdapter.instances.append(self) def search(self, queries: list[str]) -> list[ApecSearchResult]: self.search_calls.append(list(queries)) return [good_result, bad_result] def fetch_listing_html(self, url: str) -> str: self.fetch_calls.append(url) if url == good_result.url: return html raise RuntimeError(f"boom: {url}") monkeypatch.setattr("job_research.cli.ApecAdapter", MixedApecAdapter) monkeypatch.setattr("job_research.cli._utc_now", _fixed_now) result = CliRunner().invoke(app, ["fetch-apec", "--data-root", str(data_root)]) assert result.exit_code == 0 assert "fetched=2" in result.stdout assert "normalized=1" in result.stdout assert "deduplicated=1" in result.stdout assert "failed=1" in result.stdout adapter = MixedApecAdapter.instances[0] assert adapter.search_calls == [["Role From YAML"]] assert adapter.fetch_calls == [good_result.url, bad_result.url] run_dir = data_root / "apec" / "runs" / "2026-06-01T10-00-00Z" snapshot_files = list((run_dir / "snapshots").glob("*.html")) assert [snapshot.name for snapshot in snapshot_files] == ["job-123.html"] assert snapshot_files[0].read_text(encoding="utf-8") == html listings_payload = yaml.safe_load((run_dir / "listings.yaml").read_text(encoding="utf-8")) assert listings_payload == [ { "source": "apec", "source_job_id": "job-123", "url": "https://example.test/job/123", "title": "Role From YAML", "company": "Example Corp", "location": "Paris - 75", "contract_type": "CDI", "description_text": "Build pipelines", "published_at": "2026-04-20", "fetched_at": "2026-06-01T10:00:00Z", "warnings": [], } ] run_meta_payload = yaml.safe_load((run_dir / "run-meta.yaml").read_text(encoding="utf-8")) assert run_meta_payload == { "run_id": "2026-06-01T10-00-00Z", "run_started_at": "2026-06-01T10:00:00Z", "derived_queries": ["Role From YAML"], "fetched_count": 2, "normalized_count": 1, "deduplicated_count": 1, "failed_count": 1, "listing_errors": [ { "url": bad_result.url, "stage": "fetch_html", "message": f"boom: {bad_result.url}", } ], } def test_fetch_apec_records_snapshot_write_failures_without_losing_normalized_listings( tmp_path, monkeypatch, ) -> None: data_root = tmp_path / "data" data_root.mkdir() (data_root / "candidate-profile.yaml").write_text( dedent( """ target_roles: - Role From YAML """ ).strip(), encoding="utf-8", ) html = _apec_detail_html() first_result = ApecSearchResult(url="https://example.test/job/123", source_job_id="job-123") second_result = ApecSearchResult(url="https://example.test/job/456", source_job_id="job-456") class SnapshotWriteFailingAdapter: instances: list["SnapshotWriteFailingAdapter"] = [] def __init__(self, max_listings: int = 50) -> None: self.max_listings = max_listings self.search_calls: list[list[str]] = [] self.fetch_calls: list[str] = [] SnapshotWriteFailingAdapter.instances.append(self) def search(self, queries: list[str]) -> list[ApecSearchResult]: self.search_calls.append(list(queries)) return [first_result, second_result] def fetch_listing_html(self, url: str) -> str: self.fetch_calls.append(url) return html original_write_text = Path.write_text def write_text_with_snapshot_failure( self: Path, data: str, encoding: str | None = None, errors: str | None = None, newline: str | None = None, ) -> int: if self.parent.name == "snapshots" and self.name == "job-123.html": raise OSError("disk full") return original_write_text(self, data, encoding=encoding, errors=errors, newline=newline) monkeypatch.setattr("job_research.cli.ApecAdapter", SnapshotWriteFailingAdapter) monkeypatch.setattr("job_research.cli._utc_now", _fixed_now) monkeypatch.setattr(Path, "write_text", write_text_with_snapshot_failure) result = CliRunner().invoke(app, ["fetch-apec", "--data-root", str(data_root)]) assert result.exit_code == 0 assert "fetched=2" in result.stdout assert "normalized=2" in result.stdout assert "deduplicated=2" in result.stdout assert "failed=1" in result.stdout run_dir = data_root / "apec" / "runs" / "2026-06-01T10-00-00Z" snapshot_files = sorted((run_dir / "snapshots").glob("*.html")) assert [snapshot.name for snapshot in snapshot_files] == ["job-456.html"] listings_payload = yaml.safe_load((run_dir / "listings.yaml").read_text(encoding="utf-8")) assert listings_payload == [ { "source": "apec", "source_job_id": "job-123", "url": "https://example.test/job/123", "title": "Role From YAML", "company": "Example Corp", "location": "Paris - 75", "contract_type": "CDI", "description_text": "Build pipelines", "published_at": "2026-04-20", "fetched_at": "2026-06-01T10:00:00Z", "warnings": [], }, { "source": "apec", "source_job_id": "job-456", "url": "https://example.test/job/456", "title": "Role From YAML", "company": "Example Corp", "location": "Paris - 75", "contract_type": "CDI", "description_text": "Build pipelines", "published_at": "2026-04-20", "fetched_at": "2026-06-01T10:00:00Z", "warnings": [], }, ] run_meta_payload = yaml.safe_load((run_dir / "run-meta.yaml").read_text(encoding="utf-8")) assert run_meta_payload == { "run_id": "2026-06-01T10-00-00Z", "run_started_at": "2026-06-01T10:00:00Z", "derived_queries": ["Role From YAML"], "fetched_count": 2, "normalized_count": 2, "deduplicated_count": 2, "failed_count": 1, "listing_errors": [ { "url": first_result.url, "stage": "snapshot_write", "message": "disk full", } ], } def test_fetch_apec_writes_run_meta_even_when_every_snapshot_write_fails( tmp_path, monkeypatch, ) -> None: data_root = tmp_path / "data" data_root.mkdir() (data_root / "candidate-profile.yaml").write_text( dedent( """ target_roles: - Role From YAML """ ).strip(), encoding="utf-8", ) html = _apec_detail_html() first_result = ApecSearchResult(url="https://example.test/job/123", source_job_id="job-123") second_result = ApecSearchResult(url="https://example.test/job/456", source_job_id="job-456") class SnapshotWriteFailingAdapter: instances: list["SnapshotWriteFailingAdapter"] = [] def __init__(self, max_listings: int = 50) -> None: self.max_listings = max_listings self.search_calls: list[list[str]] = [] self.fetch_calls: list[str] = [] SnapshotWriteFailingAdapter.instances.append(self) def search(self, queries: list[str]) -> list[ApecSearchResult]: self.search_calls.append(list(queries)) return [first_result, second_result] def fetch_listing_html(self, url: str) -> str: self.fetch_calls.append(url) return html original_write_text = Path.write_text def write_text_with_all_snapshot_failures( self: Path, data: str, encoding: str | None = None, errors: str | None = None, newline: str | None = None, ) -> int: if self.parent.name == "snapshots": raise OSError("disk full") return original_write_text(self, data, encoding=encoding, errors=errors, newline=newline) monkeypatch.setattr("job_research.cli.ApecAdapter", SnapshotWriteFailingAdapter) monkeypatch.setattr("job_research.cli._utc_now", _fixed_now) monkeypatch.setattr(Path, "write_text", write_text_with_all_snapshot_failures) result = CliRunner().invoke(app, ["fetch-apec", "--data-root", str(data_root)]) assert result.exit_code == 0 assert "fetched=2" in result.stdout assert "normalized=2" in result.stdout assert "deduplicated=2" in result.stdout assert "failed=2" in result.stdout run_dir = data_root / "apec" / "runs" / "2026-06-01T10-00-00Z" snapshot_files = sorted((run_dir / "snapshots").glob("*.html")) assert snapshot_files == [] listings_payload = yaml.safe_load((run_dir / "listings.yaml").read_text(encoding="utf-8")) assert listings_payload == [ { "source": "apec", "source_job_id": "job-123", "url": "https://example.test/job/123", "title": "Role From YAML", "company": "Example Corp", "location": "Paris - 75", "contract_type": "CDI", "description_text": "Build pipelines", "published_at": "2026-04-20", "fetched_at": "2026-06-01T10:00:00Z", "warnings": [], }, { "source": "apec", "source_job_id": "job-456", "url": "https://example.test/job/456", "title": "Role From YAML", "company": "Example Corp", "location": "Paris - 75", "contract_type": "CDI", "description_text": "Build pipelines", "published_at": "2026-04-20", "fetched_at": "2026-06-01T10:00:00Z", "warnings": [], }, ] run_meta_payload = yaml.safe_load((run_dir / "run-meta.yaml").read_text(encoding="utf-8")) assert run_meta_payload == { "run_id": "2026-06-01T10-00-00Z", "run_started_at": "2026-06-01T10:00:00Z", "derived_queries": ["Role From YAML"], "fetched_count": 2, "normalized_count": 2, "deduplicated_count": 2, "failed_count": 2, "listing_errors": [ { "url": first_result.url, "stage": "snapshot_write", "message": "disk full", }, { "url": second_result.url, "stage": "snapshot_write", "message": "disk full", }, ], } def test_fetch_apec_preserves_snapshot_and_run_meta_when_normalization_fails( tmp_path, monkeypatch, ) -> None: data_root = tmp_path / "data" data_root.mkdir() (data_root / "candidate-profile.yaml").write_text( dedent( """ target_roles: - Role From YAML """ ).strip(), encoding="utf-8", ) html = _apec_detail_html() result_item = ApecSearchResult(url="https://example.test/job/123", source_job_id="job-123") class NormalizeFailingAdapter: instances: list["NormalizeFailingAdapter"] = [] def __init__(self, max_listings: int = 50) -> None: self.max_listings = max_listings self.search_calls: list[list[str]] = [] self.fetch_calls: list[str] = [] NormalizeFailingAdapter.instances.append(self) def search(self, queries: list[str]) -> list[ApecSearchResult]: self.search_calls.append(list(queries)) return [result_item] def fetch_listing_html(self, url: str) -> str: self.fetch_calls.append(url) return html def fail_normalization(*args, **kwargs) -> None: raise RuntimeError("normalize boom") monkeypatch.setattr("job_research.cli.ApecAdapter", NormalizeFailingAdapter) monkeypatch.setattr("job_research.cli.normalize_apec_listing", fail_normalization) monkeypatch.setattr("job_research.cli._utc_now", _fixed_now) result = CliRunner().invoke(app, ["fetch-apec", "--data-root", str(data_root)]) assert result.exit_code == 1 assert "fetched=1" in result.stdout assert "normalized=0" in result.stdout assert "deduplicated=0" in result.stdout assert "failed=1" in result.stdout run_dir = data_root / "apec" / "runs" / "2026-06-01T10-00-00Z" snapshot_file = run_dir / "snapshots" / "job-123.html" assert snapshot_file.exists() assert snapshot_file.read_text(encoding="utf-8") == html listings_payload = yaml.safe_load((run_dir / "listings.yaml").read_text(encoding="utf-8")) assert listings_payload == [] run_meta_payload = yaml.safe_load((run_dir / "run-meta.yaml").read_text(encoding="utf-8")) assert run_meta_payload == { "run_id": "2026-06-01T10-00-00Z", "run_started_at": "2026-06-01T10:00:00Z", "derived_queries": ["Role From YAML"], "fetched_count": 1, "normalized_count": 0, "deduplicated_count": 0, "failed_count": 1, "listing_errors": [ { "url": result_item.url, "stage": "normalize", "message": "normalize boom", } ], } def test_fetch_apec_writes_artifacts_when_every_normalization_attempt_fails( tmp_path, monkeypatch, ) -> None: data_root = tmp_path / "data" data_root.mkdir() (data_root / "candidate-profile.yaml").write_text( dedent( """ target_roles: - Role From YAML """ ).strip(), encoding="utf-8", ) html = _apec_detail_html() first_result = ApecSearchResult(url="https://example.test/job/123", source_job_id="job-123") second_result = ApecSearchResult(url="https://example.test/job/456", source_job_id="job-456") class NormalizeFailingAdapter: instances: list["NormalizeFailingAdapter"] = [] def __init__(self, max_listings: int = 50) -> None: self.max_listings = max_listings self.search_calls: list[list[str]] = [] self.fetch_calls: list[str] = [] NormalizeFailingAdapter.instances.append(self) def search(self, queries: list[str]) -> list[ApecSearchResult]: self.search_calls.append(list(queries)) return [first_result, second_result] def fetch_listing_html(self, url: str) -> str: self.fetch_calls.append(url) return html def fail_normalization(*args, **kwargs) -> None: raise RuntimeError("normalize boom") monkeypatch.setattr("job_research.cli.ApecAdapter", NormalizeFailingAdapter) monkeypatch.setattr("job_research.cli.normalize_apec_listing", fail_normalization) monkeypatch.setattr("job_research.cli._utc_now", _fixed_now) result = CliRunner().invoke(app, ["fetch-apec", "--data-root", str(data_root)]) assert result.exit_code == 1 assert "fetched=2" in result.stdout assert "normalized=0" in result.stdout assert "deduplicated=0" in result.stdout assert "failed=2" in result.stdout run_dir = data_root / "apec" / "runs" / "2026-06-01T10-00-00Z" snapshot_files = sorted((run_dir / "snapshots").glob("*.html")) assert [snapshot.name for snapshot in snapshot_files] == ["job-123.html", "job-456.html"] assert [snapshot.read_text(encoding="utf-8") for snapshot in snapshot_files] == [html, html] listings_payload = yaml.safe_load((run_dir / "listings.yaml").read_text(encoding="utf-8")) assert listings_payload == [] run_meta_payload = yaml.safe_load((run_dir / "run-meta.yaml").read_text(encoding="utf-8")) assert run_meta_payload == { "run_id": "2026-06-01T10-00-00Z", "run_started_at": "2026-06-01T10:00:00Z", "derived_queries": ["Role From YAML"], "fetched_count": 2, "normalized_count": 0, "deduplicated_count": 0, "failed_count": 2, "listing_errors": [ { "url": first_result.url, "stage": "normalize", "message": "normalize boom", }, { "url": second_result.url, "stage": "normalize", "message": "normalize boom", }, ], } def test_fetch_apec_fails_when_no_queries_are_derived(tmp_path, monkeypatch) -> None: data_root = tmp_path / "data" data_root.mkdir() (data_root / "candidate-profile.yaml").write_text("target_roles: []\n", encoding="utf-8") class SpyApecAdapter: instances: list["SpyApecAdapter"] = [] def __init__(self, max_listings: int = 50) -> None: self.max_listings = max_listings SpyApecAdapter.instances.append(self) def search(self, queries: list[str]) -> list[ApecSearchResult]: raise AssertionError("search should not be called when there are no queries") def fetch_listing_html(self, url: str) -> str: raise AssertionError("fetch should not be called when there are no queries") monkeypatch.setattr("job_research.cli.ApecAdapter", SpyApecAdapter) monkeypatch.setattr("job_research.cli._utc_now", _fixed_now) result = CliRunner().invoke(app, ["fetch-apec", "--data-root", str(data_root)]) assert result.exit_code == 1 assert "No usable Apec queries derived from candidate profile" in result.stderr assert SpyApecAdapter.instances == [] assert not (data_root / "apec").exists() def test_fetch_apec_processes_only_the_first_fifty_search_results(tmp_path, monkeypatch) -> None: data_root = tmp_path / "data" data_root.mkdir() (data_root / "candidate-profile.yaml").write_text( dedent( """ target_roles: - Role From YAML """ ).strip(), encoding="utf-8", ) html = _apec_detail_html() results = [ ApecSearchResult(url=f"https://example.test/job/{index}", source_job_id=f"job-{index}") for index in range(51) ] class FakeApecAdapter: instances: list["FakeApecAdapter"] = [] def __init__(self, max_listings: int = 50) -> None: self.max_listings = max_listings self.search_calls: list[list[str]] = [] self.fetch_calls: list[str] = [] FakeApecAdapter.instances.append(self) def search(self, queries: list[str]) -> list[ApecSearchResult]: self.search_calls.append(list(queries)) return list(results) def fetch_listing_html(self, url: str) -> str: self.fetch_calls.append(url) return html monkeypatch.setattr("job_research.cli.ApecAdapter", FakeApecAdapter) monkeypatch.setattr("job_research.cli._utc_now", _fixed_now) result = CliRunner().invoke(app, ["fetch-apec", "--data-root", str(data_root)]) assert result.exit_code == 0 assert "fetched=50" in result.stdout assert "normalized=50" in result.stdout assert "deduplicated=50" in result.stdout assert "failed=0" in result.stdout adapter = FakeApecAdapter.instances[0] assert adapter.max_listings == 50 assert adapter.search_calls == [["Role From YAML"]] assert adapter.fetch_calls == [result.url for result in results[:50]] run_dir = data_root / "apec" / "runs" / "2026-06-01T10-00-00Z" snapshot_files = sorted((run_dir / "snapshots").glob("*.html")) assert len(snapshot_files) == 50 run_meta_payload = yaml.safe_load((run_dir / "run-meta.yaml").read_text(encoding="utf-8")) assert run_meta_payload["fetched_count"] == 50 assert run_meta_payload["normalized_count"] == 50 assert run_meta_payload["deduplicated_count"] == 50 assert run_meta_payload["failed_count"] == 0 def test_fetch_apec_fails_when_every_listing_attempt_fails(tmp_path, monkeypatch) -> None: data_root = tmp_path / "data" data_root.mkdir() (data_root / "candidate-profile.yaml").write_text( dedent( """ target_roles: - Role From YAML """ ).strip(), encoding="utf-8", ) results = [ ApecSearchResult(url=f"https://example.test/job/{index}", source_job_id=f"job-{index}") for index in range(2) ] class FailingApecAdapter: instances: list["FailingApecAdapter"] = [] def __init__(self, max_listings: int = 50) -> None: self.max_listings = max_listings self.search_calls: list[list[str]] = [] self.fetch_calls: list[str] = [] FailingApecAdapter.instances.append(self) def search(self, queries: list[str]) -> list[ApecSearchResult]: self.search_calls.append(list(queries)) return list(results) def fetch_listing_html(self, url: str) -> str: self.fetch_calls.append(url) raise RuntimeError(f"boom: {url}") monkeypatch.setattr("job_research.cli.ApecAdapter", FailingApecAdapter) monkeypatch.setattr("job_research.cli._utc_now", _fixed_now) result = CliRunner().invoke(app, ["fetch-apec", "--data-root", str(data_root)]) assert result.exit_code == 1 assert "No listings could be fetched or normalized from Apec" in result.stderr adapter = FailingApecAdapter.instances[0] assert adapter.search_calls == [["Role From YAML"]] assert adapter.fetch_calls == [result_item.url for result_item in results] run_dir = data_root / "apec" / "runs" / "2026-06-01T10-00-00Z" assert not (run_dir / "listings.yaml").exists() assert not (run_dir / "run-meta.yaml").exists()