spb/company-atlas

1"""Shared test helpers: fixture paths, a network-free FakeFetcher, and a DB availability check for integration tests."""2from __future__ import annotations34import asyncio5from dataclasses import dataclass, field6from datetime import UTC, datetime7from pathlib import Path8from typing import Any, Self910import pytest1112from companyatlas.fetch import FetchError, FetchResult, NotModified13from companyatlas.taxonomy import FailureClass1415ROOT = Path(__file__).resolve().parents[1]16FIXTURES = ROOT / "fixtures" / "connectors"171819def fixture_path(*parts: str) -> str:20    return str(FIXTURES.joinpath(*parts))212223def fixture_bytes(*parts: str) -> bytes:24    return FIXTURES.joinpath(*parts).read_bytes()252627def fixture_text(*parts: str) -> str:28    return FIXTURES.joinpath(*parts).read_text(encoding="utf-8")293031def make_result(url: str, content: bytes | str, *, content_type: str = "text/html; charset=utf-8", status: int = 200, final_url: str | None = None,32                headers: dict[str, str] | None = None) -> FetchResult:33    body = content.encode("utf-8") if isinstance(content, str) else content34    return FetchResult(url=url, final_url=final_url or url, status=status, headers={k.lower(): v for k, v in (headers or {}).items()}, content=body,35                       content_type=content_type, fetched_at=datetime.now(UTC), duration_ms=1, transport="fake")363738@dataclass39class FakeFetcher:40    """Network-free stand-in for `fetch.Fetcher`: maps URL → (status, content_type, body) or an exception. Records every request."""41    routes: dict[str, Any] = field(default_factory=dict)42    calls: list[str] = field(default_factory=list)43    default_status: int = 4044445    def add(self, url: str, body: bytes | str, *, content_type: str = "text/html; charset=utf-8", status: int = 200, final_url: str | None = None) -> None:46        self.routes[url] = (status, content_type, body, final_url)4748    def add_error(self, url: str, exc: Exception) -> None:49        self.routes[url] = exc5051    async def open(self) -> None:52        return None5354    async def close(self) -> None:55        return None5657    async def __aenter__(self) -> Self:58        return self5960    async def __aexit__(self, *exc: object) -> None:61        return None6263    async def get(self, url: str, **kw: Any) -> FetchResult:64        self.calls.append(url)65        hit = self.routes.get(url) or self.routes.get(url.rstrip("/")) or self.routes.get(url + "/")66        if hit is None:67            raise FetchError(f"http 404 for {url}", status=404, url=url, failure=FailureClass.PAGE_REMOVED)68        if isinstance(hit, Exception):69            if isinstance(hit, NotModified):70                raise hit71            raise hit72        status, ctype, body, final = hit73        if status >= 400:74            raise FetchError(f"http {status} for {url}", status=status, url=url, failure=FailureClass.HTTP_4XX if status < 500 else FailureClass.HTTP_5XX)75        return make_result(url, body, content_type=ctype, status=status, final_url=final)7677    async def request(self, method: str, url: str, **kw: Any) -> FetchResult:78        return await self.get(url, **kw)7980    async def post_json(self, url: str, payload: Any, **kw: Any) -> FetchResult:81        self.calls.append(f"POST {url}")82        return await self.get(url, **kw)838485@pytest.fixture86def fake_fetcher() -> FakeFetcher:87    return FakeFetcher()888990def _db_available() -> bool:91    async def probe() -> bool:92        from companyatlas.db import dispose, fetch_val, transaction9394        try:95            async with transaction() as conn:96                await fetch_val(conn, "select 1")97            return True98        except Exception:  # noqa: BLE00199            return False100        finally:101            await dispose()102103    try:104        return asyncio.run(probe())105    except Exception:  # noqa: BLE001106        return False107108109_DB_OK: bool | None = None110111112@pytest.fixture113def db():  # type: ignore[no-untyped-def]114    """Skip integration tests when the local Postgres is unavailable."""115    global _DB_OK116    if _DB_OK is None:117        _DB_OK = _db_available()118    if not _DB_OK:119        pytest.skip("local Postgres not available")120    return True121