spb/immo-ka

1# -----------------------------------------------------------------------------2# Immo-Ka — Agrégateur de maisons à vendre (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# tests/test_mortgage_store.py : historisation par périodes de validité —5#   inchangé = simple last_checked ; changé = fermeture + nouvelle ligne ;6#   saut aberrant rejeté sans écraser la bonne donnée.7# -----------------------------------------------------------------------------8import tempfile9import time10import unittest11from pathlib import Path1213from immoka.mortgage import store141516def product(**over) -> dict:17    base = {18        "provider": "test", "institution": "Banque Test",19        "rate_type": "fixed", "term_months": 60, "kind": "special",20        "rate": 4.84, "apr": 4.90, "insured_status": "unknown",21        "purpose": "purchase", "product_name": "Fixe fermé 5 ans (spécial)",22        "source_url": "https://example.com/taux", "raw": {"src": "test"},23    }24    base.update(over)25    return base262728class StoreTest(unittest.TestCase):29    def setUp(self):30        self._tmp = tempfile.TemporaryDirectory()31        self._old_path = store.DB_PATH32        self._old_ready = store._SCHEMA_READY33        store.DB_PATH = Path(self._tmp.name) / "mortgage.db"34        store._SCHEMA_READY = False35        self.con = store.connect()3637    def tearDown(self):38        self.con.close()39        store.DB_PATH = self._old_path40        store._SCHEMA_READY = self._old_ready41        self._tmp.cleanup()4243    def _count(self, where: str = "1=1") -> int:44        return self.con.execute(45            f"SELECT COUNT(*) AS n FROM rate_observations WHERE {where}"46        ).fetchone()["n"]4748    def test_unchanged_updates_last_checked_only(self):49        store.record_observations(self.con, "test", [product()])50        first = self.con.execute(51            "SELECT last_checked FROM rate_observations").fetchone()[0]52        time.sleep(0.02)53        res = store.record_observations(self.con, "test", [product()])54        self.assertEqual(res["changed"], 0)55        self.assertEqual(self._count(), 1)  # aucune nouvelle ligne56        second = self.con.execute(57            "SELECT last_checked FROM rate_observations").fetchone()[0]58        self.assertGreater(second, first)5960    def test_change_closes_and_inserts(self):61        store.record_observations(self.con, "test", [product(rate=4.84)])62        res = store.record_observations(self.con, "test", [product(rate=4.69)])63        self.assertEqual(res["changed"], 1)64        self.assertEqual(self._count(), 2)65        self.assertEqual(self._count("valid_to IS NULL"), 1)66        current = self.con.execute(67            "SELECT rate FROM rate_observations WHERE valid_to IS NULL"68        ).fetchone()["rate"]69        self.assertEqual(current, 4.69)70        closed = self.con.execute(71            "SELECT rate, valid_to FROM rate_observations "72            "WHERE valid_to IS NOT NULL").fetchone()73        self.assertEqual(closed["rate"], 4.84)  # historique intact7475    def test_aberrant_jump_rejected(self):76        store.record_observations(self.con, "test", [product(rate=4.19)])77        res = store.record_observations(self.con, "test", [product(rate=8.50)])78        self.assertEqual(res["rejected"], 1)79        self.assertEqual(self._count(), 1)  # la bonne donnée n'est pas écrasée80        current = self.con.execute(81            "SELECT rate FROM rate_observations WHERE valid_to IS NULL"82        ).fetchone()["rate"]83        self.assertEqual(current, 4.19)8485    def test_small_change_accepted(self):86        store.record_observations(self.con, "test", [product(rate=4.19)])87        res = store.record_observations(self.con, "test", [product(rate=4.44)])88        self.assertEqual(res["rejected"], 0)89        self.assertEqual(res["changed"], 1)9091    def test_product_key_separates_incomparables(self):92        keys = {store.product_key(product(**over)) for over in (93            {}, {"kind": "posted"}, {"insured_status": "insured"},94            {"term_months": 36}, {"rate_type": "variable"},95            {"product_name": "Autre produit"},96        )}97        self.assertEqual(len(keys), 6)  # jamais comparer l'incomparable9899    def test_best_rate_special_beats_posted(self):100        store.record_observations(self.con, "a", [101            product(provider="a", kind="posted", rate=6.09,102                    product_name="Fixe affiché"),103            product(provider="a", kind="special", rate=4.84,104                    product_name="Fixe spécial"),105        ])106        store.record_observations(self.con, "b", [107            product(provider="b", institution="Banque B", kind="special",108                    rate=4.99, product_name="Fixe spécial B"),109        ])110        best = store.best_rate(self.con, "fixed", 60)111        self.assertEqual(best["provider"], "a")112        self.assertEqual(best["rate"], 4.84)113        self.assertEqual(best["kind"], "special")114        self.assertEqual(best["institutions_count"], 2)115        # le posted 6,09 de « a » ne doit pas être candidat116        self.assertEqual([r["rate"] for r in best["per_institution"]],117                         [4.84, 4.99])118119    def test_best_rate_none_when_empty(self):120        self.assertIsNone(store.best_rate(self.con, "fixed", 60))121122    def test_rate_at_now(self):123        store.record_observations(self.con, "test", [product(rate=4.84)])124        self.assertEqual(store.rate_at(self.con, "fixed", 60, time.time()), 4.84)125126    def test_history_returns_periods(self):127        store.record_observations(self.con, "test", [product(rate=4.84)])128        store.record_observations(self.con, "test", [product(rate=4.69)])129        rows = store.history(self.con, "fixed", 60)130        self.assertEqual(len(rows), 2)131        self.assertIsNotNone(rows[0]["valid_to"])132        self.assertIsNone(rows[1]["valid_to"])133134    def test_provider_health_levels(self):135        store.record_observations(self.con, "test", [product()])136        store.log_run(self.con, "test", ok=True, status="success", products=1)137        store.log_run(self.con, "vide", ok=False, status="http_error")138        health = {h["provider"]: h for h in store.provider_health(self.con)}139        self.assertEqual(health["test"]["level"], "OK")140        self.assertEqual(health["vide"]["level"], "ERROR")141142143if __name__ == "__main__":144    unittest.main()145