query-payslip-default/tests/test_data_management_contract.py

# -*- coding: utf-8 -*-
"""通用数据管理契约测试（demo_items fixture，不含具体业务技能内容）。"""
from __future__ import annotations

import json
import sqlite3
import unittest

from _support import IsolatedDataRoot

DEMO_ITEMS_DDL = """
CREATE TABLE IF NOT EXISTS demo_items (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    title TEXT NOT NULL,
    description TEXT,
    status TEXT NOT NULL DEFAULT 'draft',
    generated_value TEXT,
    created_at INTEGER NOT NULL DEFAULT (unixepoch()),
    updated_at INTEGER NOT NULL DEFAULT (unixepoch())
);
CREATE TRIGGER IF NOT EXISTS demo_items_set_updated_at
AFTER UPDATE ON demo_items
FOR EACH ROW
BEGIN
    UPDATE demo_items
    SET updated_at = unixepoch()
    WHERE id = NEW.id AND updated_at = OLD.updated_at;
END;
"""

DEMO_ITEMS_TABLE_METADATA = {
    "table_name": "demo_items",
    "display_name": "示例记录",
    "description": "通用数据管理契约演示表",
    "visible": 1,
    "readonly": 0,
}

DEMO_ITEMS_COLUMN_METADATA: tuple[dict[str, object], ...] = (
    {"column_name": "id", "display_name": "编号", "visible": 1, "searchable": 1, "editable": 0},
    {"column_name": "title", "display_name": "标题", "visible": 1, "searchable": 1, "editable": 1},
    {"column_name": "description", "display_name": "描述", "visible": 1, "searchable": 1, "editable": 1},
    {
        "column_name": "status",
        "display_name": "状态",
        "visible": 1,
        "searchable": 1,
        "editable": 1,
        "options_json": json.dumps(
            [
                {"value": "draft", "label": "草稿"},
                {"value": "published", "label": "已发布"},
                {"value": "archived", "label": "已归档"},
            ],
            ensure_ascii=False,
        ),
    },
    {"column_name": "generated_value", "display_name": "生成值", "visible": 1, "searchable": 0, "editable": 0},
    {
        "column_name": "created_at",
        "display_name": "创建时间",
        "visible": 1,
        "searchable": 0,
        "editable": 0,
        "display_type": "datetime_unix_seconds",
    },
    {
        "column_name": "updated_at",
        "display_name": "更新时间",
        "visible": 1,
        "searchable": 0,
        "editable": 0,
        "display_type": "datetime_unix_seconds",
    },
)


def _init_demo_items_fixture(conn: sqlite3.Connection, *, prune: bool = True) -> None:
    from db.display_metadata import (
        init_display_metadata_tables,
        list_physical_column_names,
        seed_column_metadata_batch,
        upsert_table_metadata,
    )

    cur = conn.cursor()
    init_display_metadata_tables(cur)
    cur.executescript(DEMO_ITEMS_DDL)
    meta = DEMO_ITEMS_TABLE_METADATA
    upsert_table_metadata(
        cur,
        table_name=str(meta["table_name"]),
        display_name=str(meta["display_name"]),
        description=str(meta["description"]),
        visible=int(meta["visible"]),
        readonly=int(meta["readonly"]),
    )
    physical = list_physical_column_names(cur, "demo_items")
    seed_column_metadata_batch(
        cur,
        table_name="demo_items",
        columns=DEMO_ITEMS_COLUMN_METADATA,
        valid_column_names=physical if prune else None,
    )
    conn.commit()


