feat: add Chinese display metadata for data management UI

tests/test_display_metadata.py

@@ -0,0 +1,263 @@
+# -*- coding: utf-8 -*-
+"""数据管理展示元数据：幂等初始化、中文名称、敏感字段权限。"""
+from __future__ import annotations
+
+import re
+
+import pytest
+
+from db.display_metadata import (
+    ACCOUNT_LEASES_TABLE,
+    ACCOUNTS_TABLE,
+    CREDENTIALS_TABLE,
+    PLATFORMS_TABLE,
+    USER_BUSINESS_TABLES,
+    get_table_readonly,
+    init_display_metadata_tables,
+    list_physical_column_names,
+    table_allows_manual_create,
+    upsert_column_metadata,
+    upsert_table_metadata,
+)
+
+
+def _has_cjk(text: str) -> bool:
+    return bool(re.search(r"[\u4e00-\u9fff]", text or ""))
+
+
+@pytest.fixture
+def metadata_db(clean_db):
+    """clean_db already ran init_db(); return connection for metadata assertions."""
+    yield clean_db
+
+
+def test_init_db_metadata_idempotent(metadata_db):
+    from db.connection import get_conn, init_db
+
+    init_db()
+    conn = get_conn()
+    try:
+        cur = conn.cursor()
+        cur.execute("SELECT COUNT(*) FROM _jiangchang_tables")
+        count_after_second = int(cur.fetchone()[0])
+        assert count_after_second == len(USER_BUSINESS_TABLES)
+    finally:
+        conn.close()
+
+
+@pytest.mark.parametrize(
+    "table_name,expected_display",
+    [
+        (PLATFORMS_TABLE, "平台信息"),
+        (ACCOUNTS_TABLE, "账号信息"),
+        (CREDENTIALS_TABLE, "凭证信息"),
+        (ACCOUNT_LEASES_TABLE, "账号租约"),
+    ],
+)
+def test_table_chinese_display_names(metadata_db, table_name, expected_display):
+    cur = metadata_db.cursor()
+    cur.execute(
+        "SELECT display_name FROM _jiangchang_tables WHERE table_name = ?",
+        (table_name,),
+    )
+    row = cur.fetchone()
+    assert row is not None
+    assert row[0] == expected_display
+
+
+def test_all_business_columns_have_chinese_display_names(metadata_db):
+    cur = metadata_db.cursor()
+    for table_name in USER_BUSINESS_TABLES:
+        physical = set(list_physical_column_names(cur, table_name))
+        cur.execute(
+            """
+            SELECT column_name, display_name
+            FROM _jiangchang_columns
+            WHERE table_name = ?
+            """,
+            (table_name,),
+        )
+        meta_rows = cur.fetchall()
+        assert len(meta_rows) == len(physical), f"{table_name} 元数据字段数与物理表不一致"
+        for column_name, display_name in meta_rows:
+            assert column_name in physical
+            assert _has_cjk(str(display_name)), f"{table_name}.{column_name} 缺少中文 display_name"
+
+
+def test_sensitive_credential_fields_not_editable(metadata_db):
+    cur = metadata_db.cursor()
+    sensitive = ("secret_ref", "secret_mask", "secret_ciphertext", "secret_storage")
+    for column_name in sensitive:
+        cur.execute(
+            """
+            SELECT editable, searchable
+            FROM _jiangchang_columns
+            WHERE table_name = ? AND column_name = ?
+            """,
+            (CREDENTIALS_TABLE, column_name),
+        )
+        editable, searchable = cur.fetchone()
+        assert int(editable) == 0, column_name
+        if column_name in ("secret_ref", "secret_ciphertext"):
+            assert int(searchable) == 0, column_name
+
+
+def test_lease_token_not_editable(metadata_db):
+    cur = metadata_db.cursor()
+    cur.execute(
+        """
+        SELECT editable FROM _jiangchang_columns
+        WHERE table_name = ? AND column_name = 'lease_token'
+        """,
+        (ACCOUNT_LEASES_TABLE,),
+    )
+    assert int(cur.fetchone()[0]) == 0
+
+
+def test_account_leases_readonly(metadata_db):
+    cur = metadata_db.cursor()
+    assert int(get_table_readonly(cur, ACCOUNT_LEASES_TABLE)) == 1
+    assert table_allows_manual_create(cur, ACCOUNT_LEASES_TABLE) is False
+
+
+def test_platforms_key_not_editable(metadata_db):
+    cur = metadata_db.cursor()
+    cur.execute(
+        """
+        SELECT editable FROM _jiangchang_columns
+        WHERE table_name = ? AND column_name = 'platform_key'
+        """,
+        (PLATFORMS_TABLE,),
+    )
+    assert int(cur.fetchone()[0]) == 0
+
+
+def test_profile_dir_not_editable(metadata_db):
+    cur = metadata_db.cursor()
+    cur.execute(
+        """
+        SELECT editable FROM _jiangchang_columns
+        WHERE table_name = ? AND column_name = 'profile_dir'
+        """,
+        (ACCOUNTS_TABLE,),
+    )
+    assert int(cur.fetchone()[0]) == 0
+
+
+def test_timestamp_columns_use_datetime_display_type(metadata_db):
+    cur = metadata_db.cursor()
+    for table_name in USER_BUSINESS_TABLES:
+        cur.execute(
+            """
+            SELECT column_name, display_type, editable
+            FROM _jiangchang_columns
+            WHERE table_name = ? AND column_name IN ('created_at', 'updated_at')
+            """,
+            (table_name,),
+        )
+        for column_name, display_type, editable in cur.fetchall():
+            assert display_type == "datetime_unix_seconds", f"{table_name}.{column_name}"
+            assert int(editable) == 0, f"{table_name}.{column_name}"
+
+
+def test_metadata_tables_hidden_from_user_tables(metadata_db):
+    from db.display_metadata import iter_user_tables
+
+    tables = list(iter_user_tables(metadata_db.cursor()))
+    assert "_jiangchang_tables" not in tables
+    assert "_jiangchang_columns" not in tables
+    assert "_schema_meta" not in tables
+    for name in USER_BUSINESS_TABLES:
+        assert name in tables
+
+
+def test_reinit_clears_stale_column_metadata(metadata_db):
+    from db.connection import get_conn, init_db
+
+    conn = get_conn()
+    try:
+        conn.execute(
+            """
+            INSERT INTO _jiangchang_columns
+            (table_name, column_name, display_name, visible, searchable, editable)
+            VALUES ('accounts', 'stale_field', '陈旧字段', 1, 1, 0)
+            """
+        )
+        conn.commit()
+    finally:
+        conn.close()
+
+    init_db()
+    conn = get_conn()
+    try:
+        cur = conn.cursor()
+        cur.execute(
+            """
+            SELECT COUNT(*) FROM _jiangchang_columns
+            WHERE table_name = 'accounts' AND column_name = 'stale_field'
+            """
+        )
+        assert int(cur.fetchone()[0]) == 0
+    finally:
+        conn.close()
+
+
+def test_missing_column_metadata_detected_in_memory_db():
+    import sqlite3
+
+    from db.display_metadata import init_display_metadata, list_physical_column_names
+
+    conn = sqlite3.connect(":memory:")
+    try:
+        conn.executescript(
+            """
+            CREATE TABLE platforms (
+                platform_key TEXT PRIMARY KEY,
+                display_name TEXT NOT NULL
+            );
+            """
+        )
+        init_display_metadata_tables(conn.cursor())
+        upsert_table_metadata(
+            conn.cursor(),
+            table_name="platforms",
+            display_name="平台信息",
+        )
+        upsert_column_metadata(
+            conn.cursor(),
+            table_name="platforms",
+            column_name="platform_key",
+            display_name="平台键",
+            editable=0,
+        )
+        conn.commit()
+
+        physical = list_physical_column_names(conn.cursor(), "platforms")
+        assert "display_name" in physical
+        cur = conn.cursor()
+        cur.execute(
+            "SELECT COUNT(*) FROM _jiangchang_columns WHERE table_name = 'platforms'"
+        )
+        assert int(cur.fetchone()[0]) < len(physical)
+    finally:
+        conn.close()
+
+    # Full seed should cover all physical columns
+    conn = sqlite3.connect(":memory:")
+    try:
+        from db.schema import ensure_schema
+
+        ensure_schema(conn)
+        init_display_metadata(conn.cursor())
+        conn.commit()
+        for table_name in USER_BUSINESS_TABLES:
+            physical = set(list_physical_column_names(conn.cursor(), table_name))
+            cur = conn.cursor()
+            cur.execute(
+                "SELECT column_name FROM _jiangchang_columns WHERE table_name = ?",
+                (table_name,),
+            )
+            meta_cols = {row[0] for row in cur.fetchall()}
+            assert meta_cols == physical, table_name
+    finally:
+        conn.close()