"""
Unit tests for one-time/idempotent data-fix migrations in server/db.py.

Run with: cd server && ../venv/bin/python3 -m unittest tests.test_db_migrations -v
"""
import os
import sys
import tempfile
import unittest

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))


class DedupeAuditCriteriaTests(unittest.TestCase):
    def setUp(self):
        self._tmpdir = tempfile.mkdtemp()
        os.environ["CA_DATA_DIR"] = self._tmpdir
        sys.modules.pop("db", None)
        import db as db_mod
        global db
        db = db_mod
        db.init_db()

    def test_fresh_install_has_no_duplicates(self):
        rows = db.query("SELECT area, criterion, COUNT(*) n FROM audit_criterion GROUP BY area, criterion HAVING n > 1")
        self.assertEqual(rows, [])

    def test_fresh_install_creates_the_unique_index(self):
        idx = db.query("SELECT name FROM sqlite_master WHERE type='index' AND name='ux_audit_criterion_area_criterion'", one=True)
        self.assertIsNotNone(idx)

    def test_reinitializing_is_a_no_op_on_a_clean_table(self):
        before = db.query("SELECT COUNT(*) n FROM audit_criterion", one=True)["n"]
        db.init_db()
        after = db.query("SELECT COUNT(*) n FROM audit_criterion", one=True)["n"]
        self.assertEqual(before, after)

    def test_preexisting_duplicates_are_cleaned_up_and_lowest_id_kept(self):
        # Simulates the real bug this fix addresses: a database that
        # accumulated duplicate rows before the unique index existed.
        with db.conn() as c:
            c.execute("DROP INDEX IF EXISTS ux_audit_criterion_area_criterion")
            row = c.execute("SELECT id, area, criterion, applies_to, sort_order, is_builtin"
                            " FROM audit_criterion LIMIT 1").fetchone()
            original_id = row["id"]
            for _ in range(6):  # 7 copies total, matching the real incident
                c.execute("INSERT INTO audit_criterion (area, criterion, applies_to, sort_order, is_builtin)"
                          " VALUES (?,?,?,?,?)",
                          (row["area"], row["criterion"], row["applies_to"], row["sort_order"], row["is_builtin"]))
        dup_count = db.query("SELECT COUNT(*) n FROM audit_criterion WHERE area = ? AND criterion = ?",
                             (row["area"], row["criterion"]), one=True)["n"]
        self.assertEqual(dup_count, 7)

        db.init_db()

        surviving = db.query("SELECT id FROM audit_criterion WHERE area = ? AND criterion = ?",
                             (row["area"], row["criterion"]))
        self.assertEqual(len(surviving), 1)
        self.assertEqual(surviving[0]["id"], original_id)

    def test_unique_index_prevents_future_duplicates_from_seed_reruns(self):
        row = db.query("SELECT area, criterion, applies_to, sort_order, is_builtin FROM audit_criterion LIMIT 1", one=True)
        before = db.query("SELECT COUNT(*) n FROM audit_criterion", one=True)["n"]
        db.execute("INSERT OR IGNORE INTO audit_criterion (area, criterion, applies_to, sort_order, is_builtin)"
                  " VALUES (?,?,?,?,?)",
                  (row["area"], row["criterion"], row["applies_to"], row["sort_order"], row["is_builtin"]))
        after = db.query("SELECT COUNT(*) n FROM audit_criterion", one=True)["n"]
        self.assertEqual(before, after)


class DedupeRateMasterTests(unittest.TestCase):
    """rate_master's original UNIQUE (domain, code, fiscal_year_label,
    applies_to) never deduped a standing rule (NULL fiscal_year_label —
    every India row's convention) against itself, the same NULL-vs-NULL gap
    DedupeAuditCriteriaTests above already covers for a different table.
    Confirmed live in a rehearsal MySQL database before this fix existed:
    init_db(force_seed=True) doubled every India row. See
    db._fix_rate_master_dedup()."""

    def setUp(self):
        self._tmpdir = tempfile.mkdtemp()
        os.environ["CA_DATA_DIR"] = self._tmpdir
        sys.modules.pop("db", None)
        import db as db_mod
        global db
        db = db_mod
        db.init_db()

    def test_reinitializing_a_standing_rule_row_is_a_no_op(self):
        # A standing rule (NULL fiscal_year_label) is exactly the shape
        # every India rate_master row uses — the case the original
        # constraint silently failed to dedupe.
        before = db.query("SELECT COUNT(*) n FROM rate_master WHERE country_code = 'IN'", one=True)["n"]
        db.init_db(force_seed=True)
        after = db.query("SELECT COUNT(*) n FROM rate_master WHERE country_code = 'IN'", one=True)["n"]
        self.assertGreater(before, 0)
        self.assertEqual(before, after)

    def test_reinitializing_twice_is_still_a_no_op(self):
        db.init_db(force_seed=True)
        once = db.query("SELECT COUNT(*) n FROM rate_master", one=True)["n"]
        db.init_db(force_seed=True)
        twice = db.query("SELECT COUNT(*) n FROM rate_master", one=True)["n"]
        self.assertEqual(once, twice)

    def test_unique_index_prevents_a_direct_duplicate_insert(self):
        row = db.query("SELECT domain, code, label, value_num, unit, applies_to, country_code"
                       " FROM rate_master WHERE country_code = 'IN' LIMIT 1", one=True)
        with self.assertRaises(db.IntegrityError):
            db.execute(
                "INSERT INTO rate_master (domain, code, label, value_num, unit, applies_to, country_code,"
                " verification_status) VALUES (?,?,?,?,?,?,?,'UNVERIFIED')",
                (row["domain"], row["code"], row["label"], row["value_num"], row["unit"],
                 row["applies_to"], row["country_code"]))

    def test_preexisting_duplicates_are_cleaned_up_keeping_the_lowest_id(self):
        # Simulates the real incident this fix addresses: a database that
        # accumulated a duplicate standing-rule row before the corrected
        # index existed.
        row = db.query("SELECT id, domain, code, label, value_num, unit, applies_to, country_code"
                       " FROM rate_master WHERE country_code = 'IN' LIMIT 1", one=True)
        original_id = row["id"]
        with db.conn() as c:
            c.execute("DROP INDEX ux_rate_master_dedup" + (" ON rate_master" if db.DB_ENGINE == "mysql" else ""))
            c.execute(
                "INSERT INTO rate_master (domain, code, label, value_num, unit, applies_to, country_code,"
                " verification_status) VALUES (?,?,?,?,?,?,?,'UNVERIFIED')",
                (row["domain"], row["code"], row["label"], row["value_num"], row["unit"],
                 row["applies_to"], row["country_code"]))
        dup_count = db.query(
            "SELECT COUNT(*) n FROM rate_master WHERE domain = ? AND code = ? AND country_code = ?",
            (row["domain"], row["code"], row["country_code"]), one=True)["n"]
        self.assertEqual(dup_count, 2)

        db.init_db()

        surviving = db.query(
            "SELECT id FROM rate_master WHERE domain = ? AND code = ? AND country_code = ?",
            (row["domain"], row["code"], row["country_code"]))
        self.assertEqual(len(surviving), 1)
        self.assertEqual(surviving[0]["id"], original_id)


