269 lines
9.1 KiB
Python
269 lines
9.1 KiB
Python
from datetime import datetime, timezone
|
|
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 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 = dedent(
|
|
"""
|
|
<html>
|
|
<body>
|
|
<h1>Role From YAML</h1>
|
|
<div class="company">Example Corp</div>
|
|
<div class="location">Paris</div>
|
|
<div class="contract">CDI</div>
|
|
<div class="description">Build pipelines</div>
|
|
</body>
|
|
</html>
|
|
"""
|
|
).strip()
|
|
|
|
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].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",
|
|
"contract_type": "CDI",
|
|
"description_text": "Build pipelines",
|
|
"published_at": None,
|
|
"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 == {
|
|
"derived_queries": ["Role From YAML"],
|
|
"fetched_count": 1,
|
|
"normalized_count": 1,
|
|
"deduplicated_count": 1,
|
|
"failed_count": 0,
|
|
"listing_errors": [],
|
|
}
|
|
|
|
|
|
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 = dedent(
|
|
"""
|
|
<html>
|
|
<body>
|
|
<h1>Role From YAML</h1>
|
|
<div class="company">Example Corp</div>
|
|
<div class="location">Paris</div>
|
|
<div class="contract">CDI</div>
|
|
<div class="description">Build pipelines</div>
|
|
</body>
|
|
</html>
|
|
"""
|
|
).strip()
|
|
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()
|