BuyingAgent/backend/tests/integration/test_discovery_lifecycle.py

from copy import deepcopy
from decimal import Decimal

import pytest
from sqlalchemy import text

from app.discovery.enums import DiffStatus, LifecycleStatus, MatchStatus, RunStatus
from app.errors import AppError
from app.persistence.db import SessionLocal, engine
from app.persistence.discovery_models import (
    ProductObservationRecord,
    ProjectCandidateRecord,
    SourceProductReferenceRecord,
)
from app.services.discovery_service import (
    CriterionMappingService,
    DiscoverySearchService,
    SearchExecutionService,
    TaxonomyService,
)
from app.services.ingestion_service import IngestionService
from app.services.project_service import ProjectService
from app.sources.geizhals.adapter import GeizhalsAdapter
from scripts.bootstrap import run_bootstrap
from scripts.seed_notebook_pilot import seed_notebook_pilot
from tests.e2e.test_http_walkthrough import DEFINITIONS, PROFILE

pytestmark = pytest.mark.integration

ITEM_A = {
    "source_external_id": "gh-nb-a",
    "source_url": "https://geizhals.de/example-mobile-13-a.html",
    "source_title": "Example Mobile 13 32GB",
    "manufacturer": "Example",
    "model": "Mobile 13",
    "manufacturer_sku": "EM13-32-1T",
    "gtin_ean": "4000000000001",
    "best_price": "1298.00",
    "currency": "EUR",
    "availability": "in_stock",
    "seller_count": 8,
    "raw_attributes": {"RAM": "32GB"},
    "normalized_attributes": {"ram_gb": 32, "storage_gb": 1024, "weight_kg": 1.2},
}


def _item(suffix: str, **overrides) -> dict:
    item = deepcopy(ITEM_A)
    item["source_external_id"] = f"gh-nb-{suffix}"
    item["source_url"] = f"https://geizhals.de/example-mobile-13-{suffix}.html"
    item["manufacturer_sku"] = f"EM13-{suffix}"
    item["gtin_ean"] = None
    item["source_title"] = f"Example Mobile 13 {suffix}"
    item.update(overrides)
    return item


class ScriptedAdapter:
    source_key = "geizhals"

    def __init__(self, pages: list[dict]) -> None:
        self.pages = pages
        self._inner = GeizhalsAdapter()

    def discover_taxonomy(self, category_ref: str) -> dict:
        return self._inner.discover_taxonomy(category_ref)

    def build_query(self, criteria, mappings) -> dict:
        return self._inner.build_query(criteria, mappings)

    def fetch_results(self, query, pagination) -> dict:
        page = int(pagination.get("page") or 1)
        return self.pages[page - 1]

    def parse_results(self, response) -> list:
        return list(response.get("items") or [])

    def normalize_result(self, result) -> dict:
        return self._inner.normalize_result(result)


@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}")


def _session():
    session = SessionLocal()
    run_bootstrap(session)
    seed_notebook_pilot(session)
    ingestion = IngestionService(session)
    for definition in DEFINITIONS:
        try:
            ingestion.create_definition(definition)
        except AppError as exc:
            if exc.status_code != 409:
                raise
    try:
        ingestion.create_profile(PROFILE)
    except AppError as exc:
        if exc.status_code != 409:
            raise
    return session


def _prepare_search(session, adapter, threshold=300):
    taxonomy = TaxonomyService(session, adapter).discover("geizhals", "nb")
    mappings = CriterionMappingService(session)
    mappings.create(
        {
            "criterion_key": "ram_gb",
            "source_id": "geizhals",
            "source_category_id": taxonomy["id"],
            "external_filter_key": "17023",
            "source_value": "ab 24GB",
            "encoding": {"url_param": "xf"},
        }
    )
    project = ProjectService(session).create(
        {
            "name": "Lifecycle",
            "profile_id": "mobile-development-notebook",
            "profile_version": 1,
            "criteria": {"items": [{"id": "ram", "criterion": "ram_gb", "operator": ">=", "value": 24}]},
        }
    )
    search = DiscoverySearchService(session, adapter).create(
        {
            "buying_project_id": project["id"],
            "source_id": "geizhals",
            "source_category_id": taxonomy["id"],
            "broad_result_threshold": threshold,
        }
    )
    return search["id"]


def _page(items, status=200, page_count=1):
    return {"page": 1, "page_count": page_count, "result_count": len(items), "http_status": status, "items": items}