class PruneSnapshotsTests(unittest.TestCase):
    """db.prune_snapshots()'s keep_per_file query used a derived table with
    no alias — SQLite tolerates that, but it's invalid under MySQL ("Every
    derived table must have its own alias"), so it silently never ran under
    CA_DB_ENGINE=mysql (swallowed by cmd_worker's broad except). These tests
    exercise the actual pruning behaviour, which stays correct after adding
    the alias — SQLite accepts an aliased derived table too."""

    def setUp(self):
        self._tmpdir = tempfile.mkdtemp()
        os.environ["CA_DATA_DIR"] = self._tmpdir
        sys.modules.pop("db", None)
        import db as db_mod
        global db
        db = db_mod
        db.init_db()
        # autosave_snapshot.workfile_id is a real FK — build the minimal
        # client -> engagement -> workfile chain for ids 1 and 2 rather than
        # relying on FK enforcement being off.
        cid = db.execute("INSERT INTO client (firm_id, code, name) VALUES (1, 'CL-01', 'Test Client')")
        eid = db.execute("INSERT INTO engagement (client_id, type) VALUES (?, 'advisory')", (cid,))
        db.execute("INSERT INTO workfile (engagement_id, module_key, title) VALUES (?, 'depreciation', 'WF 1')", (eid,))
        db.execute("INSERT INTO workfile (engagement_id, module_key, title) VALUES (?, 'depreciation', 'WF 2')", (eid,))

    def _snapshot(self, workfile_id, saved_at, origin="auto"):
        db.execute(
            "INSERT INTO autosave_snapshot (workfile_id, payload, origin, saved_at) VALUES (?, '{}', ?, ?)",
            (workfile_id, origin, saved_at))

    def test_old_auto_snapshots_are_pruned_by_retention_days(self):
        self._snapshot(1, "2000-01-01 00:00:00", origin="auto")
        self._snapshot(1, "2999-01-01 00:00:00", origin="auto")
        db.prune_snapshots(retention_days=120, keep_per_file=200)
        rows = db.query("SELECT saved_at FROM autosave_snapshot WHERE workfile_id = 1")
        self.assertEqual([r["saved_at"] for r in rows], ["2999-01-01 00:00:00"])

    def test_old_manual_snapshots_survive_the_retention_cutoff(self):
        self._snapshot(1, "2000-01-01 00:00:00", origin="manual")
        db.prune_snapshots(retention_days=120, keep_per_file=200)
        rows = db.query("SELECT id FROM autosave_snapshot WHERE workfile_id = 1")
        self.assertEqual(len(rows), 1)

    def test_keeps_only_the_newest_n_per_workfile(self):
        for i in range(5):
            self._snapshot(1, f"2999-01-0{i + 1} 00:00:00", origin="auto")
        db.prune_snapshots(retention_days=120, keep_per_file=2)
        rows = db.query("SELECT saved_at FROM autosave_snapshot WHERE workfile_id = 1 ORDER BY saved_at")
        self.assertEqual([r["saved_at"] for r in rows], ["2999-01-04 00:00:00", "2999-01-05 00:00:00"])

    def test_keep_per_file_is_scoped_per_workfile_not_globally(self):
        for i in range(3):
            self._snapshot(1, f"2999-01-0{i + 1} 00:00:00", origin="auto")
        for i in range(3):
            self._snapshot(2, f"2999-01-0{i + 1} 00:00:00", origin="auto")
        db.prune_snapshots(retention_days=120, keep_per_file=2)
        self.assertEqual(db.query("SELECT COUNT(*) n FROM autosave_snapshot WHERE workfile_id = 1", one=True)["n"], 2)
        self.assertEqual(db.query("SELECT COUNT(*) n FROM autosave_snapshot WHERE workfile_id = 2", one=True)["n"], 2)


if __name__ == "__main__":
    unittest.main()
