From e3666052991cfcc86cf59eabc7fa816daedae3c3 Mon Sep 17 00:00:00 2001 From: chendelian <116870791@qq.com> Date: Wed, 1 Jul 2026 18:51:33 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E8=A7=84=E8=8C=83=E6=A0=87=E5=87=86?= =?UTF-8?q?=E6=97=B6=E9=97=B4=E5=AD=97=E6=AE=B5=20created=5Fat/updated=5Fa?= =?UTF-8?q?t=EF=BC=88Unix=20=E7=A7=92=E7=BA=A7=E3=80=81=E5=85=83=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E3=80=81trigger=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- SKILL.md | 2 +- development/DEVELOPMENT.md | 2 + development/REQUIREMENTS.md | 2 +- development/RUNTIME.md | 2 +- development/TESTING.md | 1 + references/SCHEMA.md | 43 ++- scripts/db/connection.py | 6 +- scripts/db/display_metadata.py | 33 ++- scripts/db/display_metadata_validator.py | 121 ++++++++ scripts/db/task_logs_repository.py | 8 +- scripts/db/timestamp_columns.py | 36 +++ tests/README.md | 1 + tests/test_timestamp_columns.py | 336 +++++++++++++++++++++++ 13 files changed, 571 insertions(+), 22 deletions(-) create mode 100644 scripts/db/timestamp_columns.py create mode 100644 tests/test_timestamp_columns.py diff --git a/SKILL.md b/SKILL.md index 34aaad8..7bd73cf 100644 --- a/SKILL.md +++ b/SKILL.md @@ -82,7 +82,7 @@ python {baseDir}/scripts/main.py version 2. 按 `development/DEVELOPMENT.md` 完成开发与文档定制。 3. 用户市场说明写入根 `README.md`；Agent 调用契约见 `references/CLI.md`、`references/SCHEMA.md`。 4. 同步修改 `scripts/util/constants.py` 中的 `SKILL_SLUG` / `SKILL_VERSION`。 -5. 本地 SQLite 的中文表名/字段名与字段顺序规范见 `references/SCHEMA.md`（元数据表 `_jiangchang_*` + `CREATE TABLE` 物理顺序）。 +5. 本地 SQLite 的中文表名/字段名、标准时间字段与字段顺序规范见 `references/SCHEMA.md`（元数据表 `_jiangchang_*` + `CREATE TABLE` 物理顺序）。 ## 技能命名与 slug diff --git a/development/DEVELOPMENT.md b/development/DEVELOPMENT.md index 253474c..71f8718 100644 --- a/development/DEVELOPMENT.md +++ b/development/DEVELOPMENT.md @@ -447,6 +447,8 @@ metadata: 只做增删查改（模板默认表：`task_logs`） - `db/display_metadata.py` 创建 `_jiangchang_tables` / `_jiangchang_columns` 并幂等写入中文展示名（见 `references/SCHEMA.md`） +- `db/timestamp_columns.py` + 标准时间字段 trigger 与维护约定（`created_at` / `updated_at`，Unix 秒级） 不要在 `db` 层里： diff --git a/development/REQUIREMENTS.md b/development/REQUIREMENTS.md index c760d00..73ff944 100644 --- a/development/REQUIREMENTS.md +++ b/development/REQUIREMENTS.md @@ -149,7 +149,7 @@ - `health`、`version` 命令执行正常 - 主命令（如 `run`）在 mock / simulator 档位可重复验证 - `python tests/run_tests.py -v` 必跑测试全部通过 -- `task_logs` 写入和查询符合 `references/SCHEMA.md` +- `task_logs` 写入和查询符合 `references/SCHEMA.md`（含 `created_at` / `updated_at` Unix 秒级规范） - `init_db()` 已写入 `_jiangchang_tables` / `_jiangchang_columns`；用户可见表/字段具备中文 `display_name` - 字段展示顺序与 `PRAGMA table_info(task_logs)` 的 cid 一致；不依赖 `display_order` - `tests/test_display_metadata.py` 通过 diff --git a/development/RUNTIME.md b/development/RUNTIME.md index 82178ad..3ec809d 100644 --- a/development/RUNTIME.md +++ b/development/RUNTIME.md @@ -71,7 +71,7 @@ RPA 录屏成片（`RpaVideoSession`）、ffmpeg 路径、背景音乐探测均 - `scripts/main.py`：唯一 CLI 入口 - `scripts/cli/`：参数解析与命令分发 -- `scripts/db/`：SQLite 或本地持久化层 +- `scripts/db/`：SQLite 或本地持久化层（标准时间字段见 `references/SCHEMA.md`） - `scripts/service/`：业务用例与外部交互 - `scripts/util/`：通用工具、常量、日志、路径薄封装 diff --git a/development/TESTING.md b/development/TESTING.md index 327f900..0755f8d 100644 --- a/development/TESTING.md +++ b/development/TESTING.md @@ -29,6 +29,7 @@ - `SKILL.md` YAML slug vs [`constants.SKILL_SLUG`](../scripts/util/constants.py)； - SQLite 骨架：`task_logs` 创建幂等与仓储读写； - 数据管理元数据：`_jiangchang_*` 幂等初始化、中文展示名、`PRAGMA table_info` 字段顺序（见 `test_display_metadata.py`）； +- 标准时间字段：`created_at` / `updated_at` 默认值、trigger 维护与 `datetime_unix_seconds` 元数据（见 `test_timestamp_columns.py`）； - **adapter profile**：[`tests/test_adapter_profile_policy.py`](../tests/test_adapter_profile_policy.py) + [`adapter_test_utils`](../tests/adapter_test_utils.py) ——验证在未授权情况下绝不误判开启真实网络/RPA。 - **发布打包守护**：[`tests/test_release_packaging_constraints.py`](../tests/test_release_packaging_constraints.py) ——`scripts/**/*.py` 单文件 < 1000 行、文本文件 UTF-8 without BOM。 diff --git a/references/SCHEMA.md b/references/SCHEMA.md index f9282d5..f21b9e6 100644 --- a/references/SCHEMA.md +++ b/references/SCHEMA.md @@ -69,7 +69,7 @@ metadata: | `visible` | 是 | 是否展示 | | `searchable` | 是 | 是否可搜索 | | `editable` | 是 | 是否可编辑（与宿主权限结合） | -| `display_type` | 否 | 如 `text` / `textarea` | +| `display_type` | 否 | 如 `text` / `textarea` / `datetime_unix_seconds`（标准时间字段见下文） | | `options_json` | 否 | 枚举选项 JSON | 这两张表是**数据字典**，不是业务数据；宿主不会把它们当作普通业务表展示。 @@ -90,7 +90,36 @@ metadata: 4. 状态、结果、错误等辅助字段 5. created_at（如有） 6. updated_at（如有） -``` + +## 标准时间字段（新技能统一规范） + +新技能业务表若需要时间审计字段，**统一**采用 Unix **秒级**时间戳，不要使用毫秒时间戳、ISO 字符串或 SQLite 文本 datetime。 + +| 项目 | 规范 | +|------|------| +| 物理字段 | `created_at INTEGER`、`updated_at INTEGER` | +| 存储格式 | Unix 秒级整数 | +| 展示元数据 `display_type` | `datetime_unix_seconds` | +| 中文名 | `created_at` → **创建时间**；`updated_at` → **更新时间** | +| 用户编辑 | `editable = 0`（系统字段，宿主只展示格式化结果，不写回字符串） | +| 字段顺序 | 位于业务字段末尾：`… → created_at → updated_at` | +| 排序 | **不使用** `display_order`；顺序仅来自 `PRAGMA table_info` | + +### 自动生成与维护 + +- **created_at**：`INTEGER NOT NULL DEFAULT (unixepoch())`；插入时可省略，由数据库生成。 +- **updated_at**：`INTEGER NOT NULL DEFAULT (unixepoch())`；插入时同样可省略。 +- **updated_at 更新**：模板默认表 `task_logs` 通过 SQLite trigger `{table}_set_updated_at` 在业务字段更新后自动刷新；若技能所有 UPDATE 都经过统一 repository，也可在 repository 层设置 `updated_at = unixepoch()`，**不要**叠加两套机制。 + +实现参考： + +- DDL 与 trigger：`scripts/db/connection.py`、`scripts/db/timestamp_columns.py` +- 中文元数据种子：`scripts/db/display_metadata.py` +- 自检：`scripts/db/display_metadata_validator.py` + +宿主（匠厂）只负责把 `display_type = datetime_unix_seconds` 格式化为 `YYYY-MM-DD HH:mm:ss` 展示，**不修改**数据库原始值。 + +> `datetime_unix_milliseconds`、`datetime_iso` 是宿主兼容能力，**不是**新技能模板推荐标准。 ### 模板默认：`task_logs` @@ -106,11 +135,13 @@ CREATE TABLE IF NOT EXISTS task_logs ( status TEXT NOT NULL, error_msg TEXT, result_summary TEXT, - created_at INTEGER NOT NULL, - updated_at INTEGER NOT NULL + created_at INTEGER NOT NULL DEFAULT (unixepoch()), + updated_at INTEGER NOT NULL DEFAULT (unixepoch()) ); ``` +插入时可省略 `created_at`、`updated_at`；`updated_at` 在 UPDATE 时由 trigger `task_logs_set_updated_at` 自动维护（见 `scripts/db/timestamp_columns.py`）。 + 对应中文展示（由 `init_db()` 写入元数据表）： | 物理字段 | 中文名 | @@ -123,8 +154,8 @@ CREATE TABLE IF NOT EXISTS task_logs ( | status | 状态 | | error_msg | 错误信息 | | result_summary | 结果摘要 | -| created_at | 创建时间 | -| updated_at | 更新时间 | +| created_at | 创建时间（`display_type = datetime_unix_seconds`，不可编辑） | +| updated_at | 更新时间（`display_type = datetime_unix_seconds`，不可编辑） | 表中文名：**任务日志**。 diff --git a/scripts/db/connection.py b/scripts/db/connection.py index 034115b..74ffced 100644 --- a/scripts/db/connection.py +++ b/scripts/db/connection.py @@ -5,6 +5,7 @@ from __future__ import annotations import sqlite3 from db.display_metadata import init_display_metadata +from db.timestamp_columns import init_task_logs_timestamp_maintenance from util.runtime_paths import get_db_path @@ -27,11 +28,12 @@ def init_db() -> None: status TEXT NOT NULL, error_msg TEXT, result_summary TEXT, - created_at INTEGER NOT NULL, - updated_at INTEGER NOT NULL + created_at INTEGER NOT NULL DEFAULT (unixepoch()), + updated_at INTEGER NOT NULL DEFAULT (unixepoch()) ) """ ) + init_task_logs_timestamp_maintenance(cur) init_display_metadata(cur) conn.commit() finally: diff --git a/scripts/db/display_metadata.py b/scripts/db/display_metadata.py index 622a06c..4f67165 100644 --- a/scripts/db/display_metadata.py +++ b/scripts/db/display_metadata.py @@ -14,6 +14,8 @@ LEGACY_METADATA_TABLES = frozenset({"_schema_meta"}) METADATA_TABLES = INTERNAL_METADATA_TABLES | LEGACY_METADATA_TABLES TASK_LOGS_TABLE = "task_logs" +DATETIME_UNIX_SECONDS = "datetime_unix_seconds" +STANDARD_TIMESTAMP_COLUMNS = frozenset({"created_at", "updated_at"}) _JIANGCHANG_TABLES_DDL = """ CREATE TABLE IF NOT EXISTS _jiangchang_tables ( @@ -59,8 +61,24 @@ TASK_LOGS_COLUMN_METADATA: tuple[dict[str, object], ...] = ( {"column_name": "status", "display_name": "状态", "description": "任务当前状态", "visible": 1, "searchable": 1, "editable": 1}, {"column_name": "error_msg", "display_name": "错误信息", "description": "任务失败时的错误说明", "visible": 1, "searchable": 1, "editable": 1}, {"column_name": "result_summary", "display_name": "结果摘要", "description": "任务执行结果摘要", "visible": 1, "searchable": 1, "editable": 1}, - {"column_name": "created_at", "display_name": "创建时间", "description": "记录创建时间", "visible": 1, "searchable": 1, "editable": 0}, - {"column_name": "updated_at", "display_name": "更新时间", "description": "记录最后更新时间", "visible": 1, "searchable": 1, "editable": 0}, + { + "column_name": "created_at", + "display_name": "创建时间", + "description": "记录创建时间（Unix 秒级时间戳，数据库自动生成）", + "visible": 1, + "searchable": 0, + "editable": 0, + "display_type": DATETIME_UNIX_SECONDS, + }, + { + "column_name": "updated_at", + "display_name": "更新时间", + "description": "记录最后更新时间（Unix 秒级时间戳，自动维护）", + "visible": 1, + "searchable": 0, + "editable": 0, + "display_type": DATETIME_UNIX_SECONDS, + }, ) @@ -94,24 +112,26 @@ def upsert_column_metadata( visible: int = 1, searchable: int = 1, editable: int = 1, + display_type: str | None = None, ) -> None: # display_order 保留为 NULL：宿主按 PRAGMA table_info cid 展示字段，不读 display_order。 cursor.execute( """ INSERT INTO _jiangchang_columns ( table_name, column_name, display_name, description, - display_order, visible, searchable, editable + display_order, visible, searchable, editable, display_type ) - VALUES (?, ?, ?, ?, NULL, ?, ?, ?) + VALUES (?, ?, ?, ?, NULL, ?, ?, ?, ?) ON CONFLICT(table_name, column_name) DO UPDATE SET display_name = excluded.display_name, description = excluded.description, display_order = NULL, visible = excluded.visible, searchable = excluded.searchable, - editable = excluded.editable + editable = excluded.editable, + display_type = excluded.display_type """, - (table_name, column_name, display_name, description, visible, searchable, editable), + (table_name, column_name, display_name, description, visible, searchable, editable, display_type), ) @@ -135,6 +155,7 @@ def seed_task_logs_display_metadata(cursor) -> None: visible=int(column["visible"]), searchable=int(column["searchable"]), editable=int(column["editable"]), + display_type=str(column["display_type"]) if column.get("display_type") else None, ) diff --git a/scripts/db/display_metadata_validator.py b/scripts/db/display_metadata_validator.py index 0a3668d..6be3823 100644 --- a/scripts/db/display_metadata_validator.py +++ b/scripts/db/display_metadata_validator.py @@ -8,12 +8,15 @@ import sqlite3 from dataclasses import dataclass, field from db.display_metadata import ( + DATETIME_UNIX_SECONDS, METADATA_TABLES, + STANDARD_TIMESTAMP_COLUMNS, TASK_LOGS_COLUMN_METADATA, TASK_LOGS_TABLE, iter_user_tables, list_physical_column_names, ) +from db.timestamp_columns import has_updated_at_trigger from util.constants import SKILL_SLUG _TEMPLATE_PLACEHOLDER_SLUGS = frozenset({"your-skill-slug", "your_skill_slug"}) @@ -127,6 +130,116 @@ def validate_skill_frontmatter(skill_root: str, *, allow_template_placeholders: return report +def _quote_identifier(name: str) -> str: + return '"' + name.replace('"', '""') + '"' + + +def _table_column_info(cursor, table_name: str) -> dict[str, dict[str, object]]: + cursor.execute(f"PRAGMA table_info({_quote_identifier(table_name)})") + info: dict[str, dict[str, object]] = {} + for row in cursor.fetchall(): + info[str(row[1])] = { + "cid": int(row[0]), + "name": str(row[1]), + "type": str(row[2] or ""), + "notnull": int(row[3] or 0), + "dflt_value": row[4], + "pk": int(row[5] or 0), + } + return info + + +def _default_uses_unixepoch(default_value: object) -> bool: + if default_value is None: + return False + text = str(default_value).strip().lower() + return "unixepoch" in text + + +def _expected_timestamp_display_name(column_name: str) -> str: + if column_name == "created_at": + return "创建时间" + if column_name == "updated_at": + return "更新时间" + return "" + + +def validate_standard_timestamp_columns( + cursor, + table_name: str, + *, + physical_columns: list[str] | None = None, +) -> tuple[list[str], list[str]]: + errors: list[str] = [] + warnings: list[str] = [] + ordered = physical_columns or list_physical_column_names(cursor, table_name) + column_info = _table_column_info(cursor, table_name) + + for column_name in STANDARD_TIMESTAMP_COLUMNS: + if column_name not in ordered: + continue + + physical = column_info.get(column_name) + if not physical: + errors.append(f"{table_name}.{column_name} 缺少物理字段定义") + continue + + if "INT" not in str(physical["type"]).upper(): + errors.append( + f"{table_name}.{column_name} 字段类型应为 INTEGER（Unix 秒级时间戳），当前为 {physical['type']!r}" + ) + + if column_name == "created_at" and not _default_uses_unixepoch(physical["dflt_value"]): + errors.append( + f"{table_name}.created_at 应设置数据库默认值 DEFAULT (unixepoch())，当前默认值为 {physical['dflt_value']!r}" + ) + if column_name == "updated_at" and not _default_uses_unixepoch(physical["dflt_value"]): + errors.append( + f"{table_name}.updated_at 应设置数据库初始默认值 DEFAULT (unixepoch())，当前默认值为 {physical['dflt_value']!r}" + ) + + cursor.execute( + """ + SELECT display_name, display_type, editable, searchable, display_order + FROM _jiangchang_columns + WHERE table_name = ? AND column_name = ? + """, + (table_name, column_name), + ) + meta_row = cursor.fetchone() + if not meta_row: + errors.append(f"{table_name}.{column_name} 缺少 _jiangchang_columns 元数据") + continue + + display_name, display_type, editable, searchable, display_order = meta_row + expected_name = _expected_timestamp_display_name(column_name) + if str(display_name or "").strip() != expected_name: + errors.append( + f"{table_name}.{column_name} 的 display_name 应为 {expected_name!r}，当前为 {display_name!r}" + ) + if str(display_type or "").strip() != DATETIME_UNIX_SECONDS: + errors.append( + f"{table_name}.{column_name} 的 display_type 应为 {DATETIME_UNIX_SECONDS!r}，当前为 {display_type!r}" + ) + if editable is None or int(editable) != 0: + errors.append(f"{table_name}.{column_name} 的 editable 应为 0（系统时间字段不可编辑）") + if display_order is not None: + errors.append(f"{table_name}.{column_name} 的 display_order 应为 NULL") + if searchable is not None and int(searchable) != 0: + warnings.append(f"{table_name}.{column_name} 的时间字段通常不必开启 searchable") + + if "created_at" in ordered and "updated_at" in ordered: + if ordered.index("updated_at") < ordered.index("created_at"): + warnings.append(f"{table_name}: updated_at 不应出现在 created_at 之前") + + if "updated_at" in ordered and not has_updated_at_trigger(cursor, table_name): + errors.append( + f"{table_name}.updated_at 缺少自动维护 trigger（期望 {table_name}_set_updated_at）" + ) + + return errors, warnings + + def column_order_warnings(table_name: str, ordered_names: list[str]) -> list[str]: warnings: list[str] = [] if "id" in ordered_names and ordered_names.index("id") != 0: @@ -208,6 +321,14 @@ def validate_display_metadata(conn: sqlite3.Connection, *, check_column_order: b f"字段 {table_name}.{column_name} 的 display_name 应使用中文：{display_name!r}" ) + ts_errors, ts_warnings = validate_standard_timestamp_columns( + cur, + table_name, + physical_columns=physical_columns, + ) + report.errors.extend(ts_errors) + report.warnings.extend(ts_warnings) + cur.execute("SELECT table_name, column_name FROM _jiangchang_columns") for table_name, column_name in cur.fetchall(): if table_name in METADATA_TABLES: diff --git a/scripts/db/task_logs_repository.py b/scripts/db/task_logs_repository.py index 504ebea..5cd87c9 100644 --- a/scripts/db/task_logs_repository.py +++ b/scripts/db/task_logs_repository.py @@ -9,7 +9,6 @@ from __future__ import annotations from typing import Any, List, Optional, Tuple from db.connection import get_conn, init_db -from util.timeutil import now_unix def save_task_log( @@ -22,17 +21,16 @@ def save_task_log( result_summary: Optional[str] = None, ) -> int: init_db() - now = now_unix() conn = get_conn() try: cur = conn.cursor() cur.execute( """ INSERT INTO task_logs - (task_type, target_id, input_id, input_title, status, error_msg, result_summary, created_at, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + (task_type, target_id, input_id, input_title, status, error_msg, result_summary) + VALUES (?, ?, ?, ?, ?, ?, ?) """, - (task_type, target_id, input_id, input_title or "", status, error_msg, result_summary, now, now), + (task_type, target_id, input_id, input_title or "", status, error_msg, result_summary), ) new_id = int(cur.lastrowid) conn.commit() diff --git a/scripts/db/timestamp_columns.py b/scripts/db/timestamp_columns.py new file mode 100644 index 0000000..9fa7923 --- /dev/null +++ b/scripts/db/timestamp_columns.py @@ -0,0 +1,36 @@ +"""标准时间字段：Unix 秒级时间戳、默认值与 updated_at 自动维护。""" + +from __future__ import annotations + +STANDARD_TIMESTAMP_COLUMNS = frozenset({"created_at", "updated_at"}) +DATETIME_UNIX_SECONDS = "datetime_unix_seconds" +TASK_LOGS_UPDATED_AT_TRIGGER = "task_logs_set_updated_at" + +_TASK_LOGS_UPDATED_AT_TRIGGER_DDL = f""" +CREATE TRIGGER IF NOT EXISTS {TASK_LOGS_UPDATED_AT_TRIGGER} +AFTER UPDATE ON task_logs +FOR EACH ROW +BEGIN + UPDATE task_logs + SET updated_at = unixepoch() + WHERE id = NEW.id AND updated_at = OLD.updated_at; +END +""" + + +def init_task_logs_timestamp_maintenance(cursor) -> None: + """幂等创建 task_logs.updated_at 自动维护 trigger。""" + cursor.execute(_TASK_LOGS_UPDATED_AT_TRIGGER_DDL) + + +def has_updated_at_trigger(cursor, table_name: str) -> bool: + trigger_name = f"{table_name}_set_updated_at" + cursor.execute( + """ + SELECT 1 FROM sqlite_master + WHERE type = 'trigger' AND name = ? AND tbl_name = ? + LIMIT 1 + """, + (trigger_name, table_name), + ) + return cursor.fetchone() is not None diff --git a/tests/README.md b/tests/README.md index 581ddd2..42c5fbd 100644 --- a/tests/README.md +++ b/tests/README.md @@ -74,6 +74,7 @@ python tests/run_tests.py test_cli_smoke | ``SKILL.md`` slug 与 ``constants.SKILL_SLUG`` | ``test_skill_metadata.py`` | | DB / ``task_logs`` 骨架冒烟 | ``test_db_smoke.py`` | | 数据管理元数据 / 字段顺序 | ``test_display_metadata.py`` | +| 标准时间字段（Unix 秒 / trigger / 元数据） | ``test_timestamp_columns.py`` | | **adapter profile / 外呼授权策略** | ``test_adapter_profile_policy.py`` + ``adapter_test_utils.py``（非 ``test_`` 前缀，不自动当用例收集） | | **无 vendored ``jiangchang_skill_core``** | ``test_platform_import.py`` | | **runtime diagnostics 架构守护** | ``test_runtime_diagnostics_integration.py`` | diff --git a/tests/test_timestamp_columns.py b/tests/test_timestamp_columns.py new file mode 100644 index 0000000..7b15de1 --- /dev/null +++ b/tests/test_timestamp_columns.py @@ -0,0 +1,336 @@ +# -*- coding: utf-8 -*- +"""标准时间字段：Unix 秒级时间戳、默认值、trigger 与元数据。""" +from __future__ import annotations + +import sqlite3 +import time +import unittest + +from _support import IsolatedDataRoot + + +class TestTimestampColumns(unittest.TestCase): + def test_init_db_creates_integer_timestamp_columns_with_defaults(self) -> None: + with IsolatedDataRoot(): + from db.connection import get_conn, init_db + from db.display_metadata import list_physical_column_names + from db.display_metadata_validator import _table_column_info + + init_db() + conn = get_conn() + try: + cur = conn.cursor() + physical = list_physical_column_names(cur, "task_logs") + self.assertEqual(physical[-2:], ["created_at", "updated_at"]) + + info = _table_column_info(cur, "task_logs") + for column_name in ("created_at", "updated_at"): + column = info[column_name] + self.assertIn("INT", str(column["type"]).upper()) + self.assertTrue(str(column["dflt_value"]).lower().find("unixepoch") >= 0) + finally: + conn.close() + + def test_insert_without_timestamps_auto_generates_unix_seconds(self) -> None: + with IsolatedDataRoot(): + from db.connection import get_conn, init_db + + init_db() + conn = get_conn() + try: + cur = conn.cursor() + before = int(time.time()) + cur.execute( + "INSERT INTO task_logs (task_type, status) VALUES ('auto-ts', 'success')" + ) + conn.commit() + cur.execute( + "SELECT created_at, updated_at FROM task_logs ORDER BY id DESC LIMIT 1" + ) + created_at, updated_at = cur.fetchone() + after = int(time.time()) + self.assertIsInstance(created_at, int) + self.assertIsInstance(updated_at, int) + self.assertGreaterEqual(created_at, before) + self.assertLessEqual(created_at, after) + self.assertEqual(created_at, updated_at) + finally: + conn.close() + + def test_update_business_field_keeps_created_at_and_refreshes_updated_at(self) -> None: + with IsolatedDataRoot(): + from db.connection import get_conn, init_db + + init_db() + conn = get_conn() + try: + cur = conn.cursor() + cur.execute( + """ + INSERT INTO task_logs (task_type, status, created_at, updated_at) + VALUES ('update-ts', 'success', 1000, 1000) + """ + ) + log_id = int(cur.lastrowid) + conn.commit() + + cur.execute("UPDATE task_logs SET status = 'failed' WHERE id = ?", (log_id,)) + conn.commit() + + cur.execute( + "SELECT created_at, updated_at FROM task_logs WHERE id = ?", + (log_id,), + ) + created_at, updated_at = cur.fetchone() + self.assertEqual(created_at, 1000) + self.assertGreater(updated_at, 1000) + finally: + conn.close() + + def test_reinit_does_not_rewrite_existing_timestamps(self) -> None: + with IsolatedDataRoot(): + from db.connection import get_conn, init_db + + init_db() + conn = get_conn() + try: + cur = conn.cursor() + cur.execute( + """ + INSERT INTO task_logs (task_type, status, created_at, updated_at) + VALUES ('stable-ts', 'success', 1234567890, 1234567891) + """ + ) + conn.commit() + finally: + conn.close() + + init_db() + conn = get_conn() + try: + cur = conn.cursor() + cur.execute( + "SELECT created_at, updated_at FROM task_logs WHERE task_type = 'stable-ts'" + ) + self.assertEqual(cur.fetchone(), (1234567890, 1234567891)) + finally: + conn.close() + + def test_timestamp_metadata_contract(self) -> None: + with IsolatedDataRoot(): + from db.connection import get_conn, init_db + from db.display_metadata import DATETIME_UNIX_SECONDS + + init_db() + conn = get_conn() + try: + cur = conn.cursor() + cur.execute( + """ + SELECT column_name, display_name, display_type, editable, display_order + FROM _jiangchang_columns + WHERE table_name = 'task_logs' + AND column_name IN ('created_at', 'updated_at') + """ + ) + by_name = { + name: (display_name, display_type, editable, display_order) + for name, display_name, display_type, editable, display_order in cur.fetchall() + } + self.assertEqual(by_name["created_at"][0], "创建时间") + self.assertEqual(by_name["updated_at"][0], "更新时间") + self.assertEqual(by_name["created_at"][1], DATETIME_UNIX_SECONDS) + self.assertEqual(by_name["updated_at"][1], DATETIME_UNIX_SECONDS) + self.assertEqual(int(by_name["created_at"][2]), 0) + self.assertEqual(int(by_name["updated_at"][2]), 0) + self.assertIsNone(by_name["created_at"][3]) + self.assertIsNone(by_name["updated_at"][3]) + finally: + conn.close() + + def test_metadata_upsert_keeps_timestamp_display_type(self) -> None: + with IsolatedDataRoot(): + from db.connection import get_conn, init_db + from db.display_metadata import DATETIME_UNIX_SECONDS + + init_db() + init_db() + conn = get_conn() + try: + cur = conn.cursor() + cur.execute( + """ + SELECT display_type FROM _jiangchang_columns + WHERE table_name = 'task_logs' AND column_name = 'created_at' + """ + ) + self.assertEqual(cur.fetchone()[0], DATETIME_UNIX_SECONDS) + finally: + conn.close() + + def test_repository_insert_uses_database_timestamps(self) -> None: + with IsolatedDataRoot(): + from db import task_logs_repository as tlr + from db.connection import get_conn, init_db + + init_db() + before = int(time.time()) + new_id = tlr.save_task_log(task_type="repo-ts", status="success") + after = int(time.time()) + + conn = get_conn() + try: + cur = conn.cursor() + cur.execute( + "SELECT created_at, updated_at FROM task_logs WHERE id = ?", + (new_id,), + ) + created_at, updated_at = cur.fetchone() + self.assertGreaterEqual(int(created_at), before) + self.assertLessEqual(int(created_at), after) + self.assertEqual(created_at, updated_at) + finally: + conn.close() + + def test_validator_rejects_missing_display_type(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: + conn.execute( + "UPDATE _jiangchang_columns SET display_type = NULL WHERE table_name = 'task_logs' AND column_name = 'created_at'" + ) + conn.commit() + report = validate_template_database(conn) + self.assertFalse(report.ok) + self.assertTrue(any("display_type" in err for err in report.errors)) + finally: + conn.close() + + def test_validator_rejects_wrong_timestamp_type(self) -> None: + conn = sqlite3.connect(":memory:") + try: + from db.display_metadata import init_display_metadata + from db.display_metadata_validator import validate_display_metadata + + conn.executescript( + """ + CREATE TABLE demo_items ( + id INTEGER PRIMARY KEY, + title TEXT, + created_at TEXT + ); + """ + ) + init_display_metadata(conn.cursor()) + conn.execute( + """ + INSERT INTO _jiangchang_tables (table_name, display_name, visible, readonly) + VALUES ('demo_items', '演示', 1, 0) + """ + ) + for column_name, display_name in ( + ("id", "编号"), + ("title", "标题"), + ("created_at", "创建时间"), + ): + conn.execute( + """ + INSERT INTO _jiangchang_columns + (table_name, column_name, display_name, visible, searchable, editable, display_type) + VALUES ('demo_items', ?, ?, 1, 0, 0, 'datetime_unix_seconds') + """, + (column_name, display_name), + ) + conn.commit() + report = validate_display_metadata(conn) + self.assertFalse(report.ok) + self.assertTrue(any("created_at 字段类型应为 INTEGER" in err for err in report.errors)) + finally: + conn.close() + + def test_validator_rejects_editable_timestamp(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: + conn.execute( + "UPDATE _jiangchang_columns SET editable = 1 WHERE table_name = 'task_logs' AND column_name = 'updated_at'" + ) + conn.commit() + report = validate_template_database(conn) + self.assertFalse(report.ok) + self.assertTrue(any("editable 应为 0" in err for err in report.errors)) + finally: + conn.close() + + def test_validator_does_not_flag_amount_as_timestamp(self) -> None: + conn = sqlite3.connect(":memory:") + try: + from db.display_metadata import init_display_metadata + from db.display_metadata_validator import validate_display_metadata + + conn.executescript( + """ + CREATE TABLE demo_items ( + id INTEGER PRIMARY KEY, + amount INTEGER, + count INTEGER + ); + """ + ) + init_display_metadata(conn.cursor()) + conn.execute( + """ + INSERT INTO _jiangchang_tables (table_name, display_name, visible, readonly) + VALUES ('demo_items', '演示', 1, 0) + """ + ) + for column_name, display_name in ( + ("id", "编号"), + ("amount", "数量"), + ("count", "计数"), + ): + conn.execute( + """ + INSERT INTO _jiangchang_columns + (table_name, column_name, display_name, visible, searchable, editable) + VALUES ('demo_items', ?, ?, 1, 1, 1) + """, + (column_name, display_name), + ) + conn.commit() + report = validate_display_metadata(conn) + self.assertFalse(any("amount" in err and "display_type" in err for err in report.errors)) + self.assertFalse(any("count" in err and "display_type" in err for err in report.errors)) + finally: + conn.close() + + def test_trigger_creation_is_idempotent(self) -> None: + with IsolatedDataRoot(): + from db.connection import get_conn, init_db + from db.timestamp_columns import TASK_LOGS_UPDATED_AT_TRIGGER, has_updated_at_trigger + + init_db() + init_db() + conn = get_conn() + try: + self.assertTrue(has_updated_at_trigger(conn.cursor(), "task_logs")) + cur = conn.cursor() + cur.execute( + "SELECT COUNT(*) FROM sqlite_master WHERE type = 'trigger' AND name = ?", + (TASK_LOGS_UPDATED_AT_TRIGGER,), + ) + self.assertEqual(int(cur.fetchone()[0]), 1) + finally: + conn.close() + + +if __name__ == "__main__": + unittest.main()