"""Soft-deleted rows must not be mistaken for seeded rows.

Three places in the applier used to count them:

* `pre_resolve_existing_ids` looks a seeded row up by its `match` fields and adopts the id of
  the row already in the table. Without a soft-delete predicate that lookup also sees deleted
  rows. An object whose match fields are immutable can only be corrected by deleting and
  re-creating a row, so the old row and its live replacement share a match key: the lookup
  finds two ids and aborts the phase with `duplicate existing <object> match`. If only the
  deleted row exists, its id is adopted and the seed writes into a row nothing can see.
* the post-upsert verification counted rows by id with no such predicate, so a write that
  landed on a soft-deleted row -- the ids are deterministic, so an upsert can collide with one
  by primary key -- reported success.

The verification is now live-only, but it is a backstop, not the gate: `apply_rows` sends its
batches through the mysql CLI with no enclosing transaction, so anything it detects has
already committed, along with every earlier object in the layer. `preflight_database` is the
gate. It searches by match key *and* by seed id, because neither route contains the other: a
row that pre-dates the seed holds a native id and collides on a natural unique key, while a
row whose updateable match field was edited before it was deleted still carries the seed id
and collides on the primary key.

Note what is deliberately *not* here: nothing revives a soft-deleted row. Seeding leaves the
deleted row deleted and fails loudly, because an operator deleting a row through the UI is the
only retirement mechanism these objects have, and silently undoing it on the next deployment
would invert that. `IsDeleted` is in `NEVER_UPDATE_COLUMNS` so that stays true no matter what
a reference-data source declares.

These assertions capture generated SQL and stubbed query output rather than talking to a
database, so they run anywhere. The behaviour was also reproduced against a live table before
the fix.

Two fixture choices are deliberate:

* the read-side assertions use two seed rows and match each statement exactly. With a single
  row the parentheses around the OR-chain are inert and a mutation that drops them -- the
  original defect -- passes unnoticed; and asserting that a predicate is *present* is
  satisfied by a predicate that is present and wrong.
* the column fixtures carry no audit columns (`CreatedDate`, `LastModifiedDate`,
  `SystemModstamp`, `OwnerId`), which keeps every emitted statement deterministic enough to
  match exactly. The cost is that `normalize_row`'s audit-stamping loops and
  `fetch_admin_user_id` are not exercised here. Nothing in this file's subject touches them,
  and statements are selected by shape rather than by position so an added query cannot
  silently shift an assertion onto the wrong one.
"""
from __future__ import annotations

import importlib.util
import pathlib
import unittest

REPO = pathlib.Path(__file__).resolve().parents[1]
APPLIER = REPO / "scripts/apply_pack_reference_data.py"

CANONICAL_COLUMNS = {
    "id": "Id",
    "isocode": "IsoCode",
    "startdate": "StartDate",
    "conversionrate": "ConversionRate",
    "isdeleted": "IsDeleted",
}
NO_SOFT_DELETE_COLUMNS = {k: v for k, v in CANONICAL_COLUMNS.items() if k != "isdeleted"}
# The shipped source matches on Name, whose values contain spaces ("CNY 1900-01-01"). Row
# parsing must stay tab-delimited; splitting on whitespace would drop every row silently.
NAME_MATCH_COLUMNS = {"id": "Id", "name": "Name", "conversionrate": "ConversionRate"}


def _load_applier():
    spec = importlib.util.spec_from_file_location("apply_pack_reference_data", APPLIER)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def _row(iso: str, start: str, rate: float, row_id: str) -> dict:
    return {
        "object": "DatedConversionRate",
        "key": f"dated_conversion_rate_{iso}_{start}",
        "id": row_id,
        "alias_keys": [],
        "match": {"IsoCode": iso, "StartDate": start},
        "fields": {"ConversionRate": rate},
    }


def _named_row(name: str, rate: float, row_id: str) -> dict:
    return {
        "object": "DatedConversionRate",
        "key": f"dated_conversion_rate_{name}",
        "id": row_id,
        "alias_keys": [],
        "match": {"Name": name},
        "fields": {"ConversionRate": rate},
    }


TWO_ROWS = [
    _row("JPY", "2026-04-01", 159.651, "seed-jpy"),
    _row("EUR", "2026-01-01", 0.9214, "seed-eur"),
]


