booleanbeyond/jobfetch

1"""HTTP contract tests."""2 3from __future__ import annotations4 5import pytest6from fastapi.testclient import TestClient7 8from app.cache import RateLimiter9from app.main import app10 11client = TestClient(app)12 13 14def test_index_serves_the_ui():15    r = client.get("/")16    assert r.status_code == 20017    assert "JobFetch" in r.text18    assert 'id="url"' in r.text19 20 21def test_healthz_reports_subsystems():22    r = client.get("/healthz")23    assert r.status_code == 20024    body = r.json()25    assert body["ok"] is True26    assert "browser" in body and "llm" in body and "cache" in body27 28 29def test_config_endpoint_never_leaks_the_api_key():30    body = client.get("/api/config").json()31    assert "anthropic_api_key" not in body32    assert body["anthropic_api_key_present"] in (True, False)33    assert not any("sk-" in str(v) for v in body.values())34 35 36@pytest.mark.parametrize(37    "url",38    # The fixture server lives on 127.0.0.1, so this suite runs with the39    # ALLOW_PRIVATE_HOSTS escape hatch on. Strict-mode rejection of private40    # ranges is covered in test_security.py.41    ["file:///etc/passwd", "http://169.254.169.254/", "http://metadata.google.internal/",42     "notaurl::", "", "ftp://example.com/x"],43)44def test_unsafe_urls_are_rejected_with_400(url):45    r = client.post("/api/scrape", json={"url": url})46    assert r.status_code in (400, 422)47    if r.status_code == 400:48        body = r.json()49        assert body["ok"] is False50        assert body["error"] == "invalid_url"51        assert body["trace_id"]52 53 54def test_successful_scrape_response_shape(server):55    r = client.post(56        "/api/scrape",57        json={"url": f"{server}/footer_trap.html", "use_browser": False, "use_llm": False},58    )59    assert r.status_code == 20060    body = r.json()61 62    assert body["ok"] is True63    assert body["job_count"] == 664    assert body["company"] == "Acme Robotics"65    assert 0 < body["confidence"] <= 166    assert body["diagnostics"]["strategy"] == "dom_heuristic"67    assert body["diagnostics"]["trace_id"]68    assert r.headers["X-Trace-Id"]69    assert r.headers["X-Content-Type-Options"] == "nosniff"70 71    # Every job carries the full key set, so consumers never need .get() guards.72    expected = {73        "id", "requisition_id", "title", "department", "team", "location", "locations",74        "workplace_type", "employment_type", "seniority", "experience", "salary", "posted_at",75        "updated_at", "apply_url", "source", "confidence",76    }77    for job in body["jobs"]:78        assert set(job) == expected79        assert isinstance(job["title"], str) and job["title"].strip()80        assert isinstance(job["locations"], list)81 82 83def test_get_endpoint_matches_post(server):84    a = client.get("/api/scrape", params={"url": f"{server}/table.html",85                                          "use_browser": False, "use_llm": False}).json()86    b = client.post("/api/scrape", json={"url": f"{server}/table.html",87                                         "use_browser": False, "use_llm": False}).json()88    assert a["job_count"] == b["job_count"] == 589 90 91@pytest.mark.asyncio92async def test_rate_limiter_token_bucket():93    limiter = RateLimiter(per_minute=60, burst=2)94    assert await limiter.allow("ip") is True95    assert await limiter.allow("ip") is True96    assert await limiter.allow("ip") is False97    assert await limiter.allow("other-ip") is True98 99 100@pytest.mark.asyncio101async def test_rate_limiter_key_space_is_bounded():102    limiter = RateLimiter(per_minute=60, burst=2, max_keys=10)103    for i in range(100):104        await limiter.allow(f"ip-{i}")105    assert len(limiter._buckets) <= 10106