138 lines
4.5 KiB
Python
138 lines
4.5 KiB
Python
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
EXTRACTION_RESULTS = []
|
|
|
|
|
|
def pytest_addoption(parser):
|
|
parser.addoption("--results-file", help="Write the human-readable terminal report to this file.")
|
|
parser.addoption("--extraction-results", help="Write cached source extraction counts to this file.")
|
|
|
|
|
|
def pytest_configure(config):
|
|
results_file = config.getoption("--results-file")
|
|
if results_file:
|
|
config.pluginmanager.register(_ResultsFilePlugin(Path(results_file)), "results-file")
|
|
|
|
|
|
def pytest_sessionfinish(session, exitstatus):
|
|
report_option = session.config.getoption("--extraction-results")
|
|
if not report_option or not EXTRACTION_RESULTS:
|
|
return
|
|
report_path = Path(report_option)
|
|
rows = sorted(EXTRACTION_RESULTS, key=lambda row: row["name"].casefold())
|
|
lines = ["What2Do cached source extraction results", "=" * 39, "", f"{'Source':40} {'Mode':12} {'Events':>6} Status", "-" * 75]
|
|
for row in rows:
|
|
lines.append(f"{row['name'][:40]:40} {row['mode'][:12]:12} {row['events']:6} {row['status'].upper()}")
|
|
if row["message"]:
|
|
lines.append(f" {row['message']}")
|
|
lines.extend(("", f"Sources tested: {len(rows)}", f"Events found: {sum(row['events'] for row in rows)}", f"Successful extractions: {sum(row['status'] == 'passed' for row in rows)}", f"Failed extractions: {sum(row['status'] == 'failed' for row in rows)}"))
|
|
report_path.parent.mkdir(parents=True, exist_ok=True)
|
|
report_path.write_text("\n".join(lines) + "\n")
|
|
|
|
|
|
class _ResultsFilePlugin:
|
|
def __init__(self, path):
|
|
self.path = path
|
|
self.reports = []
|
|
|
|
def pytest_runtest_logreport(self, report):
|
|
if report.when == "call" or report.failed:
|
|
self.reports.append(report)
|
|
|
|
def pytest_sessionfinish(self, session, exitstatus):
|
|
lines = ["What2Do automated test results", "=" * 32, ""]
|
|
for report in self.reports:
|
|
outcome = report.outcome.upper()
|
|
lines.append(f"{outcome:7} {report.nodeid} ({report.duration:.3f}s)")
|
|
if report.failed:
|
|
lines.extend(("", str(report.longrepr), ""))
|
|
counts = {outcome: sum(report.outcome == outcome for report in self.reports) for outcome in ("passed", "failed", "skipped")}
|
|
lines.extend(("", "Summary", "-------", f"Passed: {counts['passed']}", f"Failed: {counts['failed']}", f"Skipped: {counts['skipped']}", f"Exit status: {exitstatus}"))
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
self.path.write_text("\n".join(lines) + "\n")
|
|
|
|
|
|
@pytest.fixture
|
|
def record_extraction_result():
|
|
return EXTRACTION_RESULTS.append
|
|
|
|
|
|
@pytest.fixture
|
|
def source():
|
|
return {
|
|
"url": "https://example.org/events",
|
|
"timezone": "Australia/Sydney",
|
|
"venue_name": "Example Venue",
|
|
"venue_address": "1 Example St, Wollongong NSW 2500",
|
|
"latitude": -34.42,
|
|
"longitude": 150.89,
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def fixture_html():
|
|
def load(name):
|
|
return (Path(__file__).resolve().parent / "fixtures" / name).read_text()
|
|
return load
|
|
|
|
|
|
@pytest.fixture
|
|
def html_cards(fixture_html):
|
|
return fixture_html("event_cards.html")
|
|
|
|
|
|
class FakeResponse:
|
|
def __init__(self, *, text="", payload=None, status_code=200, url="https://example.org/events"):
|
|
self.text = text
|
|
self._payload = payload
|
|
self.status_code = status_code
|
|
self.url = url
|
|
self.content = text.encode() if payload is None else json.dumps(payload).encode()
|
|
|
|
def json(self):
|
|
return self._payload
|
|
|
|
def raise_for_status(self):
|
|
if self.status_code >= 400:
|
|
import httpx
|
|
raise httpx.HTTPStatusError("failed", request=httpx.Request("GET", self.url), response=httpx.Response(self.status_code))
|
|
|
|
|
|
class RecordingClient:
|
|
def __init__(self, responses):
|
|
self.responses = list(responses)
|
|
self.calls = []
|
|
|
|
def request(self, method, url, **kwargs):
|
|
self.calls.append((method, url, kwargs))
|
|
return self.responses.pop(0)
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_response():
|
|
return FakeResponse
|
|
|
|
|
|
@pytest.fixture
|
|
def recording_client():
|
|
return RecordingClient
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_db(tmp_path, monkeypatch):
|
|
import app.db as db_module
|
|
path = tmp_path / "nested" / "what2do.db"
|
|
monkeypatch.setattr(db_module, "DB_PATH", str(path))
|
|
db_module.init_db()
|
|
return path
|
|
|
|
|
|
@pytest.fixture
|
|
def source_manifest():
|
|
path = Path(__file__).parent / "fixtures" / "illawarra_sources.json"
|
|
return json.loads(path.read_text())
|