134 lines
4.3 KiB
Python
134 lines
4.3 KiB
Python
"""Root conftest: make the repo importable and provide shared test fixtures.
|
|
|
|
Adds the repository root to sys.path so both the ``scraper`` package and the
|
|
``tests`` package import cleanly regardless of the invocation directory, and
|
|
exposes helpers the test-suite shares -- notably a fake HTTP transport that
|
|
drives the REAL retry/backoff loop with no network, and a per-test log capture.
|
|
Nothing here touches the live network unless a test opts in via the
|
|
``CALI_SCRAPER_SMOKE`` env flag.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
# --- make the repo importable from any cwd --------------------------------
|
|
ROOT = os.path.dirname(os.path.abspath(__file__))
|
|
if ROOT not in sys.path:
|
|
sys.path.insert(0, ROOT)
|
|
|
|
# An absolute path to the real config.yaml, so tests that chdir() (e.g. to a
|
|
# tmp_path) still find it.
|
|
CONFIG_PATH = os.path.join(ROOT, "config.yaml")
|
|
|
|
from scraper.cache import RawCache # noqa: E402,F401
|
|
from scraper.http_client import ( # noqa: E402,F401
|
|
AntibotError,
|
|
FetchResult,
|
|
NotFoundError,
|
|
TransientError,
|
|
)
|
|
from scraper.ingest import Ingestor # noqa: E402,F401
|
|
from scraper.request_log import RequestLog # noqa: E402,F401
|
|
|
|
|
|
# ---------------------------------------------------------------- shared fakes
|
|
#
|
|
# A fake transport that mimics httpx.Client.get() so the *real* HttpClient
|
|
# retry loop runs with no network. Routed per-URL so a single test can exercise
|
|
# 200 / 404 / 403 / 5xx behaviour.
|
|
|
|
class FakeResponse:
|
|
def __init__(self, status, content, content_type="text/html"):
|
|
self.status_code = status
|
|
self.content = content
|
|
self.headers = {"content-type": content_type}
|
|
|
|
|
|
class FakeTransport:
|
|
def __init__(self, route):
|
|
self.route = route # callable(url) -> (status, content_bytes)
|
|
self.calls = []
|
|
|
|
def get(self, url, follow_redirects=False):
|
|
self.calls.append(url)
|
|
status, content = self.route(url)
|
|
return FakeResponse(status, content)
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
|
|
def make_ingestor(config_path, raw_dir, log_path, route=None):
|
|
"""Build an Ingestor wired to a temp cache+log and (optionally) a fake
|
|
transport, with the backoff sleep neutralised so the test is instant.
|
|
|
|
Returns (ingestor, transport-or-None). ``ing._sleeps`` records the backoff
|
|
delays that WOULD have been slept, so tests can assert exponential timing.
|
|
"""
|
|
from pathlib import Path
|
|
|
|
ing = Ingestor(config_path=config_path, force=True)
|
|
ing.cache.raw_dir = Path(raw_dir)
|
|
ing.log.path = Path(log_path)
|
|
sleeps = []
|
|
ing.http.sleep = lambda d: sleeps.append(d)
|
|
ing._sleeps = sleeps
|
|
transport = None
|
|
if route is not None:
|
|
transport = FakeTransport(route)
|
|
ing.http._client = transport
|
|
return ing, transport
|
|
|
|
|
|
class LogCapture(logging.Handler):
|
|
def __init__(self, logger):
|
|
super().__init__()
|
|
self.records = []
|
|
self._logger = logger
|
|
self._prev_level = logger.level
|
|
logger.addHandler(self)
|
|
logger.setLevel(logging.DEBUG)
|
|
|
|
def emit(self, record):
|
|
# A no-op emit that records every log message for later inspection.
|
|
self.records.append(record)
|
|
|
|
@property
|
|
def warnings(self):
|
|
return [r.getMessage() for r in self.records
|
|
if r.levelno >= logging.WARNING]
|
|
|
|
@property
|
|
def messages(self):
|
|
return [r.getMessage() for r in self.records]
|
|
|
|
def remove(self):
|
|
self._logger.removeHandler(self)
|
|
self._logger.setLevel(self._prev_level)
|
|
|
|
|
|
# ---------------------------------------------------------------- pytest fixtures
|
|
|
|
@pytest.fixture
|
|
def logcap():
|
|
def _cap(name):
|
|
h = LogCapture(logging.getLogger(name))
|
|
return h
|
|
# No per-call teardown is required: LogCapture installs and uninstalls a
|
|
# handler on a module logger. Loggers are cached process-wide, so the
|
|
# handler lingers but is harmless (it just keeps appended records it was
|
|
# never asked about). For tests that must observe warnings, attach the
|
|
# capture at the start of the test and read `.warnings` before the next
|
|
# logcap() call.
|
|
return _cap
|
|
|
|
|
|
def read_jsonl(path):
|
|
with open(path, encoding="utf-8") as fh:
|
|
return [json.loads(line) for line in fh if line.strip()]
|