From 0ab5acd81c15f358cd11daac5111ddabbe9e054a Mon Sep 17 00:00:00 2001 From: smail Date: Tue, 22 Sep 2026 08:10:15 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A1=A5=E5=85=85=E5=8D=95=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E5=BA=93=E9=A2=9D=E5=A4=96=E6=96=87=E4=BB=B6=E6=8B=92=E7=BB=9D?= =?UTF-8?q?=E9=AA=8C=E6=94=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- test/deployment/test_formal_review_tools.py | 67 +++++++++++++++++++++ test/deployment/test_runtime_cleanup.py | 10 +++ test/tools/deploy_review.py | 3 +- test/tools/runtime_review.py | 5 +- test/tools/verify_review.py | 2 +- 5 files changed, 84 insertions(+), 3 deletions(-) create mode 100644 test/deployment/test_formal_review_tools.py diff --git a/test/deployment/test_formal_review_tools.py b/test/deployment/test_formal_review_tools.py new file mode 100644 index 0000000..1199c76 --- /dev/null +++ b/test/deployment/test_formal_review_tools.py @@ -0,0 +1,67 @@ +"""Formal cloud acceptance tool single-database boundary tests.""" + +from contextlib import closing +from pathlib import Path +import sqlite3 +import sys +import tempfile +import unittest + + +TOOLS = Path(__file__).resolve().parents[1] / 'tools' +sys.path.insert(0, str(TOOLS)) +import deploy_review +import internal_smtp_check +import restore_review +import verify_review + + +class FormalReviewDatabaseTest(unittest.TestCase): + modules = (deploy_review, internal_smtp_check, restore_review, verify_review) + + def make_database(self, root): + data = root / 'data' + data.mkdir() + with closing(sqlite3.connect(data / 'smail.db')) as database: + database.execute('CREATE TABLE marker (id INTEGER)') + + def test_all_formal_tools_accept_one_database(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + self.make_database(root) + for module in self.modules: + with self.subTest(module=module.__name__): + self.assertEqual(module.database_path(root), root / 'data/smail.db') + + def test_all_formal_tools_reject_legacy_database(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + self.make_database(root) + (root / 'data/smail-state.db').touch() + for module in self.modules: + with self.subTest(module=module.__name__), self.assertRaises(AssertionError): + module.database_path(root) + + def test_all_formal_tools_reject_nested_database(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + self.make_database(root) + nested = root / 'data' / 'archive' + nested.mkdir() + (nested / 'old.db').touch() + for module in self.modules: + with self.subTest(module=module.__name__), self.assertRaises(AssertionError): + module.database_path(root) + + def test_all_formal_tools_reject_missing_database_without_creating_one(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / 'data').mkdir() + for module in self.modules: + with self.subTest(module=module.__name__), self.assertRaises(AssertionError): + module.database_path(root) + self.assertEqual(list((root / 'data').iterdir()), []) + + +if __name__ == '__main__': + unittest.main() \ No newline at end of file diff --git a/test/deployment/test_runtime_cleanup.py b/test/deployment/test_runtime_cleanup.py index 660b88a..b258d8f 100644 --- a/test/deployment/test_runtime_cleanup.py +++ b/test/deployment/test_runtime_cleanup.py @@ -61,6 +61,16 @@ class RuntimeCleanupTest(unittest.TestCase): with self.assertRaises(AssertionError): review.database_integrity(root) + def test_nested_database_is_rejected(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / 'data').mkdir() + with closing(sqlite3.connect(root / 'data/smail.db')) as database: + database.execute('CREATE TABLE config_records (id TEXT)') + (root / 'data/archive').mkdir() + (root / 'data/archive/old.db').touch() + with self.assertRaises(AssertionError): + review.database_integrity(root) def test_resources_then_stop_all(self): result = {} with patch.object(review.subprocess, 'check_output', return_value='Result=success') as stats, \ diff --git a/test/tools/deploy_review.py b/test/tools/deploy_review.py index 4db8152..66a4141 100644 --- a/test/tools/deploy_review.py +++ b/test/tools/deploy_review.py @@ -114,7 +114,8 @@ def main(): except BaseException: if installed: subprocess.run(['systemctl', 'stop', 'smail.service'], check=True) - database = database_path(ROOT) + database = ROOT / 'data' / DATABASE_NAME + assert database.is_file() and not database.is_symlink() snapshot(database, backup / ('failed-' + DATABASE_NAME)) snapshot(backup / DATABASE_NAME, database) shutil.chown(database, user='smail', group='smail') diff --git a/test/tools/runtime_review.py b/test/tools/runtime_review.py index fdfdacd..a5775c9 100644 --- a/test/tools/runtime_review.py +++ b/test/tools/runtime_review.py @@ -28,7 +28,10 @@ def database_integrity(root): # 验收器不能创建数据库，也不能允许第二个状态库。 data = root / 'data' assert not (data / 'smail-state.db').exists(), '单数据库实例生成了旧状态库' - assert sorted(path.name for path in data.glob('*.db')) == ['smail.db'] + files = sorted(path for path in data.rglob('*.db') + if path.is_file() or path.is_symlink()) + assert files == [data / 'smail.db'] + assert (data / 'smail.db').is_file() and not (data / 'smail.db').is_symlink() with closing(sqlite3.connect((data / 'smail.db').as_uri() + '?mode=ro', uri=True)) as database: integrity = database.execute('PRAGMA integrity_check').fetchall() assert database.execute("SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='config_records'").fetchone()[0] == 1 diff --git a/test/tools/verify_review.py b/test/tools/verify_review.py index d3394a3..544b021 100644 --- a/test/tools/verify_review.py +++ b/test/tools/verify_review.py @@ -76,7 +76,7 @@ def main(): integrity = connection.execute('PRAGMA integrity_check').fetchall() assert integrity == [('ok',)] summary['database_integrity'][DATABASE_NAME] = 'ok' - summary['migration'] = database.execute('SELECT MAX(version) FROM schema_migration').fetchone()[0] + summary['migration'] = connection.execute('SELECT MAX(version) FROM schema_migration').fetchone()[0] summary['queue_states'] = dict(connection.execute('SELECT state,COUNT(*) FROM outbound_delivery GROUP BY state')) control = Path('/run/smail/admin-control.sock').lstat() assert stat.S_ISSOCK(control.st_mode) and stat.S_IMODE(control.st_mode) == 0o600