ila
38d8975325
- scripts/ops.py：install-check（依赖与资源版本、客户端不得低于服务端）、init-database（建空库 并提示最小权限）、backup（全库 dump ＋ manifest，不含凭据）、restore（默认只写空库、 --confirm 必需、覆盖需 --force、拒绝系统库与带 CREATE DATABASE/USE 的 dump、比对源库校验和）、 verify（完整性 ＋ 可选两账号接口闭环）、smoke（干净实例上建两个演练账号走通学习闭环） - scripts/bench.py：写明规模的人造数据集性能测量，记录数据量、机器与 p50/p95 - tests/test_lexgo_ops.py：版本解析、库名白名单、dump 安全性、manifest 字段白名单等无库测试 - Wiki 新增 Deployment-and-Operations 页面与 wiki-docs.json 映射（含升级回滚与已知限制） - 更新 Architecture、Business-Rules、Local-Development、Product-Requirements、Home 与 README/AGENTS；本单无 schema 与接口变化
91 lines
4.0 KiB
Python
91 lines
4.0 KiB
Python
"""Unit tests for the operations helper rules that must hold without a database.
|
|
|
|
The database paths (backup, restore, verify) are exercised by the #15 drill against real MySQL;
|
|
what is checked here is the logic that decides whether an operation is allowed at all.
|
|
"""
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "scripts"))
|
|
|
|
import ops # noqa: E402
|
|
|
|
|
|
class VersionParsingTests(unittest.TestCase):
|
|
def test_reads_the_version_after_distrib(self):
|
|
# A client banner's first number is the protocol version, not the server version.
|
|
self.assertEqual(ops.version_of("mysql Ver 14.14 Distrib 5.7.38, for Win64 (x86_64)"), (5, 7, 38))
|
|
self.assertEqual(ops.version_of("mysql Ver 8.4.3 for Win64 on x86_64"), (8, 4, 3))
|
|
self.assertIsNone(ops.version_of(""))
|
|
self.assertIsNone(ops.version_of("no version here"))
|
|
|
|
def test_compares_client_and_server_versions(self):
|
|
self.assertLess(ops.version_of("Ver 14.14 Distrib 5.7.38"), ops.version_of("8.4.3"))
|
|
self.assertGreaterEqual(ops.version_of("Ver 8.0.36"), ops.version_of("8.4.3".replace("8.4.3", "8.0.36")))
|
|
self.assertGreaterEqual(ops.version_of("8.4.3"), ops.version_of("8.4.3"))
|
|
|
|
|
|
class DatabaseNameTests(unittest.TestCase):
|
|
def test_accepts_lexgo_names(self):
|
|
for name in ("lexgo_dev", "lexgo_prod", "lexgo_restore_drill"):
|
|
self.assertEqual(ops.validate_database_name(name, "测试"), name)
|
|
|
|
def test_refuses_system_and_unrelated_names(self):
|
|
for name in ("mysql", "information_schema", "performance_schema", "sys", "app_prod", ""):
|
|
with self.assertRaises(SystemExit):
|
|
ops.validate_database_name(name, "测试")
|
|
|
|
|
|
class DumpSafetyTests(unittest.TestCase):
|
|
"""A dump that carries CREATE DATABASE/USE would restore into the wrong schema."""
|
|
|
|
def _write_dump(self, body):
|
|
import gzip
|
|
import tempfile
|
|
|
|
handle = tempfile.NamedTemporaryFile(delete=False, suffix=".sql.gz")
|
|
handle.close()
|
|
with gzip.open(handle.name, "wb") as target:
|
|
target.write(body.encode("utf-8"))
|
|
self.addCleanup(Path(handle.name).unlink)
|
|
return Path(handle.name)
|
|
|
|
def test_flags_a_dump_that_switches_database(self):
|
|
dump = self._write_dump("-- MySQL dump\nCREATE DATABASE /*!32312 IF NOT EXISTS*/ `lexgo_dev`;\nUSE `lexgo_dev`;\nCREATE TABLE t (id INT);\n")
|
|
self.assertTrue(ops.dump_targets_instead_of_source(dump))
|
|
|
|
def test_flags_a_dump_with_use_only(self):
|
|
dump = self._write_dump("-- MySQL dump\nUSE `lexgo_dev`;\nCREATE TABLE t (id INT);\n")
|
|
self.assertTrue(ops.dump_targets_instead_of_source(dump))
|
|
|
|
def test_accepts_a_table_only_dump(self):
|
|
dump = self._write_dump("-- MySQL dump\nDROP TABLE IF EXISTS `lexgo_terms`;\nCREATE TABLE `lexgo_terms` (id INT);\nINSERT INTO `lexgo_terms` VALUES (1);\n")
|
|
self.assertFalse(ops.dump_targets_instead_of_source(dump))
|
|
|
|
|
|
class ManifestTests(unittest.TestCase):
|
|
def test_manifest_keys_never_name_a_credential(self):
|
|
"""The manifest is written to disk, so its field names are checked, not its prose."""
|
|
import re
|
|
|
|
source = (ROOT / "scripts" / "ops.py").read_text(encoding="utf-8")
|
|
block = source[source.index(" manifest = {"):source.index('(out_dir / "manifest.json")')]
|
|
keys = re.findall(r'^\s+"([a-z_]+)":', block, re.M)
|
|
self.assertIn("schema_version", keys)
|
|
self.assertIn("row_counts", keys)
|
|
for key in keys:
|
|
for banned in ("password", "secret", "token", "credential"):
|
|
self.assertNotIn(banned, key.lower(), key)
|
|
|
|
def test_audit_tables_declare_no_sensitive_column(self):
|
|
"""The integrity check must keep naming the columns that may not appear in the audit logs."""
|
|
self.assertTrue({"password", "token", "body", "content"} <= ops.AUDIT_BANNED_COLUMNS)
|
|
self.assertIn("lexgo_login_logs", ops.TABLES)
|
|
self.assertIn("lexgo_operation_logs", ops.TABLES)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|