account-manager/tests/test_display_metadata.py

# -*- 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_account_and_credential_platform_lookups(metadata_db):
    import json

    cur = metadata_db.cursor()
    for table_name, column_name in (
        (ACCOUNTS_TABLE, "platform_key"),
        (CREDENTIALS_TABLE, "platform_key"),
    ):
        cur.execute(
            """
            SELECT display_type, options_json
            FROM _jiangchang_columns
            WHERE table_name = ? AND column_name = ?
            """,
            (table_name, column_name),
        )
        display_type, options_json = cur.fetchone()
        assert display_type == "lookup", table_name
        payload = json.loads(options_json)
        assert payload["ref_table"] == "platforms"
        assert payload["value_column"] == "platform_key"
        assert payload["label_column"] == "display_name"

    cur.execute(
        """
        SELECT display_type, options_json
        FROM _jiangchang_columns
        WHERE table_name = ? AND column_name = 'account_id'
        """,
        (CREDENTIALS_TABLE,),
    )
    display_type, options_json = cur.fetchone()
    assert display_type == "lookup"
    payload = json.loads(options_json)
    assert payload["ref_table"] == "accounts"
    assert payload["value_column"] == "id"


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_account_edit_form_columns_and_enums(metadata_db):
    import json

    cur = metadata_db.cursor()
    expected_editable = {
        "account_label",
        "login_id",
        "login_id_type",
        "url",
        "status",
    }
    readonly_runtime = {
        "platform_key",
        "tenant_id",
        "environment",
        "role",
        "provider_code",
        "session_status",
        "auth_strategy",
        "session_persistent",
        "device_fingerprint",
    }
    for column_name in expected_editable:
        cur.execute(
            """
            SELECT editable FROM _jiangchang_columns
            WHERE table_name = ? AND column_name = ?
            """,
            (ACCOUNTS_TABLE, column_name),
        )
        assert int(cur.fetchone()[0]) == 1, column_name
    for column_name in readonly_runtime:
        cur.execute(
            """
            SELECT editable FROM _jiangchang_columns
            WHERE table_name = ? AND column_name = ?
            """,
            (ACCOUNTS_TABLE, column_name),
        )
        assert int(cur.fetchone()[0]) == 0, column_name

    cur.execute(
        """
        SELECT options_json FROM _jiangchang_columns
        WHERE table_name = ? AND column_name = 'status'
        """,
        (ACCOUNTS_TABLE,),
    )
    options = json.loads(cur.fetchone()[0])
    values = {item["value"] for item in options}
    labels = {item["label"] for item in options}
    assert values == {"active", "disabled", "needs_review"}
    assert "启用" in labels
    assert "停用" in labels


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()