import json from pathlib import Path import pytest from fastapi.testclient import TestClient from sqlalchemy import text from app.main import app from app.persistence.db import SessionLocal, engine from scripts.bootstrap import run_bootstrap pytestmark = [pytest.mark.e2e, pytest.mark.integration] ROOT = Path(__file__).resolve().parents[2] PROFILE = json.loads((ROOT / "profiles" / "mobile-development-notebook.json").read_text(encoding="utf-8")) FIXTURE = json.loads((ROOT / "fixtures" / "manual-demo.json").read_text(encoding="utf-8")) AS_OF = "2026-10-03T12:00:00Z" DEFINITIONS = [ {"key": "ram_gb", "value_type": "number", "canonical_unit": "GB", "minimum": 0}, {"key": "storage_gb", "value_type": "number", "canonical_unit": "GB", "minimum": 0}, {"key": "weight_kg", "value_type": "number", "canonical_unit": "kg", "minimum": 0}, {"key": "battery_score", "value_type": "number", "minimum": 0, "maximum": 100}, {"key": "display_score", "value_type": "number", "minimum": 0, "maximum": 100}, {"key": "cpu_score", "value_type": "number", "minimum": 0, "maximum": 100}, ] @pytest.fixture(scope="module", autouse=True) def require_postgres() -> None: try: with engine.connect() as connection: connection.execute(text("SELECT 1")) except Exception as exc: pytest.skip(f"PostgreSQL not available: {exc}") session = SessionLocal() try: run_bootstrap(session) finally: session.close() @pytest.fixture def client() -> TestClient: return TestClient(app) def _ok(response, allowed=(200, 201, 409)) -> None: assert response.status_code in allowed, response.text def _seed(client: TestClient) -> None: for definition in DEFINITIONS: _ok(client.post("/api/v1/attribute-definitions", json=definition)) _ok(client.post("/api/v1/profiles", json=PROFILE)) product = FIXTURE["products"][0] _ok( client.post( "/api/v1/products", json={ "id": product["id"], "category": product["category"], "brand": product["brand"], "name": product["name"], "model": product.get("model"), }, ) ) for variant in product["variants"]: _ok( client.post( f"/api/v1/products/{product['id']}/variants", json={ "id": variant["id"], "name": variant["name"], "manufacturer_sku": variant.get("manufacturer_sku"), "identifiers": variant.get("identifiers") or {}, }, ) ) for attribute in variant["attributes"]: _ok(client.post(f"/api/v1/variants/{variant['id']}/attributes", json={**attribute, "source_id": "manual"})) for offer in FIXTURE["offers"]: _ok( client.post( f"/api/v1/variants/{offer['variant_id']}/offers", json={ "id": offer["id"], "source_id": "manual", "external_id": offer["external_id"], "seller": offer["seller"], "url": offer.get("url"), "market": offer["market"], "currency": offer["currency"], "condition": offer["condition"], }, ) ) for price in FIXTURE["prices"]: offer = next(item for item in FIXTURE["offers"] if item["external_id"] == price["external_offer_id"]) _ok( client.post( f"/api/v1/offers/{offer['id']}/prices", json={ "id": price["id"], "source_id": "manual", "external_observation_id": price["external_observation_id"], "observed_at": price["observed_at"], "item_price": price["item_price"], "shipping_price": price["shipping_price"], "currency": price["currency"], "availability": price["availability"], "raw_payload": price["raw_payload"], "confidence": price["confidence"], }, ), allowed=(200, 201), ) def test_http_cases_and_replay(client: TestClient) -> None: _seed(client) cases = { "A": ("22222222-2222-4222-8222-222222222222", "33333333-3333-4333-8333-333333333333"), "B": ("22222222-2222-4222-8222-22222222222b", "33333333-3333-4333-8333-33333333333b"), "C": ("22222222-2222-4222-8222-22222222222c", "33333333-3333-4333-8333-33333333333c"), "D": ("22222222-2222-4222-8222-222222222222", "33333333-3333-4333-8333-33333333333d"), } created = {} for label, (variant_id, offer_id) in cases.items(): response = client.post( f"/api/v1/evaluate/{variant_id}", json={ "profile_id": "mobile-development-notebook", "profile_version": 1, "offer_id": offer_id, "as_of": AS_OF, }, ) assert response.status_code == 201, response.text created[label] = response.json() assert created["A"]["product_score"] == 90.21 assert created["A"]["price_score"] == 98.24 assert created["A"]["deal_score"] == 93.42 assert created["A"]["classification"] == "EXCEPTIONAL" assert created["A"]["eligible"] is True assert created["A"]["price_reference_scope"] == "offer" assert created["B"]["product_eligibility"] == "INELIGIBLE" assert created["B"]["offer_eligibility"] == "ELIGIBLE" assert created["B"]["deal_score"] is None assert created["B"]["product_score"] is not None assert created["C"]["product_eligibility"] == "UNKNOWN" assert created["C"]["classification"] == "INSUFFICIENT_DATA" assert created["D"]["product_eligibility"] == "ELIGIBLE" assert created["D"]["offer_eligibility"] == "INELIGIBLE" assert created["D"]["product_score"] == 90.21 assert created["D"]["price_score"] is not None assert created["D"]["deal_score"] is None stored = client.get(f"/api/v1/evaluations/{created['A']['id']}") assert stored.status_code == 200 assert stored.json()["deal_score"] == created["A"]["deal_score"] assert stored.json()["input_snapshot"] == created["A"]["input_snapshot"] extra = client.post( "/api/v1/offers/33333333-3333-4333-8333-333333333333/prices", json={ "source_id": "manual", "external_observation_id": "a-later", "observed_at": "2026-10-03T11:00:00Z", "item_price": "999.00", "shipping_price": "0.00", "currency": "EUR", "availability": "in_stock", "raw_payload": {"total": "999.00"}, "confidence": 1.0, }, ) assert extra.status_code == 201 replay = client.get(f"/api/v1/evaluations/{created['A']['id']}") assert replay.json()["deal_score"] == created["A"]["deal_score"] assert replay.json()["price_statistics"]["current_total"] == created["A"]["price_statistics"]["current_total"] def test_error_contracts(client: TestClient) -> None: missing = client.get("/api/v1/products/00000000-0000-4000-8000-000000000000") assert missing.status_code == 404 assert "error" in missing.json() assert missing.json()["error"]["code"] == "NOT_FOUND" bad = client.post("/api/v1/products", json={"category": "notebook"}) assert bad.status_code == 422 assert "error" in bad.json() client.post("/api/v1/attribute-definitions", json={"key": "ports", "value_type": "string_list"}) again = client.post("/api/v1/attribute-definitions", json={"key": "ports", "value_type": "string_list"}) assert again.status_code == 409 no_version = client.get("/api/v1/profiles/mobile-development-notebook") assert no_version.status_code == 422