60 lines
2.3 KiB
Python
60 lines
2.3 KiB
Python
|
from pathlib import Path
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
from sqlalchemy import text
|
||
|
|
|
||
|
|
from app.persistence.db import SessionLocal, engine
|
||
|
|
from app.services.import_service import ImportService
|
||
|
|
from app.services.ingestion_service import IngestionService
|
||
|
|
from app.sources.manual import ManualSource
|
||
|
|
from scripts.bootstrap import run_bootstrap
|
||
|
|
|
||
|
|
pytestmark = pytest.mark.integration
|
||
|
|
ROOT = Path(__file__).resolve().parents[2]
|
||
|
|
|
||
|
|
|
||
|
|
@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 test_manual_source_import_is_idempotent() -> None:
|
||
|
|
session = SessionLocal()
|
||
|
|
try:
|
||
|
|
run_bootstrap(session)
|
||
|
|
ingestion = IngestionService(session)
|
||
|
|
for item in (
|
||
|
|
{"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},
|
||
|
|
):
|
||
|
|
try:
|
||
|
|
ingestion.create_definition(item)
|
||
|
|
except Exception:
|
||
|
|
session.rollback()
|
||
|
|
source = ManualSource(ROOT / "fixtures" / "manual-demo.json")
|
||
|
|
importer = ImportService(ingestion)
|
||
|
|
importer.import_product(source, "example-mobile-13")
|
||
|
|
session.commit()
|
||
|
|
importer.import_product(source, "example-mobile-13")
|
||
|
|
session.commit()
|
||
|
|
from sqlalchemy import func, select
|
||
|
|
|
||
|
|
from app.persistence.models import OfferRecord, PriceObservationRecord, ProductRecord
|
||
|
|
|
||
|
|
products = session.scalar(select(func.count()).select_from(ProductRecord))
|
||
|
|
offers = session.scalar(select(func.count()).select_from(OfferRecord))
|
||
|
|
prices = session.scalar(select(func.count()).select_from(PriceObservationRecord))
|
||
|
|
assert products >= 1
|
||
|
|
assert offers >= 4
|
||
|
|
assert prices >= 16
|
||
|
|
finally:
|
||
|
|
session.close()
|