# -*- coding: utf-8 -*- """数据管理展示元数据：幂等初始化、中文名称、PRAGMA 字段顺序、校验规则。""" from __future__ import annotations import os import sqlite3 import unittest from _support import ( DEMO_ITEMS_COLUMN_METADATA, DEMO_ITEMS_TABLE, IsolatedDataRoot, get_skill_root, seed_demo_items, ) class TestDisplayMetadata(unittest.TestCase): def test_init_db_metadata_idempotent(self) -> None: with IsolatedDataRoot(): from db.connection import get_conn, init_db init_db() init_db() conn = get_conn() try: cur = conn.cursor() cur.execute("SELECT COUNT(*) FROM _jiangchang_tables") self.assertEqual(int(cur.fetchone()[0]), 0) cur.execute("SELECT COUNT(*) FROM _jiangchang_columns") self.assertEqual(int(cur.fetchone()[0]), 0) cur.execute( "SELECT 1 FROM sqlite_master WHERE type='table' AND name='task_logs'" ) self.assertIsNone(cur.fetchone()) finally: conn.close() def test_demo_items_fixture_has_chinese_names(self) -> None: with IsolatedDataRoot(): from db.connection import get_conn, init_db init_db() conn = get_conn() try: cur = conn.cursor() seed_demo_items(cur) conn.commit() cur.execute( "SELECT display_name, readonly FROM _jiangchang_tables WHERE table_name = ?", (DEMO_ITEMS_TABLE,), ) display_name, readonly = cur.fetchone() self.assertEqual(display_name, "示例记录") self.assertEqual(int(readonly), 0) cur.execute( """ SELECT column_name, display_name, editable FROM _jiangchang_columns WHERE table_name = ? ORDER BY column_name """, (DEMO_ITEMS_TABLE,), ) by_name = {row[0]: (row[1], row[2]) for row in cur.fetchall()} self.assertEqual(by_name["status"][0], "状态") self.assertEqual(int(by_name["status"][1]), 1) self.assertEqual(by_name["title"][0], "标题") self.assertEqual(by_name["created_at"][0], "创建时间") self.assertEqual(int(by_name["created_at"][1]), 0) finally: conn.close() def test_readonly_table_not_manual_create_in_host_ui(self) -> None: with IsolatedDataRoot(): from db.connection import get_conn, init_db from db.display_metadata import table_allows_manual_create init_db() conn = get_conn() try: cur = conn.cursor() seed_demo_items(cur, readonly=1) conn.commit() self.assertFalse(table_allows_manual_create(cur, DEMO_ITEMS_TABLE)) finally: conn.close() def test_demo_items_physical_order_matches_pragma_cid(self) -> None: with IsolatedDataRoot(): from db.connection import get_conn, init_db from db.display_metadata import list_physical_column_names init_db() conn = get_conn() try: cur = conn.cursor() seed_demo_items(cur) conn.commit() physical = list_physical_column_names(cur, DEMO_ITEMS_TABLE) expected = [str(item["column_name"]) for item in DEMO_ITEMS_COLUMN_METADATA] 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_seed_sets_display_order_null(self) -> None: with IsolatedDataRoot(): from db.connection import get_conn, init_db init_db() conn = get_conn() try: cur = conn.cursor() seed_demo_items(cur) conn.commit() cur.execute( """ SELECT column_name, display_order FROM _jiangchang_columns WHERE table_name = ? """, (DEMO_ITEMS_TABLE,), ) for column_name, display_order in cur.fetchall(): self.assertIsNone( display_order, msg=f"{column_name} display_order should be NULL", ) finally: conn.close() def test_reseed_clears_stale_display_order(self) -> None: with IsolatedDataRoot(): from db.connection import get_conn, init_db init_db() conn = get_conn() try: cur = conn.cursor() seed_demo_items(cur) conn.execute( """ UPDATE _jiangchang_columns SET display_order = 99 WHERE table_name = ? AND column_name = 'id' """, (DEMO_ITEMS_TABLE,), ) conn.commit() seed_demo_items(cur) conn.commit() cur.execute( """ SELECT display_order FROM _jiangchang_columns WHERE table_name = ? AND column_name = 'id' """, (DEMO_ITEMS_TABLE,), ) self.assertIsNone(cur.fetchone()[0]) finally: conn.close() def test_legacy_display_order_does_not_change_pragma_physical_order(self) -> None: with IsolatedDataRoot(): from db.connection import get_conn, init_db from db.display_metadata import list_physical_column_names init_db() conn = get_conn() try: cur = conn.cursor() seed_demo_items(cur) conn.commit() cur.execute( """ UPDATE _jiangchang_columns SET display_order = 1 WHERE table_name = ? AND column_name = 'updated_at' """, (DEMO_ITEMS_TABLE,), ) cur.execute( """ UPDATE _jiangchang_columns SET display_order = 99 WHERE table_name = ? AND column_name = 'id' """, (DEMO_ITEMS_TABLE,), ) conn.commit() physical = list_physical_column_names(cur, DEMO_ITEMS_TABLE) self.assertEqual(physical[0], "id") self.assertEqual(physical[-1], "updated_at") finally: conn.close() def test_metadata_tables_are_not_user_tables(self) -> None: with IsolatedDataRoot(): from db.connection import get_conn, init_db from db.display_metadata import iter_user_tables init_db() conn = get_conn() try: tables = list(iter_user_tables(conn.cursor())) self.assertEqual(tables, []) finally: conn.close() def test_template_database_validation_passes(self) -> None: with IsolatedDataRoot(): from db.connection import get_conn, init_db from db.display_metadata_validator import validate_template_database init_db() conn = get_conn() try: report = validate_template_database(conn) self.assertTrue(report.ok, msg="; ".join(report.errors)) finally: conn.close() def test_validate_rejects_sqlite_foreign_keys_on_business_tables(self) -> None: from db.display_metadata import ( _JIANGCHANG_COLUMNS_DDL, _JIANGCHANG_TABLES_DDL, ) from db.display_metadata_validator import validate_display_metadata conn = sqlite3.connect(":memory:") try: conn.execute(_JIANGCHANG_TABLES_DDL) conn.execute(_JIANGCHANG_COLUMNS_DDL) conn.execute("CREATE TABLE parent_rows (id INTEGER PRIMARY KEY, title TEXT)") conn.execute( """ CREATE TABLE child_rows ( id INTEGER PRIMARY KEY, parent_id INTEGER NOT NULL, FOREIGN KEY (parent_id) REFERENCES parent_rows(id) ) """ ) report = validate_display_metadata(conn) self.assertFalse(report.ok) self.assertTrue( any("FOREIGN KEY" in err or "REFERENCES" in err for err in report.errors), msg=f"expected FK error, got: {report.errors}", ) self.assertTrue(any("child_rows" in err for err in report.errors)) finally: conn.close() def test_validate_missing_both_metadata_tables(self) -> None: conn = sqlite3.connect(":memory:") try: conn.execute("CREATE TABLE demo_items (id INTEGER PRIMARY KEY, title TEXT)") from db.display_metadata_validator import validate_display_metadata report = validate_display_metadata(conn) self.assertFalse(report.ok) self.assertIn("缺少数据管理元数据表 _jiangchang_tables", report.errors) self.assertIn("缺少数据管理元数据表 _jiangchang_columns", report.errors) finally: conn.close() def test_validate_missing_jiangchang_tables_only(self) -> None: from db.display_metadata import _JIANGCHANG_COLUMNS_DDL from db.display_metadata_validator import validate_display_metadata conn = sqlite3.connect(":memory:") try: conn.execute(_JIANGCHANG_COLUMNS_DDL) report = validate_display_metadata(conn) self.assertFalse(report.ok) self.assertIn("缺少数据管理元数据表 _jiangchang_tables", report.errors) self.assertNotIn("缺少数据管理元数据表 _jiangchang_columns", report.errors) finally: conn.close() def test_validate_missing_jiangchang_columns_only(self) -> None: from db.display_metadata import _JIANGCHANG_TABLES_DDL from db.display_metadata_validator import validate_display_metadata conn = sqlite3.connect(":memory:") try: conn.execute(_JIANGCHANG_TABLES_DDL) report = validate_display_metadata(conn) self.assertFalse(report.ok) self.assertIn("缺少数据管理元数据表 _jiangchang_columns", report.errors) self.assertNotIn("缺少数据管理元数据表 _jiangchang_tables", report.errors) finally: conn.close() def test_missing_table_metadata_is_reported(self) -> None: with IsolatedDataRoot(): from db.connection import get_conn, init_db from db.display_metadata_validator import validate_display_metadata init_db() conn = get_conn() try: conn.execute("CREATE TABLE demo_items (id INTEGER PRIMARY KEY, title TEXT)") conn.execute("DELETE FROM _jiangchang_tables WHERE table_name = 'demo_items'") conn.commit() 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_is_reported(self) -> None: with IsolatedDataRoot(): from db.connection import get_conn, init_db from db.display_metadata_validator import validate_display_metadata init_db() conn = get_conn() try: seed_demo_items(conn.cursor()) conn.execute( "DELETE FROM _jiangchang_columns WHERE table_name = ? AND column_name = 'status'", (DEMO_ITEMS_TABLE,), ) conn.commit() report = validate_display_metadata(conn) self.assertFalse(report.ok) self.assertTrue(any("demo_items.status" in err for err in report.errors)) finally: conn.close() def test_metadata_reference_to_missing_column_is_reported(self) -> None: with IsolatedDataRoot(): from db.connection import get_conn, init_db from db.display_metadata_validator import validate_display_metadata init_db() conn = get_conn() try: seed_demo_items(conn.cursor()) conn.execute( """ INSERT INTO _jiangchang_columns (table_name, column_name, display_name, visible, searchable, editable) VALUES (?, 'not_exists', '不存在', 1, 1, 1) """, (DEMO_ITEMS_TABLE,), ) conn.commit() 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_ordering_test_table_not_alphabetized(self) -> None: conn = sqlite3.connect(":memory:") try: conn.executescript( """ CREATE TABLE ordering_test ( z_field TEXT, id INTEGER PRIMARY KEY, a_field TEXT, created_at INTEGER, updated_at INTEGER ); """ ) from db.display_metadata import list_physical_column_names physical = list_physical_column_names(conn.cursor(), "ordering_test") self.assertEqual(physical, ["z_field", "id", "a_field", "created_at", "updated_at"]) finally: conn.close() def test_skill_frontmatter_validation_for_template(self) -> None: from db.display_metadata_validator import validate_skill_frontmatter report = validate_skill_frontmatter(get_skill_root(), allow_template_placeholders=True) self.assertTrue(report.ok, msg="; ".join(report.errors)) def test_skill_frontmatter_rejects_english_placeholder_name(self) -> None: import os import tempfile from db.display_metadata_validator import validate_skill_frontmatter with tempfile.TemporaryDirectory() as tmp: skill_md = """--- name: My Skill description: demo metadata: openclaw: slug: query-demo-info --- """ with open(os.path.join(tmp, "SKILL.md"), "w", encoding="utf-8") as f: f.write(skill_md) report = validate_skill_frontmatter(tmp, allow_template_placeholders=False) self.assertFalse(report.ok) self.assertTrue(any("name" in err for err in report.errors)) def test_column_order_warning_for_created_at_in_middle(self) -> None: from db.display_metadata_validator import column_order_warnings warnings = column_order_warnings( "bad_order", ["created_at", "id", "title", "updated_at"], ) self.assertTrue(any("id" in item for item in warnings)) def test_generated_column_editable_must_be_zero(self) -> None: conn = sqlite3.connect(":memory:") try: conn.executescript( """ CREATE TABLE gen_demo ( id INTEGER PRIMARY KEY, base_value INTEGER NOT NULL, generated_value INTEGER GENERATED ALWAYS AS (base_value * 2) STORED ); """ ) from db.display_metadata import init_display_metadata_tables, upsert_column_metadata from db.display_metadata_validator import validate_column_permissions cur = conn.cursor() init_display_metadata_tables(cur) upsert_column_metadata( cur, table_name="gen_demo", column_name="generated_value", display_name="生成值", editable=1, ) conn.commit() errors, _ = validate_column_permissions(cur, "gen_demo") self.assertTrue(any("generated" in err for err in errors)) finally: conn.close() def test_blob_column_editable_must_be_zero(self) -> None: conn = sqlite3.connect(":memory:") try: conn.execute( "CREATE TABLE blob_demo (id INTEGER PRIMARY KEY, payload BLOB NOT NULL)" ) from db.display_metadata import init_display_metadata_tables, upsert_column_metadata from db.display_metadata_validator import validate_column_permissions cur = conn.cursor() init_display_metadata_tables(cur) upsert_column_metadata( cur, table_name="blob_demo", column_name="payload", display_name="载荷", editable=1, ) conn.commit() errors, _ = validate_column_permissions(cur, "blob_demo") self.assertTrue(any("BLOB" in err for err in errors)) finally: conn.close() def test_primary_key_editable_must_be_zero(self) -> None: conn = sqlite3.connect(":memory:") try: conn.execute("CREATE TABLE pk_demo (id INTEGER PRIMARY KEY, title TEXT)") from db.display_metadata import init_display_metadata_tables, upsert_column_metadata from db.display_metadata_validator import validate_column_permissions cur = conn.cursor() init_display_metadata_tables(cur) upsert_column_metadata( cur, table_name="pk_demo", column_name="id", display_name="编号", editable=1, ) conn.commit() errors, _ = validate_column_permissions(cur, "pk_demo") self.assertTrue(any("主键" in err for err in errors)) finally: conn.close() def test_readonly_table_with_editable_fields_warns(self) -> None: conn = sqlite3.connect(":memory:") try: conn.executescript( """ CREATE TABLE warn_demo (id INTEGER PRIMARY KEY, title TEXT); """ ) from db.display_metadata import ( init_display_metadata_tables, upsert_column_metadata, upsert_table_metadata, ) from db.display_metadata_validator import validate_column_permissions cur = conn.cursor() init_display_metadata_tables(cur) upsert_table_metadata(cur, table_name="warn_demo", display_name="警告演示", readonly=1) upsert_column_metadata( cur, table_name="warn_demo", column_name="title", display_name="标题", editable=1, ) conn.commit() _, warnings = validate_column_permissions(cur, "warn_demo") self.assertTrue(any("readonly=1" in item for item in warnings)) finally: conn.close() def test_writable_table_without_editable_fields_warns(self) -> None: conn = sqlite3.connect(":memory:") try: conn.executescript( """ CREATE TABLE warn_demo2 (id INTEGER PRIMARY KEY, title TEXT); """ ) from db.display_metadata import ( init_display_metadata_tables, upsert_column_metadata, upsert_table_metadata, ) from db.display_metadata_validator import validate_column_permissions cur = conn.cursor() init_display_metadata_tables(cur) upsert_table_metadata(cur, table_name="warn_demo2", display_name="警告演示二", readonly=0) upsert_column_metadata( cur, table_name="warn_demo2", column_name="title", display_name="标题", editable=0, ) conn.commit() _, warnings = validate_column_permissions(cur, "warn_demo2") self.assertTrue(any("readonly=0" in item and "editable=0" in item for item in warnings)) finally: conn.close() def test_enum_column_without_searchable_warns(self) -> None: conn = sqlite3.connect(":memory:") try: conn.executescript( """ CREATE TABLE filter_demo (id INTEGER PRIMARY KEY, enabled INTEGER NOT NULL DEFAULT 1); """ ) from db.display_metadata import ( init_display_metadata_tables, upsert_column_metadata, upsert_table_metadata, ) from db.display_metadata_validator import validate_filter_search_metadata cur = conn.cursor() init_display_metadata_tables(cur) upsert_table_metadata(cur, table_name="filter_demo", display_name="筛选演示", readonly=0) upsert_column_metadata( cur, table_name="filter_demo", column_name="enabled", display_name="是否启用", visible=1, searchable=0, editable=1, display_type="boolean", options_json='[{"value":0,"label":"停用"},{"value":1,"label":"启用"}]', ) conn.commit() _, warnings = validate_filter_search_metadata(cur, "filter_demo") self.assertTrue(any("searchable=1" in item for item in warnings)) finally: conn.close() def test_schema_md_documents_readonly_and_seed_helpers(self) -> None: path = os.path.join(get_skill_root(), "references", "SCHEMA.md") with open(path, encoding="utf-8") as f: text = f.read() self.assertIn("readonly = 1", text) self.assertIn("seed_column_metadata_batch", text) self.assertIn("valid_column_names", text) def test_default_visible_columns_json_upsert_and_migrate(self) -> None: conn = sqlite3.connect(":memory:") try: from db.display_metadata import ( init_display_metadata_tables, upsert_table_metadata, ) cur = conn.cursor() # Simulate pre-1.0.79 table without the new column. cur.execute( """ CREATE TABLE _jiangchang_tables ( table_name TEXT PRIMARY KEY, display_name TEXT NOT NULL, description TEXT, sort_order INTEGER, visible INTEGER NOT NULL DEFAULT 1, readonly INTEGER NOT NULL DEFAULT 0, default_sort_json TEXT ) """ ) init_display_metadata_tables(cur) cur.execute("PRAGMA table_info('_jiangchang_tables')") cols = {str(row[1]) for row in cur.fetchall()} self.assertIn("default_visible_columns_json", cols) payload = '["stat_type","period","stage"]' upsert_table_metadata( cur, table_name="timeliness_stats", display_name="时效统计", readonly=1, default_visible_columns_json=payload, ) conn.commit() cur.execute( """ SELECT default_visible_columns_json FROM _jiangchang_tables WHERE table_name = 'timeliness_stats' """ ) self.assertEqual(cur.fetchone()[0], payload) finally: conn.close() def test_grid_help_upsert_and_migrate_from_legacy_schema(self) -> None: conn = sqlite3.connect(":memory:") try: from db.display_metadata import ( init_display_metadata_tables, upsert_column_metadata, upsert_table_metadata, ) cur = conn.cursor() cur.execute( """ CREATE TABLE _jiangchang_tables ( table_name TEXT PRIMARY KEY, display_name TEXT NOT NULL, description TEXT, sort_order INTEGER, visible INTEGER NOT NULL DEFAULT 1, readonly INTEGER NOT NULL DEFAULT 0, default_sort_json TEXT, default_visible_columns_json TEXT ) """ ) cur.execute( """ CREATE TABLE _jiangchang_columns ( table_name TEXT NOT NULL, column_name TEXT NOT NULL, display_name TEXT NOT NULL, description TEXT, display_order INTEGER, visible INTEGER NOT NULL DEFAULT 1, searchable INTEGER NOT NULL DEFAULT 1, editable INTEGER NOT NULL DEFAULT 1, display_type TEXT, options_json TEXT, PRIMARY KEY (table_name, column_name) ) """ ) init_display_metadata_tables(cur) cur.execute("PRAGMA table_info('_jiangchang_tables')") self.assertIn("grid_help", {str(row[1]) for row in cur.fetchall()}) cur.execute("PRAGMA table_info('_jiangchang_columns')") self.assertIn("grid_help", {str(row[1]) for row in cur.fetchall()}) upsert_table_metadata( cur, table_name="demo_items", display_name="示例记录", description="表单备注，不应出现在表格横幅", grid_help="表格横幅说明", ) upsert_column_metadata( cur, table_name="demo_items", column_name="title", display_name="标题", description="填标题", grid_help="列头问号说明", ) upsert_table_metadata( cur, table_name="empty_help", display_name="无表格说明", ) conn.commit() cur.execute( """ SELECT description, grid_help FROM _jiangchang_tables WHERE table_name = 'demo_items' """ ) self.assertEqual(cur.fetchone(), ("表单备注，不应出现在表格横幅", "表格横幅说明")) cur.execute( """ SELECT description, grid_help FROM _jiangchang_columns WHERE table_name = 'demo_items' AND column_name = 'title' """ ) self.assertEqual(cur.fetchone(), ("填标题", "列头问号说明")) cur.execute( """ SELECT grid_help FROM _jiangchang_tables WHERE table_name = 'empty_help' """ ) self.assertTrue(not cur.fetchone()[0]) finally: conn.close() def test_drop_deprecated_user_table_removes_physical_and_metadata(self) -> None: conn = sqlite3.connect(":memory:") try: cur = conn.cursor() cur.execute( """ CREATE TABLE _jiangchang_tables ( table_name TEXT PRIMARY KEY, display_name TEXT, description TEXT, sort_order INTEGER, visible INTEGER, readonly INTEGER ) """ ) cur.execute( """ CREATE TABLE _jiangchang_columns ( table_name TEXT, column_name TEXT, display_name TEXT, description TEXT, display_order INTEGER, visible INTEGER, searchable INTEGER, editable INTEGER, display_type TEXT, options_json TEXT, PRIMARY KEY (table_name, column_name) ) """ ) cur.execute( """ CREATE TABLE task_logs ( id INTEGER PRIMARY KEY, status TEXT, created_at INTEGER, updated_at INTEGER ) """ ) cur.execute( "INSERT INTO _jiangchang_tables VALUES ('task_logs', '任务日志', NULL, 1, 1, 0)" ) cur.execute( "INSERT INTO _jiangchang_columns VALUES " "('task_logs', 'id', '编号', NULL, 1, 1, 1, 0, 'text', NULL)" ) cur.execute("INSERT INTO task_logs (status) VALUES ('success')") from db.display_metadata import drop_deprecated_user_table, iter_user_tables drop_deprecated_user_table(cur, "task_logs") cur.execute( "SELECT 1 FROM sqlite_master WHERE type='table' AND name='task_logs'" ) self.assertIsNone(cur.fetchone()) cur.execute( "SELECT COUNT(*) FROM _jiangchang_tables WHERE table_name='task_logs'" ) self.assertEqual(int(cur.fetchone()[0]), 0) self.assertNotIn("task_logs", list(iter_user_tables(cur))) finally: conn.close() if __name__ == "__main__": unittest.main()