# SPDX-License-Identifier: MIT # Copyright (c) 2026 Rug Munch Media LLC # # Part of Pry — https://git.rugmunch.io/RugMunchMedia/pryscraper # Licensed under MIT. See LICENSE. """Tests for competitive intelligence engine.""" import time from collections.abc import Iterator import pytest from db import Base, get_engine from intelligence import ( compute_field_statistics, detect_anomalies_numeric, generate_alert, generate_weekly_report, get_snapshots, record_snapshot, ) @pytest.fixture(autouse=True) def _intel_db(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]: """Use in-memory SQLite for each test.""" monkeypatch.setenv("PRY_DATABASE_URL", "sqlite://") import db as db_module db_module._engine = None db_module._SessionLocal = None engine = get_engine() Base.metadata.create_all(engine) yield Base.metadata.drop_all(engine) # -- Snapshot recording -- def test_record_snapshot() -> None: result = record_snapshot( "comp_1", "Competitor A", "https://example.com", {"price": 99.99, "stock": "in_stock"}, ) assert "ts" in result assert "unix_ts" in result assert result["competitor_name"] == "Competitor A" assert result["competitor_id"] == "comp_1" assert result["fields"]["price"] == 99.99 def test_record_snapshot_returns_iso_ts() -> None: result = record_snapshot("comp_ts", "TS Co", "https://example.com", {"x": 1}) ts = result["ts"] assert isinstance(ts, str) assert ts.endswith("Z") or "+" in ts def test_record_snapshot_unix_ts() -> None: before = time.time() result = record_snapshot("comp_ut", "UT Co", "https://example.com", {"x": 1}) after = time.time() assert before <= result["unix_ts"] <= after + 1 # -- Retreiving snapshots -- def test_get_snapshots() -> None: record_snapshot("comp_test", "Test", "https://example.com", {"a": 1}) snapshots = get_snapshots("comp_test") assert len(snapshots) >= 1 assert snapshots[0]["competitor_name"] == "Test" assert snapshots[0]["fields"]["a"] == 1 def test_get_snapshots_limit() -> None: for i in range(10): record_snapshot("comp_limit", "Limit Co", "https://example.com", {"i": i}) snapshots = get_snapshots("comp_limit", limit=3) assert len(snapshots) == 3 def test_get_snapshots_ordered_by_ts_desc() -> None: for i in range(5): record_snapshot("comp_order", "Order Co", "https://example.com", {"i": i}) time.sleep(0.01) snapshots = get_snapshots("comp_order") assert len(snapshots) == 5 assert snapshots[0]["fields"]["i"] == 4 def test_get_snapshots_since_hours() -> None: record_snapshot("comp_since", "Since Co", "https://example.com", {"val": 1}) snapshots = get_snapshots("comp_since", since_hours=1) assert len(snapshots) >= 1 def test_get_snapshots_empty_for_unknown() -> None: snapshots = get_snapshots("nonexistent_comp") assert snapshots == [] # -- Field statistics -- class TestComputeFieldStatistics: def test_numeric(self) -> None: snapshots = [ {"fields": {"price": 100}}, {"fields": {"price": 110}}, {"fields": {"price": 105}}, ] stats = compute_field_statistics(snapshots, "price") assert stats["has_history"] is True assert stats["mean"] == 105.0 assert stats["min"] == 100 assert stats["max"] == 110 def test_insufficient_data(self) -> None: snapshots = [{"fields": {"price": 100}}] stats = compute_field_statistics(snapshots, "price") assert stats["has_history"] is False def test_empty_snapshots(self) -> None: stats = compute_field_statistics([], "price") assert stats["count"] == 0 assert stats["has_history"] is False def test_missing_field(self) -> None: snapshots = [{"fields": {"price": 100}}, {"fields": {"other": 200}}] stats = compute_field_statistics(snapshots, "price") assert stats["count"] == 1 assert stats["has_history"] is False def test_string_field(self) -> None: snapshots = [ {"fields": {"status": "active"}}, {"fields": {"status": "active"}}, {"fields": {"status": "paused"}}, ] stats = compute_field_statistics(snapshots, "status") assert stats["has_history"] is True assert stats["unique_values"] == 2 assert stats["latest"] == "paused" # -- Anomaly detection -- class TestDetectAnomaliesNumeric: def test_detects_anomaly(self) -> None: history = [100, 102, 101, 103, 99, 101, 102] result = detect_anomalies_numeric(150, history) assert result["anomaly"] is True assert result["z_score"] >= 2.0 def test_insufficient_history(self) -> None: result = detect_anomalies_numeric(100, []) assert result["anomaly"] is False assert "Insufficient history" in result["reason"] def test_no_anomaly(self) -> None: history = [100, 101, 100, 101, 100] result = detect_anomalies_numeric(100, history) assert result["anomaly"] is False def test_constant_history_same_value(self) -> None: history = [100, 100, 100] result = detect_anomalies_numeric(100, history) assert result["anomaly"] is False def test_constant_history_diff_value(self) -> None: history = [100, 100, 100] result = detect_anomalies_numeric(200, history) assert result["anomaly"] is True assert "constant history" in result["reason"] # -- Alerts -- class TestGenerateAlert: def test_price_increase(self) -> None: alert = generate_alert("Competitor A", "price", 100, 120) assert "Competitor A" in alert assert "increased" in alert def test_price_decrease(self) -> None: alert = generate_alert("Competitor B", "price", 200, 150) assert "decreased" in alert def test_text_change_short(self) -> None: alert = generate_alert("Competitor C", "description", "Old desc", "New and improved") assert "Competitor C" in alert assert '"Old desc"' in alert def test_text_change_long(self) -> None: long_old = "x" * 60 long_new = "y" * 60 alert = generate_alert("Competitor D", "description", long_old, long_new) assert "chars" in alert def test_with_anomaly_info(self) -> None: alert = generate_alert( "Competitor F", "price", 100, 200, anomaly_info={"anomaly": True, "z_score": 3.5, "pct_change": 100.0}, ) assert "ANOMALY DETECTED" in alert assert "z-score" in alert # -- Weekly report -- class TestGenerateWeeklyReport: def test_basic_report(self) -> None: record_snapshot("comp_rpt", "Report Co", "https://example.com", {"price": 50}) report = generate_weekly_report([{"id": "comp_rpt", "name": "Report Co"}], days_back=7) assert "report_period" in report assert "generated_at" in report assert report["competitors_tracked"] == 1 def test_report_with_changes(self) -> None: record_snapshot("comp_chg", "Chg Co", "https://example.com", {"price": 100}) record_snapshot("comp_chg", "Chg Co", "https://example.com", {"price": 150}) report = generate_weekly_report([{"id": "comp_chg", "name": "Chg Co"}], days_back=7) assert report["total_changes"] >= 1 assert report["most_active_competitor"] == "Chg Co" def test_report_no_activity(self) -> None: report = generate_weekly_report( [{"id": "comp_inactive", "name": "Inactive Co"}], days_back=7, ) assert report["total_changes"] == 0 def test_report_multiple_competitors(self) -> None: record_snapshot("comp_a", "Alpha Co", "https://example.com", {"price": 50}) record_snapshot("comp_b", "Beta Co", "https://example.com", {"price": 75}) competitors = [ {"id": "comp_a", "name": "Alpha Co"}, {"id": "comp_b", "name": "Beta Co"}, ] report = generate_weekly_report(competitors, days_back=7) assert report["competitors_tracked"] == 2