359 lines
14 KiB
Python
359 lines
14 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""标准时间字段：Unix 秒级时间戳、默认值、trigger 与元数据。"""
|
||
from __future__ import annotations
|
||
|
||
import sqlite3
|
||
import time
|
||
import unittest
|
||
|
||
from _support import DEMO_ITEMS_TABLE, IsolatedDataRoot, seed_demo_items
|
||
|
||
|
||
class TestTimestampColumns(unittest.TestCase):
|
||
def test_seed_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()
|
||
seed_demo_items(cur)
|
||
conn.commit()
|
||
physical = list_physical_column_names(cur, DEMO_ITEMS_TABLE)
|
||
self.assertEqual(physical[-2:], ["created_at", "updated_at"])
|
||
|
||
info = _table_column_info(cur, DEMO_ITEMS_TABLE)
|
||
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()
|
||
seed_demo_items(cur)
|
||
before = int(time.time())
|
||
cur.execute(
|
||
"INSERT INTO demo_items (title, status) VALUES ('auto-ts', 'draft')"
|
||
)
|
||
conn.commit()
|
||
cur.execute(
|
||
"SELECT created_at, updated_at FROM demo_items 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()
|
||
seed_demo_items(cur)
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO demo_items (title, status, created_at, updated_at)
|
||
VALUES ('update-ts', 'draft', 1000, 1000)
|
||
"""
|
||
)
|
||
row_id = int(cur.lastrowid)
|
||
conn.commit()
|
||
|
||
cur.execute("UPDATE demo_items SET status = 'published' WHERE id = ?", (row_id,))
|
||
conn.commit()
|
||
|
||
cur.execute(
|
||
"SELECT created_at, updated_at FROM demo_items WHERE id = ?",
|
||
(row_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()
|
||
seed_demo_items(cur)
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO demo_items (title, status, created_at, updated_at)
|
||
VALUES ('stable-ts', 'draft', 1234567890, 1234567891)
|
||
"""
|
||
)
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
init_db()
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"SELECT created_at, updated_at FROM demo_items WHERE title = '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()
|
||
seed_demo_items(cur)
|
||
conn.commit()
|
||
cur.execute(
|
||
"""
|
||
SELECT column_name, display_name, display_type, editable, display_order
|
||
FROM _jiangchang_columns
|
||
WHERE table_name = ?
|
||
AND column_name IN ('created_at', 'updated_at')
|
||
""",
|
||
(DEMO_ITEMS_TABLE,),
|
||
)
|
||
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()
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
seed_demo_items(cur)
|
||
seed_demo_items(cur)
|
||
conn.commit()
|
||
cur.execute(
|
||
"""
|
||
SELECT display_type FROM _jiangchang_columns
|
||
WHERE table_name = ? AND column_name = 'created_at'
|
||
""",
|
||
(DEMO_ITEMS_TABLE,),
|
||
)
|
||
self.assertEqual(cur.fetchone()[0], DATETIME_UNIX_SECONDS)
|
||
finally:
|
||
conn.close()
|
||
|
||
def test_insert_uses_database_timestamps(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)
|
||
before = int(time.time())
|
||
cur.execute(
|
||
"INSERT INTO demo_items (title, status) VALUES ('repo-ts', 'draft')"
|
||
)
|
||
row_id = int(cur.lastrowid)
|
||
conn.commit()
|
||
after = int(time.time())
|
||
cur.execute(
|
||
"SELECT created_at, updated_at FROM demo_items WHERE id = ?",
|
||
(row_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:
|
||
seed_demo_items(conn.cursor())
|
||
conn.execute(
|
||
"UPDATE _jiangchang_columns SET display_type = NULL "
|
||
"WHERE table_name = ? AND column_name = 'created_at'",
|
||
(DEMO_ITEMS_TABLE,),
|
||
)
|
||
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:
|
||
seed_demo_items(conn.cursor())
|
||
conn.execute(
|
||
"UPDATE _jiangchang_columns SET editable = 1 "
|
||
"WHERE table_name = ? AND column_name = 'updated_at'",
|
||
(DEMO_ITEMS_TABLE,),
|
||
)
|
||
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 has_updated_at_trigger, updated_at_trigger_name
|
||
|
||
init_db()
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
seed_demo_items(cur)
|
||
seed_demo_items(cur)
|
||
conn.commit()
|
||
self.assertTrue(has_updated_at_trigger(cur, DEMO_ITEMS_TABLE))
|
||
cur.execute(
|
||
"SELECT COUNT(*) FROM sqlite_master WHERE type = 'trigger' AND name = ?",
|
||
(updated_at_trigger_name(DEMO_ITEMS_TABLE),),
|
||
)
|
||
self.assertEqual(int(cur.fetchone()[0]), 1)
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|