feat: standardize skill data management and actions
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
400
tests/test_data_management_contract.py
Normal file
400
tests/test_data_management_contract.py
Normal file
@@ -0,0 +1,400 @@
|
||||
# -*- 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) -> None:
|
||||
from db.display_metadata import (
|
||||
init_display_metadata_tables,
|
||||
list_physical_column_names,
|
||||
prune_stale_column_metadata,
|
||||
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"]),
|
||||
)
|
||||
seed_column_metadata_batch(cur, table_name="demo_items", columns=DEMO_ITEMS_COLUMN_METADATA)
|
||||
physical = list_physical_column_names(cur, "demo_items")
|
||||
prune_stale_column_metadata(cur, "demo_items", physical)
|
||||
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, prune_stale_column_metadata
|
||||
|
||||
cur = conn.cursor()
|
||||
physical = list_physical_column_names(cur, "demo_items")
|
||||
prune_stale_column_metadata(cur, "demo_items", 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_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()
|
||||
Reference in New Issue
Block a user