spb/company-atlas

1from __future__ import annotations23import asyncio45import pytest6from conftest import FakeFetcher, fixture_bytes, fixture_text78from companyatlas.services import discovery as D9from companyatlas.taxonomy import OnboardingStatus, Surface1011BASE = "https://www.acme-cloud.example"12COMPANY = {"id": "co_test", "slug": "ztest-acme", "display_name": "Acme Cloud", "canonical_domain": "acme-cloud.example", "website": BASE + "/", "tier": 2, "importance": 0.5}131415def _fake_site(ff: FakeFetcher) -> None:16    ff.add(BASE + "/", fixture_text("generic_html", "homepage.html"))17    ff.add(BASE + "/robots.txt", "User-agent: *\nAllow: /\nSitemap: https://www.acme-cloud.example/sitemap.xml\n", content_type="text/plain")18    ff.add(BASE + "/sitemap.xml", fixture_bytes("sitemap", "sitemap_index.xml"), content_type="application/xml")19    ff.add(BASE + "/sitemap-pages.xml", fixture_bytes("sitemap", "sitemap_pages.xml"), content_type="application/xml")20    ff.add(BASE + "/sitemap-blog.xml", "<urlset></urlset>", content_type="application/xml")21    ff.add(BASE + "/careers", fixture_text("generic_html", "careers.html"))22    ff.add(BASE + "/pricing", fixture_text("generic_html", "pricing.html"))23    ff.add("https://boards-api.greenhouse.io/v1/boards/acmecloud/jobs?content=false", fixture_bytes("greenhouse", "stripe_jobs.json"), content_type="application/json")242526@pytest.fixture(autouse=True)27def _no_dns(monkeypatch: pytest.MonkeyPatch) -> None:28    async def never(host: str) -> bool:29        return False30    monkeypatch.setattr(D, "_resolves", never)313233def test_discover_company_dry_run(fake_fetcher: FakeFetcher) -> None:34    _fake_site(fake_fetcher)35    res = asyncio.run(D.discover_company(COMPANY, fetcher=fake_fetcher, dry_run=True))  # type: ignore[arg-type]36    assert res.status == OnboardingStatus.ACTIVE, res.error37    surfaces = {s["surface"]: s for s in res.sensors}38    assert {"homepage", "pricing", "careers", "jobs_board", "newsroom", "leadership", "locations", "sitemap", "legal_terms", "legal_privacy", "feed"} <= set(surfaces)39    gh = surfaces["jobs_board"]40    assert gh["connector_id"] == "greenhouse-v1" and gh["config"]["token"] == "acmecloud" and gh["config"].get("verified_job_count") == 2041    assert surfaces["careers"]["url"] == BASE + "/careers" and surfaces["careers"]["connector_id"] == "generic-html-v1"42    assert surfaces["sitemap"]["connector_id"] == "sitemap-v1"43    assert surfaces["feed"]["connector_id"] == "feed-v1"44    # tier 2 → ×0.75 on the surface base interval, clamped45    assert surfaces["pricing"]["base_interval_s"] == int(12 * 3600 * 0.75)46    assert surfaces["homepage"]["tier"] in ("C", "D") and 0 < surfaces["pricing"]["quality_score"] <= 10047    assert all(s["status"] == "pending" for s in res.sensors)48    assert len(res.sensors) <= 40 and len({s["canonical_url"] for s in res.sensors}) == len(res.sensors)49    # bounded politeness: no more than ~25 requests for a whole company50    assert res.requests <= 2551    assert all(c.startswith(("https://www.acme-cloud.example", "https://boards-api.greenhouse.io")) for c in fake_fetcher.calls)525354def test_discover_no_website(fake_fetcher: FakeFetcher) -> None:55    from companyatlas.fetch import FetchError56    from companyatlas.taxonomy import FailureClass5758    for u in (BASE + "/", "https://www.acme-cloud.example/", "https://acme-cloud.example/", "http://www.acme-cloud.example/", "http://acme-cloud.example/"):59        fake_fetcher.add_error(u, FetchError("dns", url=u, failure=FailureClass.DNS))60    res = asyncio.run(D.discover_company(COMPANY, fetcher=fake_fetcher, dry_run=True))  # type: ignore[arg-type]61    assert res.status == OnboardingStatus.NO_WEBSITE and not res.sensors626364def test_redirect_to_other_domain_is_recorded(fake_fetcher: FakeFetcher) -> None:65    fake_fetcher.add(BASE + "/", fixture_text("generic_html", "homepage.html"), final_url="https://www.acme-group.example/")66    fake_fetcher.add("https://www.acme-group.example/robots.txt", "", content_type="text/plain")67    res = asyncio.run(D.discover_company(COMPANY, fetcher=fake_fetcher, dry_run=True))  # type: ignore[arg-type]68    assert res.redirect_domain == "acme-group.example" and res.canonical_domain == "acme-group.example"69    assert any("redirects" in n for n in res.notes)707172def test_select_sensors_caps_and_prefers_best_candidate() -> None:73    from datetime import UTC, datetime7475    cands = [D.Candidate(url=BASE + "/pricing", surface=str(Surface.PRICING), confidence=0.7, method="sitemap"),76             D.Candidate(url=BASE + "/plans", surface=str(Surface.PRICING), confidence=0.95, method="nav", verified=True),77             D.Candidate(url=BASE + "/x", surface=str(Surface.OTHER), confidence=0.3, method="nav")]78    rows = D.select_sensors(cands, company=COMPANY, canonical_domain="acme-cloud.example", now=datetime.now(UTC))79    assert len(rows) == 1 and rows[0]["url"] == BASE + "/plans" and rows[0]["discovery_method"] == "nav"80