class TestDataManagementContract(unittest.TestCase):
    def test_demo_items_metadata_permissions(self) -> None:
        conn = sqlite3.connect(":memory:")
        try:
            _init_demo_items_fixture(conn)
            from db.display_metadata import (
                get_column_editable,
                get_table_readonly,
                list_editable_business_columns,
                table_allows_manual_create,
            )

            cur = conn.cursor()
            self.assertEqual(get_table_readonly(cur, "demo_items"), 0)
            self.assertEqual(get_column_editable(cur, "demo_items", "id"), 0)
            self.assertEqual(get_column_editable(cur, "demo_items", "title"), 1)
            self.assertEqual(get_column_editable(cur, "demo_items", "generated_value"), 0)
            self.assertEqual(get_column_editable(cur, "demo_items", "created_at"), 0)
            self.assertEqual(get_column_editable(cur, "demo_items", "updated_at"), 0)

            editable_cols = list_editable_business_columns(cur, "demo_items")
            self.assertEqual(editable_cols, ["title", "description", "status"])
            self.assertTrue(table_allows_manual_create(cur, "demo_items"))
        finally:
            conn.close()

    def test_demo_items_init_idempotent_no_duplicates(self) -> None:
        conn = sqlite3.connect(":memory:")
        try:
            _init_demo_items_fixture(conn)
            _init_demo_items_fixture(conn)
            cur = conn.cursor()
            cur.execute("SELECT COUNT(*) FROM _jiangchang_tables WHERE table_name = 'demo_items'")
            self.assertEqual(int(cur.fetchone()[0]), 1)
            cur.execute("SELECT COUNT(*) FROM _jiangchang_columns WHERE table_name = 'demo_items'")
            self.assertEqual(int(cur.fetchone()[0]), 7)
        finally:
            conn.close()

    def test_reinit_updates_stale_editable_value(self) -> None:
        conn = sqlite3.connect(":memory:")
        try:
            _init_demo_items_fixture(conn)
            conn.execute(
                "UPDATE _jiangchang_columns SET editable = 0 WHERE table_name = 'demo_items' AND column_name = 'title'"
            )
            conn.commit()
            _init_demo_items_fixture(conn)
            from db.display_metadata import get_column_editable

            self.assertEqual(get_column_editable(conn.cursor(), "demo_items", "title"), 1)
        finally:
            conn.close()

    def test_missing_table_metadata_detected(self) -> None:
        conn = sqlite3.connect(":memory:")
        try:
            _init_demo_items_fixture(conn)
            conn.execute("DELETE FROM _jiangchang_tables WHERE table_name = 'demo_items'")
            conn.commit()
            from db.display_metadata_validator import validate_display_metadata

            report = validate_display_metadata(conn)
            self.assertFalse(report.ok)
            self.assertTrue(any("demo_items" in err for err in report.errors))
        finally:
            conn.close()

    def test_missing_column_metadata_detected(self) -> None:
        conn = sqlite3.connect(":memory:")
        try:
            _init_demo_items_fixture(conn)
            conn.execute(
                "DELETE FROM _jiangchang_columns WHERE table_name = 'demo_items' AND column_name = 'title'"
            )
            conn.commit()
            from db.display_metadata_validator import validate_display_metadata

            report = validate_display_metadata(conn)
            self.assertFalse(report.ok)
            self.assertTrue(any("demo_items.title" in err for err in report.errors))
        finally:
            conn.close()

    def test_metadata_reference_to_missing_column_detected(self) -> None:
        conn = sqlite3.connect(":memory:")
        try:
            _init_demo_items_fixture(conn)
            conn.execute(
                """
                INSERT INTO _jiangchang_columns
                (table_name, column_name, display_name, visible, searchable, editable)
                VALUES ('demo_items', 'not_exists', '不存在', 1, 1, 0)
                """
            )
            conn.commit()
            from db.display_metadata_validator import validate_display_metadata

            report = validate_display_metadata(conn)
            self.assertFalse(report.ok)
            self.assertTrue(any("not_exists" in err for err in report.errors))
        finally:
            conn.close()

    def test_empty_display_name_detected_for_visible_column(self) -> None:
        conn = sqlite3.connect(":memory:")
        try:
            _init_demo_items_fixture(conn)
            conn.execute(
                "UPDATE _jiangchang_columns SET display_name = '' WHERE table_name = 'demo_items' AND column_name = 'title'"
            )
            conn.commit()
            from db.display_metadata_validator import validate_display_metadata

            report = validate_display_metadata(conn)
            self.assertFalse(report.ok)
            self.assertTrue(any("display_name 为空" in err for err in report.errors))
        finally:
            conn.close()

    def test_visible_columns_use_chinese_display_name(self) -> None:
        conn = sqlite3.connect(":memory:")
        try:
            _init_demo_items_fixture(conn)
            cur = conn.cursor()
            cur.execute(
                """
                SELECT column_name, display_name
                FROM _jiangchang_columns
                WHERE table_name = 'demo_items' AND visible = 1
                """
            )
            for column_name, display_name in cur.fetchall():
                self.assertTrue(
                    any("\u4e00" <= ch <= "\u9fff" for ch in str(display_name)),
                    msg=f"{column_name} display_name should be Chinese: {display_name!r}",
                )
        finally:
            conn.close()

    def test_timestamp_fields_type_default_and_editable(self) -> None:
        conn = sqlite3.connect(":memory:")
        try:
            _init_demo_items_fixture(conn)
            cur = conn.cursor()
            for column_name in ("created_at", "updated_at"):
                cur.execute(f"PRAGMA table_info('demo_items')")
                info = {row[1]: row for row in cur.fetchall()}
                row = info[column_name]
                self.assertIn("INT", str(row[2]).upper())
                self.assertIn("unixepoch", str(row[4]).lower())
                cur.execute(
                    """
                    SELECT editable, display_type FROM _jiangchang_columns
                    WHERE table_name = 'demo_items' AND column_name = ?
                    """,
                    (column_name,),
                )
                editable, display_type = cur.fetchone()
                self.assertEqual(int(editable), 0)
                self.assertEqual(display_type, "datetime_unix_seconds")
        finally:
            conn.close()

    def test_physical_order_matches_pragma_cid(self) -> None:
        conn = sqlite3.connect(":memory:")
        try:
            _init_demo_items_fixture(conn)
            from db.display_metadata import list_physical_column_names

            cur = conn.cursor()
            expected = [str(item["column_name"]) for item in DEMO_ITEMS_COLUMN_METADATA]
            physical = list_physical_column_names(cur, "demo_items")
            self.assertEqual(physical, expected)
            cur.execute("PRAGMA table_info('demo_items')")
            cid_order = [row[1] for row in sorted(cur.fetchall(), key=lambda row: int(row[0]))]
            self.assertEqual(cid_order, expected)
        finally:
            conn.close()

    def test_options_json_parses_for_enum_field(self) -> None:
        conn = sqlite3.connect(":memory:")
        try:
            _init_demo_items_fixture(conn)
            cur = conn.cursor()
            cur.execute(
                """
                SELECT options_json FROM _jiangchang_columns
                WHERE table_name = 'demo_items' AND column_name = 'status'
                """
            )
            raw = cur.fetchone()[0]
            from db.display_metadata_validator import validate_options_json

            errors, _ = validate_options_json(raw)
            self.assertEqual(errors, [])
            parsed = json.loads(raw)
            self.assertEqual(len(parsed), 3)
            self.assertEqual(parsed[0]["value"], "draft")
        finally:
            conn.close()

    def test_system_table_no_editable_fields_not_manual_create(self) -> None:
        conn = sqlite3.connect(":memory:")
        try:
            from db.display_metadata import (
                init_display_metadata_tables,
                seed_column_metadata_batch,
                table_allows_manual_create,
                upsert_table_metadata,
            )

            cur = conn.cursor()
            init_display_metadata_tables(cur)
            cur.execute(
                """
                CREATE TABLE source_records (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    item_id TEXT NOT NULL,
                    synced_at INTEGER NOT NULL DEFAULT (unixepoch())
                )
                """
            )
            upsert_table_metadata(
                cur,
                table_name="source_records",
                display_name="来源记录",
                description="只由同步流程写入",
                visible=1,
                readonly=0,
            )
            seed_column_metadata_batch(
                cur,
                table_name="source_records",
                columns=(
                    {"column_name": "id", "display_name": "编号", "editable": 0},
                    {"column_name": "item_id", "display_name": "记录编号", "editable": 0},
                    {"column_name": "synced_at", "display_name": "同步时间", "editable": 0, "display_type": "datetime_unix_seconds"},
                ),
            )
            conn.commit()
            self.assertFalse(table_allows_manual_create(cur, "source_records"))
        finally:
            conn.close()

    def test_table_readonly_blocks_manual_create_even_with_editable_fields(self) -> None:
        conn = sqlite3.connect(":memory:")
        try:
            _init_demo_items_fixture(conn)
            conn.execute("UPDATE _jiangchang_tables SET readonly = 1 WHERE table_name = 'demo_items'")
            conn.commit()
            from db.display_metadata import table_allows_manual_create

            self.assertFalse(table_allows_manual_create(conn.cursor(), "demo_items"))
        finally:
            conn.close()

    def test_prune_stale_column_metadata(self) -> None:
        conn = sqlite3.connect(":memory:")
        try:
            _init_demo_items_fixture(conn)
            conn.execute(
                """
                INSERT INTO _jiangchang_columns
                (table_name, column_name, display_name, visible, searchable, editable)
                VALUES ('demo_items', 'removed_field', '已删除', 0, 0, 0)
                """
            )
            conn.commit()
            from db.display_metadata import list_physical_column_names, seed_column_metadata_batch

            cur = conn.cursor()
            physical = list_physical_column_names(cur, "demo_items")
            seed_column_metadata_batch(
                cur,
                table_name="demo_items",
                columns=DEMO_ITEMS_COLUMN_METADATA,
                valid_column_names=physical,
            )
            conn.commit()
            cur.execute(
                "SELECT 1 FROM _jiangchang_columns WHERE table_name = 'demo_items' AND column_name = 'removed_field'"
            )
            self.assertIsNone(cur.fetchone())
        finally:
            conn.close()

    def test_seed_batch_without_valid_column_names_does_not_prune(self) -> None:
        conn = sqlite3.connect(":memory:")
        try:
            _init_demo_items_fixture(conn, prune=False)
            conn.execute(
                """
                INSERT INTO _jiangchang_columns
                (table_name, column_name, display_name, visible, searchable, editable)
                VALUES ('demo_items', 'removed_field', '已删除', 0, 0, 0)
                """
            )
            conn.commit()
            from db.display_metadata import seed_column_metadata_batch

            seed_column_metadata_batch(
                conn.cursor(),
                table_name="demo_items",
                columns=DEMO_ITEMS_COLUMN_METADATA,
            )
            conn.commit()
            cur = conn.cursor()
            cur.execute(
                "SELECT 1 FROM _jiangchang_columns WHERE table_name = 'demo_items' AND column_name = 'removed_field'"
            )
            self.assertIsNotNone(cur.fetchone())
        finally:
            conn.close()

    def test_seed_batch_accepts_list_columns(self) -> None:
        conn = sqlite3.connect(":memory:")
        try:
            _init_demo_items_fixture(conn, prune=False)
            from db.display_metadata import get_column_editable, seed_column_metadata_batch

            columns = list(DEMO_ITEMS_COLUMN_METADATA)
            columns[1] = {**columns[1], "display_name": "标题（列表）"}
            seed_column_metadata_batch(
                conn.cursor(),
                table_name="demo_items",
                columns=columns,
                valid_column_names=[str(item["column_name"]) for item in DEMO_ITEMS_COLUMN_METADATA],
            )
            conn.commit()
            cur = conn.cursor()
            cur.execute(
                "SELECT display_name FROM _jiangchang_columns WHERE table_name = 'demo_items' AND column_name = 'title'"
            )
            self.assertEqual(cur.fetchone()[0], "标题（列表）")
            self.assertEqual(get_column_editable(cur, "demo_items", "title"), 1)
        finally:
            conn.close()

    def test_prune_does_not_remove_existing_column_metadata(self) -> None:
        conn = sqlite3.connect(":memory:")
        try:
            _init_demo_items_fixture(conn)
            from db.display_metadata import get_column_editable, seed_column_metadata_batch

            cur = conn.cursor()
            before = get_column_editable(cur, "demo_items", "title")
            seed_column_metadata_batch(
                cur,
                table_name="demo_items",
                columns=DEMO_ITEMS_COLUMN_METADATA,
                valid_column_names=[str(item["column_name"]) for item in DEMO_ITEMS_COLUMN_METADATA],
            )
            conn.commit()
            self.assertEqual(get_column_editable(cur, "demo_items", "title"), before)
        finally:
            conn.close()

    def test_demo_items_validation_passes(self) -> None:
        conn = sqlite3.connect(":memory:")
        try:
            _init_demo_items_fixture(conn)
            from db.display_metadata_validator import validate_display_metadata

            report = validate_display_metadata(conn)
            self.assertTrue(report.ok, msg="; ".join(report.errors))
        finally:
            conn.close()


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