gendesign/backend/tests/services/scrapers/test_domrf_kn_upsert_sql.py

"""#2464 cluster E + F — UPSERT DO UPDATE SET column-omission regression tests.

Both UPSERT_OBJECT_SQL and UPSERT_PHOTO_SQL had columns populated on every
INSERT (fresh scrape values, see _norm_object / catalog scraper) but silently
omitted from `ON CONFLICT ... DO UPDATE SET`. Re-running the scraper for an
already-seen (obj_id, snapshot_date) / (obj_id, obj_file_id) row left those
columns stale forever (INSERT wins once, DO UPDATE never refreshes them).

SQL-string assertions against the module-level `text(...)` constants — no live
DB needed; the INSERT column list is the source of truth for "must this column
also be kept fresh on conflict", so each test also cross-checks against the
INSERT list to prevent the same class of omission from creeping back in for a
newly-added column.
"""

from __future__ import annotations

import re

from app.services.scrapers.domrf_kn import UPSERT_OBJECT_SQL, UPSERT_PHOTO_SQL

_OBJECT_PREVIOUSLY_OMITTED = [
    "problem_flag",
    "green_house",
    "floor_min",
    "floor_max",
    "hobj_id",
    "short_addr",
    "dev_inn",
    "region_cd",
]


def _insert_columns(sql: str, table: str) -> list[str]:
    """Extract the `INSERT INTO <table> (col1, col2, ...)` column list."""
    m = re.search(rf"INSERT INTO {table}\s*\(([^)]+)\)", sql)
    assert m, f"could not locate INSERT INTO {table} (...) column list"
    return [c.strip() for c in m.group(1).split(",")]


def _do_update_set_columns(sql: str) -> set[str]:
    """Extract every `col = ` LHS from the `DO UPDATE SET` clause."""
    m = re.search(r"DO UPDATE SET(.*)", sql, re.DOTALL)
    assert m, "could not locate ON CONFLICT ... DO UPDATE SET clause"
    body = m.group(1)
    return set(re.findall(r"(?:^|,)\s*(\w+)\s*=", body, re.MULTILINE))


class TestUpsertObjectSqlDoUpdateSet:
    def test_previously_omitted_columns_now_present(self) -> None:
        sql = str(UPSERT_OBJECT_SQL)
        updated = _do_update_set_columns(sql)
        missing = [c for c in _OBJECT_PREVIOUSLY_OMITTED if c not in updated]
        assert not missing, f"still missing from DO UPDATE SET: {missing}"

    def test_previously_omitted_columns_use_direct_excluded_not_coalesce(self) -> None:
        # These 8 are freshly computed every scrape call (per _norm_object) --
        # COALESCE would permanently pin a stale value once ever set (most
        # visibly wrong for problem_flag: a resolved problem must be able to
        # clear back to NULL, not get stuck "still problematic" forever).
        sql = " ".join(str(UPSERT_OBJECT_SQL).split())
        for col in _OBJECT_PREVIOUSLY_OMITTED:
            assert (
                f"{col} = EXCLUDED.{col}" in sql
            ), f"{col} should be direct `= EXCLUDED.{col}` (not COALESCE):\n{sql}"
            # guard against a COALESCE(EXCLUDED.col, ...) formulation, which would
            # permanently pin a stale value once ever set instead of refreshing it.
            assert (
                f"COALESCE(EXCLUDED.{col}," not in sql
            ), f"{col} must not be wrapped in COALESCE (needs to be able to clear to NULL)"

    def test_all_insert_columns_covered_by_do_update_set_or_conflict_target(self) -> None:
        # Locks in the class of bug (INSERT populates a column, DO UPDATE SET
        # silently drops it) so it can't recur for a column added later.
        sql = str(UPSERT_OBJECT_SQL)
        insert_cols = set(_insert_columns(sql, "domrf_kn_objects"))
        updated_cols = _do_update_set_columns(sql)
        conflict_target = {"obj_id", "snapshot_date"}  # immutable conflict key, not updated
        missing = insert_cols - updated_cols - conflict_target
        assert not missing, f"INSERT columns missing from DO UPDATE SET: {sorted(missing)}"

    def test_no_double_colon_cast(self) -> None:
        sql = str(UPSERT_OBJECT_SQL)
        assert re.search(r":[a-z_]+::[a-z]", sql) is None


class TestUpsertPhotoSqlDoUpdateSet:
    def test_build_type_now_present(self) -> None:
        sql = str(UPSERT_PHOTO_SQL)
        updated = _do_update_set_columns(sql)
        assert "build_type" in updated

    def test_build_type_uses_direct_excluded(self) -> None:
        sql = " ".join(str(UPSERT_PHOTO_SQL).split())
        assert "build_type = EXCLUDED.build_type" in sql

    def test_all_insert_columns_covered_by_do_update_set_or_conflict_target(self) -> None:
        sql = str(UPSERT_PHOTO_SQL)
        insert_cols = set(_insert_columns(sql, "domrf_kn_photos"))
        updated_cols = _do_update_set_columns(sql)
        conflict_target = {"obj_id", "obj_file_id"}
        missing = insert_cols - updated_cols - conflict_target
        assert not missing, f"INSERT columns missing from DO UPDATE SET: {sorted(missing)}"

    def test_no_double_colon_cast(self) -> None:
        sql = str(UPSERT_PHOTO_SQL)
        assert re.search(r":[a-z_]+::[a-z]", sql) is None