chendelian
766d162245
All checks were successful
技能自动化发布 / release (push) Successful in 4s
Align scaffold slug replacement, Action manifest strict boundaries, metadata prune API, field permission validation, and task_logs readonly semantics with documented template contracts. Co-authored-by: Cursor <cursoragent@cursor.com>
480 lines
18 KiB
Python
480 lines
18 KiB
Python
# -*- 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()
|