feat: 工资查询助手首版 v1.0.105
基于 skill-template 脚手架开发，包含： - payroll_config（费率配置，可编辑）+ payslip_snapshots（解析历史，只读）2 张业务表 - 全国默认 + 北上广 4 套费率（社保/公积金/个税起征点/7 级税率表） - 5 个业务 Action：calculate-payslip-simple / calculate-payslip-full / explain-term / view-config / edit-config - 个税累计预扣法计算，支持多月历史累加 - 20 个薪资名词库（社保/公积金/个税/全勤/加班费等） - 完整市场文档（README/TUTORIAL/DEMO/CHANGELOG） - 296 测试全部通过
This commit is contained in:
skill-dev
812
tests/test_display_metadata.py
Normal file
812
tests/test_display_metadata.py
Normal file
@@ -0,0 +1,812 @@
|
||||
# -*- 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()
|
||||
# 本技能 init_db 会建 2 张业务表（payroll_config + payslip_snapshots）
|
||||
cur.execute("SELECT COUNT(*) FROM _jiangchang_tables")
|
||||
self.assertEqual(int(cur.fetchone()[0]), 2)
|
||||
cur.execute("SELECT COUNT(*) FROM _jiangchang_columns")
|
||||
self.assertGreaterEqual(int(cur.fetchone()[0]), 2)
|
||||
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()))
|
||||
# 本技能有 2 张业务表
|
||||
self.assertEqual(tables, ["payroll_config", "payslip_snapshots"])
|
||||
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()
|
||||
Reference in New Issue
Block a user