def test_partial_does_not_mark_missing_or_disappeared():
    session = _session()
    try:
        adapter = ScriptedAdapter([_page([_item("partial")])])
        search_id = _prepare_search(session, adapter)
        first = SearchExecutionService(session, adapter).execute(search_id)
        assert first["status"] == RunStatus.SUCCESS.value
        ref_id = first["items"][0]["source_product_reference_id"]
        adapter.pages = [_page([], status=429)]
        second = SearchExecutionService(session, adapter).execute(search_id)
        assert second["status"] == RunStatus.PARTIAL.value
        assert all(item["diff_status"] != DiffStatus.MISSING.value for item in second["items"])
        ref = session.get(SourceProductReferenceRecord, ref_id)
        assert ref is not None
        assert ref.lifecycle_status == LifecycleStatus.ACTIVE.value
        assert ref.consecutive_missing_success_runs == 0
    finally:
        session.close()


def test_three_success_misses_disappear_and_reappear():
    session = _session()
    try:
        item = _item("gone")
        adapter = ScriptedAdapter([_page([item])])
        search_id = _prepare_search(session, adapter)
        first = SearchExecutionService(session, adapter).execute(search_id)
        ref_id = first["items"][0]["source_product_reference_id"]
        adapter.pages = [_page([])]
        for _ in range(3):
            SearchExecutionService(session, adapter).execute(search_id)
        ref = session.get(SourceProductReferenceRecord, ref_id)
        assert ref is not None
        assert ref.lifecycle_status == LifecycleStatus.DISAPPEARED.value
        changed = deepcopy(item)
        changed["best_price"] = "999.00"
        adapter.pages = [_page([changed])]
        run = SearchExecutionService(session, adapter).execute(search_id)
        session.refresh(ref)
        assert ref.lifecycle_status == LifecycleStatus.ACTIVE.value
        assert any(item["diff_status"] == DiffStatus.CHANGED.value for item in run["items"])
        assert all(item["diff_status"] != DiffStatus.NEW.value for item in run["items"])
        prices = [
            row.best_price
            for row in session.query(ProductObservationRecord).filter(
                ProductObservationRecord.source_product_reference_id == ref_id
            )
        ]
        assert Decimal("1298.00") in prices
        assert Decimal("999.00") in prices
    finally:
        session.close()


def test_uncertain_match_stays_separate():
    session = _session()
    try:
        first = _item("matched", manufacturer="Example", model="Uncertain 13")
        second = _item(
            "uncertain",
            source_external_id=None,
            manufacturer_sku=None,
            source_url="https://geizhals.de/other-uncertain.html",
            source_title=first["source_title"],
            manufacturer="Example",
            model="Uncertain 13",
            normalized_attributes={"ram_gb": 16, "storage_gb": 512, "weight_kg": 2.0},
        )
        adapter = ScriptedAdapter([_page([first, second])])
        search_id = _prepare_search(session, adapter)
        project_id = DiscoverySearchService(session, adapter).get(search_id)["buying_project_id"]
        run = SearchExecutionService(session, adapter).execute(search_id)
        assert run["status"] == RunStatus.SUCCESS.value
        ref_ids = {item["source_product_reference_id"] for item in run["items"]}
        assert len(ref_ids) == 2
        refs = [session.get(SourceProductReferenceRecord, ref_id) for ref_id in ref_ids]
        assert MatchStatus.MATCH_UNCERTAIN.value in {ref.match_status for ref in refs if ref is not None}
        candidates = (
            session.query(ProjectCandidateRecord)
            .filter(ProjectCandidateRecord.buying_project_id == project_id)
            .all()
        )
        assert len(candidates) == 2
        assert MatchStatus.MATCH_UNCERTAIN.value in {row.match_status for row in candidates}
    finally:
        session.close()


def test_backoff_and_blocked():
    session = _session()
    try:
        adapter = ScriptedAdapter([_page([ITEM_A], status=429)])
        search_id = _prepare_search(session, adapter)
        backoff = SearchExecutionService(session, adapter).execute(search_id)
        assert backoff["status"] == RunStatus.PARTIAL.value
        assert backoff["error_code"] == "BACKOFF"
        adapter.pages = [_page([ITEM_A], status=403)]
        blocked = SearchExecutionService(session, adapter).execute(search_id)
        assert blocked["status"] == RunStatus.BLOCKED.value
        stored = DiscoverySearchService(session, adapter).get(search_id)
        assert stored["status"] == "BLOCKED"
        assert stored["next_run_at"] is None
    finally:
        session.close()