302 lines
9.0 KiB
Python
302 lines
9.0 KiB
Python
# -*- 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_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()
|