class SoftDeleteApplierTests(unittest.TestCase):
    def setUp(self):
        self.applier = _load_applier()
        self.captured: list[str] = []

    def _stub(self, columns: dict[str, str], query_results: list):
        """Point the applier's three database calls at in-memory stand-ins.

        A result may be an exception instance, which the fake raises instead of returning --
        that is how a probe failure is simulated.
        """
        applier = self.applier
        pending = list(query_results)
        self.captured = []

        def fake_query(_args, sql):
            self.captured.append(sql)
            result = pending.pop(0) if pending else ""
            if isinstance(result, Exception):
                raise result
            return result

        originals = (applier.resolve_table_name, applier.fetch_columns, applier.mysql_query)
        applier.resolve_table_name = lambda *_a, **_k: "DatedConversionRate"
        applier.fetch_columns = lambda *_a, **_k: columns
        applier.mysql_query = fake_query
        return originals

    def _restore(self, originals):
        (self.applier.resolve_table_name,
         self.applier.fetch_columns,
         self.applier.mysql_query) = originals

    def _only(self, prefix: str) -> str:
        """The one captured statement starting with `prefix`, by shape rather than position."""
        matches = [s for s in self.captured if s.startswith(prefix)]
        self.assertEqual(len(matches), 1, f"want exactly one {prefix!r} statement in {self.captured}")
        return matches[0]

    def _resolve(self, columns: dict[str, str], rows: list[dict], results: list) -> list[dict]:
        originals = self._stub(columns, results)
        try:
            self.applier.pre_resolve_existing_ids(object(), "DatedConversionRate", rows)
        finally:
            self._restore(originals)
        return rows

    def _apply(self, columns: dict[str, str], rows: list[dict], results: list):
        originals = self._stub(columns, [""] + results)
        try:
            self.applier.apply_rows(
                object(), "DatedConversionRate", rows, "DatedConversionRate", columns
            )
        finally:
            self._restore(originals)

    # --- match resolution: the query ---------------------------------------------------

    def test_match_query_excludes_soft_deleted_rows(self):
        """The predicate is ANDed onto the *whole* OR-chain, and NULL reads as live.

        Losing the parentheses leaves the filter applying to the last branch only, which is
        the original defect. Reading NULL as deleted would hide an existing row from match
        resolution and produce a duplicate insert or a natural-key failure.
        """
        self._resolve(CANONICAL_COLUMNS, list(TWO_ROWS), [])
        self.assertEqual(
            self._only("SELECT"),
            "SELECT `Id`, `IsoCode`, `StartDate` FROM `DatedConversionRate` "
            "WHERE ((`IsoCode` <=> 'JPY' AND `StartDate` <=> '2026-04-01') "
            "OR (`IsoCode` <=> 'EUR' AND `StartDate` <=> '2026-01-01')) "
            "AND COALESCE(`IsDeleted`, 0) = 0",
        )

    def test_match_query_is_unchanged_without_a_soft_delete_column(self):
        self._resolve(NO_SOFT_DELETE_COLUMNS, list(TWO_ROWS), [])
        self.assertEqual(
            self._only("SELECT"),
            "SELECT `Id`, `IsoCode`, `StartDate` FROM `DatedConversionRate` "
            "WHERE ((`IsoCode` <=> 'JPY' AND `StartDate` <=> '2026-04-01') "
            "OR (`IsoCode` <=> 'EUR' AND `StartDate` <=> '2026-01-01'))",
        )

    # --- match resolution: what it does with the answer ---------------------------------

    def test_a_surviving_live_row_lends_its_id_to_the_seeded_row(self):
        """Excluding deleted rows is only worth anything if the live one is still adopted.

        The match value carries a space, as every shipped record's does, so this also pins
        that rows are split on tabs -- `split()` would discard all of them and quietly stop
        adopting anything.
        """
        rows = [_named_row("CNY 1900-01-01", 6.6666, "seed-derived-id")]
        self._resolve(NAME_MATCH_COLUMNS, rows, ["native-id-1\tCNY 1900-01-01\n"])
        self.assertEqual(rows[0]["id"], "native-id-1")

    def test_two_live_rows_sharing_a_match_key_abort_the_phase(self):
        """The abort named in this module's first paragraph, exercised rather than described."""
        rows = [_named_row("CNY 1900-01-01", 6.6666, "seed-derived-id")]
        originals = self._stub(
            NAME_MATCH_COLUMNS,
            ["native-id-1\tCNY 1900-01-01\nnative-id-2\tCNY 1900-01-01\n"],
        )
        try:
            with self.assertRaises(self.applier.PackReferenceDataError) as raised:
                self.applier.pre_resolve_existing_ids(object(), "DatedConversionRate", rows)
        finally:
            self._restore(originals)
        self.assertIn("duplicate existing DatedConversionRate match", str(raised.exception))
        self.assertEqual(rows[0]["id"], "seed-derived-id")

    # --- post-upsert verification -------------------------------------------------------

    def test_verification_counts_only_live_rows(self):
        """The upsert leaves a soft-deleted row deleted, so the count must not include it."""
        self._apply(CANONICAL_COLUMNS, list(TWO_ROWS), ["2"])
        self.assertEqual(
            self._only("SELECT COUNT(*)"),
            "SELECT COUNT(*) FROM `DatedConversionRate` "
            "WHERE `Id` IN ('seed-jpy', 'seed-eur') AND COALESCE(`IsDeleted`, 0) = 0",
        )

    def test_the_backstop_says_the_count_was_live_only(self):
        """"live rows" is the operator's signal that the shortfall is not a missing insert."""
        originals = self._stub(CANONICAL_COLUMNS, ["", "1"])
        try:
            with self.assertRaises(self.applier.PackReferenceDataError) as raised:
                self.applier.apply_rows(
                    object(), "DatedConversionRate", list(TWO_ROWS),
                    "DatedConversionRate", CANONICAL_COLUMNS,
                )
        finally:
            self._restore(originals)
        self.assertEqual(str(raised.exception),
                         "DatedConversionRate verification found 1/2 live rows")

    def test_verification_is_unchanged_without_a_soft_delete_column(self):
        self._apply(NO_SOFT_DELETE_COLUMNS, list(TWO_ROWS), ["2"])
        self.assertEqual(
            self._only("SELECT COUNT(*)"),
            "SELECT COUNT(*) FROM `DatedConversionRate` WHERE `Id` IN ('seed-jpy', 'seed-eur')",
        )

    # --- preflight: refuse before anything is written ------------------------------------

    def test_preflight_refuses_when_a_seeded_match_key_sits_on_a_deleted_row(self):
        """The gate has to be before the write, not after it.

        `apply_rows` sends its batches through the mysql CLI with no enclosing transaction,
        so a post-upsert count notices the shortfall only once the write has committed --
        and once earlier objects in the layer have committed too.
        """
        originals = self._stub(CANONICAL_COLUMNS, ["native-id-1\nnative-id-2\n"])
        try:
            with self.assertRaises(self.applier.PackReferenceDataError) as raised:
                self.applier.preflight_database(
                    object(), {"baseline_records": list(TWO_ROWS)}, ["DatedConversionRate"]
                )
        finally:
            self._restore(originals)
        message = str(raised.exception)
        self.assertIn("2 soft-deleted rows the seed would land on", message)
        self.assertIn("native-id-1", message)
        self.assertIn("restore them", message)
        self.assertNotIn("INSERT", " ".join(self.captured))

    def test_preflight_probes_both_match_keys_and_seed_ids(self):
        """Neither route contains the other, so the gate has to search both.

        By match key: a row that pre-dates the seed holds a native id, so the insert collides
        on a natural unique key and no row carrying the seed id exists at all. By seed id: the
        ids derive from `key`, not from `match`, so an operator who edits an updateable match
        field and then deletes the row leaves it carrying the seed id while its match no
        longer matches. The ids below are `stable_uuid("DatedConversionRate", key)`.
        """
        originals = self._stub(CANONICAL_COLUMNS, [""])
        try:
            self.applier.preflight_database(
                object(), {"baseline_records": list(TWO_ROWS)}, ["DatedConversionRate"]
            )
        finally:
            self._restore(originals)
        self.assertEqual(
            self._only("SELECT `Id`"),
            "SELECT `Id` FROM `DatedConversionRate` "
            "WHERE ((`IsoCode` <=> 'JPY' AND `StartDate` <=> '2026-04-01') "
            "OR (`IsoCode` <=> 'EUR' AND `StartDate` <=> '2026-01-01') "
            "OR `Id` IN ('b79e5632-0e1e-5a3c-8fb9-4529929e1ce6', "
            "'d7e6c372-3ae7-578c-851a-7ca9f2c1e862')) "
            "AND COALESCE(`IsDeleted`, 0) = 1",
        )

    def test_preflight_refuses_a_match_value_it_cannot_resolve(self):
        """Match values are materialized after preflight; none carry templates today.

        A probe that compared the literal `{{ ... }}` would under-report rather than fail,
        which is the wrong direction for a gate.
        """
        row = dict(TWO_ROWS[0], match={"IsoCode": "{{ record.x.id }}", "StartDate": "2026-04-01"})
        originals = self._stub(CANONICAL_COLUMNS, [""])
        try:
            with self.assertRaises(self.applier.PackReferenceDataError) as raised:
                self.applier.preflight_database(
                    object(), {"baseline_records": [row]}, ["DatedConversionRate"]
                )
        finally:
            self._restore(originals)
        self.assertIn("uses a template in its match", str(raised.exception))

    def test_preflight_passes_when_no_seeded_key_is_soft_deleted(self):
        originals = self._stub(CANONICAL_COLUMNS, [""])
        try:
            self.applier.preflight_database(
                object(), {"baseline_records": list(TWO_ROWS)}, ["DatedConversionRate"]
            )
        finally:
            self._restore(originals)

    def test_preflight_issues_no_probe_without_a_soft_delete_column(self):
        originals = self._stub(NO_SOFT_DELETE_COLUMNS, [])
        try:
            self.applier.preflight_database(
                object(), {"baseline_records": list(TWO_ROWS)}, ["DatedConversionRate"]
            )
        finally:
            self._restore(originals)
        self.assertEqual([s for s in self.captured if s.startswith("SELECT `Id`")], [])

    # --- seeding must not resurrect -----------------------------------------------------

    def test_the_upsert_does_not_revive_a_soft_deleted_row(self):
        """Deleting a seeded row through the UI is how an operator retires one.

        Adding IsDeleted to the update set would undo that on the next deployment, silently,
        for every seeded object -- including `Industry_Vertical__c`, whose rows gate saves
        through runtime-scoped picklist validation.
        """
        self._apply(CANONICAL_COLUMNS, [TWO_ROWS[0]], ["1"])
        self.assertEqual(
            self._only("INSERT"),
            "INSERT INTO `DatedConversionRate` "
            "(`Id`, `IsoCode`, `StartDate`, `ConversionRate`, `IsDeleted`) "
            "VALUES ('seed-jpy', 'JPY', '2026-04-01', 159.651, 0) "
            "ON DUPLICATE KEY UPDATE "
            "`ConversionRate` = VALUES(`ConversionRate`), `IsoCode` = VALUES(`IsoCode`), "
            "`StartDate` = VALUES(`StartDate`)",
        )

    def test_a_source_declaring_IsDeleted_still_cannot_revive_a_row(self):
        """The rule belongs to the applier, not to the current contents of five JSON files.

        A reference-data source is free to put any column in `fields`; without IsDeleted in
        NEVER_UPDATE_COLUMNS, one line in a source file restores resurrection for every row
        of that object.
        """
        row = dict(TWO_ROWS[0], fields={"ConversionRate": 159.651, "IsDeleted": False})
        self._apply(CANONICAL_COLUMNS, [row], ["1"])
        statement = self._only("INSERT")
        self.assertIn("`IsDeleted`", statement.split("ON DUPLICATE KEY UPDATE")[0])
        self.assertNotIn("IsDeleted", statement.split("ON DUPLICATE KEY UPDATE")[1])

    # --- the column name is schema-derived, not a literal --------------------------------

    def test_every_soft_delete_predicate_takes_its_column_name_from_the_schema(self):
        """`fetch_columns` reports the physical name; all three predicates must use it.

        A hardcoded `IsDeleted` is indistinguishable from the resolved name against today's
        schema, so this fixture declares a different physical spelling. Note the update-set
        filter is *not* covered here: `NEVER_UPDATE_COLUMNS` matches physical names as
        literals, exactly as it already does for `Id`, `CreatedDate` and `CreatedById`, so
        this one rule follows its three siblings rather than the predicates.
        """
        columns = dict(CANONICAL_COLUMNS, isdeleted="is_deleted_flag")

        self._resolve(columns, list(TWO_ROWS), [])
        self.assertIn("AND COALESCE(`is_deleted_flag`, 0) = 0", self._only("SELECT"))

        originals = self._stub(columns, [""])
        try:
            self.applier.preflight_database(
                object(), {"baseline_records": list(TWO_ROWS)}, ["DatedConversionRate"]
            )
        finally:
            self._restore(originals)
        self.assertIn("AND COALESCE(`is_deleted_flag`, 0) = 1", self._only("SELECT `Id`"))

        self._apply(columns, list(TWO_ROWS), ["2"])
        self.assertIn("AND COALESCE(`is_deleted_flag`, 0) = 0", self._only("SELECT COUNT(*)"))


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