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 from scripts.seed_notebook_pilot import seed_notebook_pilot 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")) 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}, ] CRITERIA = { "items": [ {"id": "ram", "criterion": "ram_gb", "operator": ">=", "value": 24, "mandatory": True}, {"id": "price", "criterion": "price_eur", "operator": "<=", "value": 2000, "mandatory": True}, {"id": "weight", "criterion": "weight_kg", "operator": "<=", "value": 1.6, "mandatory": True}, ] } @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) seed_notebook_pilot(session) session.commit() 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_profile(client: TestClient) -> None: for definition in DEFINITIONS: _ok(client.post("/api/v1/attribute-definitions", json=definition)) _ok(client.post("/api/v1/profiles", json=PROFILE)) def test_notebook_discovery_pilot(client: TestClient) -> None: _seed_profile(client) taxonomy = client.post("/api/v1/sources/geizhals/categories/nb/discover-taxonomy") assert taxonomy.status_code == 201, taxonomy.text category_id = taxonomy.json()["id"] filters = client.get("/api/v1/sources/geizhals/categories/nb/filters") assert filters.status_code == 200 assert {item["external_filter_key"] for item in filters.json()["filters"]} >= {"17023", "price"} ram = client.post( "/api/v1/criterion-mappings", json={ "criterion_key": "ram_gb", "source_id": "geizhals", "source_category_id": category_id, "external_filter_key": "17023", "source_value": "ab 24GB", "encoding": {"url_param": "xf"}, }, ) assert ram.status_code == 201, ram.text assert ram.json()["status"] == "VERIFIED" price = client.post( "/api/v1/criterion-mappings", json={ "criterion_key": "price_eur", "source_id": "geizhals", "source_category_id": category_id, "external_filter_key": "price", "source_value": None, "encoding": {"url_param": "maxp"}, "probe_value": 2000, "operator": "<=", }, ) assert price.status_code == 201, price.text assert price.json()["status"] == "VERIFIED" project = client.post( "/api/v1/buying-projects", json={ "name": "Notebook-Pilot", "profile_id": "mobile-development-notebook", "profile_version": 1, "shortlist_size": 5, "schedule_interval_minutes": 180, "criteria": CRITERIA, }, ) assert project.status_code == 201, project.text project_id = project.json()["id"] assert project.json()["workspace_id"] is None assert project.json()["criteria_version"] == 1 available = client.get(f"/api/v1/buying-projects/{project_id}/available-criteria") assert available.status_code == 200, available.text assert available.json()["product_category_id"] == "notebook" scoring_keys = {item.get("scoring_attribute_key") for item in available.json()["criteria"]} assert {"ram_gb", "price_eur"} <= scoring_keys ram = next(item for item in available.json()["criteria"] if item.get("scoring_attribute_key") == "ram_gb") assert ram["mapping_status"] == "VERIFIED" search = client.post( "/api/v1/discovery-searches", json={ "buying_project_id": project_id, "source_id": "geizhals", "source_category_id": category_id, "broad_result_threshold": 300, }, ) assert search.status_code == 201, search.text search_id = search.json()["id"] assert "xf=17023_ab+24GB" in search.json()["generated_url"] assert "maxp=2000" in search.json()["generated_url"] assert search.json()["schedule_interval_minutes"] == 180 due = client.get("/api/v1/discovery-searches?due=true") assert due.status_code == 200 assert any(item["id"] == search_id for item in due.json()["items"]) run = client.post(f"/api/v1/discovery-searches/{search_id}/run") assert run.status_code == 201, run.text assert run.json()["status"] == "SUCCESS" assert run.json()["page_count"] == 2 assert run.json()["result_count"] == 3 assert {item["diff_status"] for item in run.json()["items"]} == {"NEW"} shortlist = client.get(f"/api/v1/buying-projects/{project_id}/shortlist") assert shortlist.status_code == 200 assert len(shortlist.json()["items"]) == 1 assert shortlist.json()["items"][0]["product_eligibility"] == "ELIGIBLE" assert shortlist.json()["items"][0]["status"] == "SHORTLISTED" candidates = client.get(f"/api/v1/buying-projects/{project_id}/candidates") assert candidates.status_code == 200 eligibilities = {item["product_eligibility"] for item in candidates.json()["items"]} assert eligibilities == {"ELIGIBLE", "INELIGIBLE", "UNKNOWN"} assert all(item.get("deal_score") is None for item in candidates.json()["items"]) updated = client.post( f"/api/v1/buying-projects/{project_id}/criteria", json={"items": CRITERIA["items"][:2]}, ) assert updated.status_code == 201 assert updated.json()["criteria_version"] == 2 stale = client.get(f"/api/v1/discovery-searches/{search_id}") assert stale.json()["status"] == "OUT_OF_SYNC" blocked_run = client.post(f"/api/v1/discovery-searches/{search_id}/run") assert blocked_run.status_code == 409 refreshed = client.post(f"/api/v1/discovery-searches/{search_id}/generate-query") assert refreshed.status_code == 200 assert refreshed.json()["status"] == "ACTIVE" assert "weight_kg" not in refreshed.json().get("internal_only", []) assert "weight_kg" not in refreshed.json()["generated_url"] client.patch(f"/api/v1/discovery-searches/{search_id}", json={"status": "DISABLED"}) broad = client.post( "/api/v1/discovery-searches", json={ "buying_project_id": project_id, "source_id": "geizhals", "source_category_id": category_id, "broad_result_threshold": 2, }, ) assert broad.status_code == 201, broad.text broad_run = client.post(f"/api/v1/discovery-searches/{broad.json()['id']}/run") assert broad_run.status_code == 201, broad_run.text assert broad_run.json()["status"] == "TOO_BROAD" assert broad_run.json()["result_count"] == 3 stored = client.get(f"/api/v1/discovery-searches/{broad.json()['id']}") assert stored.json()["status"] == "TOO_BROAD"