补充单数据库额外文件拒绝验收

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()

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, \

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')

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

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