完善单数据库识别并对齐统一配置迁移验收

test/deployment/test_formal_review_tools.py

@@ -14,6 +14,9 @@ import deploy_review
 import internal_smtp_check
 import restore_review
 import verify_review
+import runtime_review
+from database_inventory import sqlite_inventory
+from unittest.mock import patch
 
 
 class FormalReviewDatabaseTest(unittest.TestCase):
@@ -63,5 +66,86 @@ class FormalReviewDatabaseTest(unittest.TestCase):
             self.assertEqual(list((root / 'data').iterdir()), [])
 
 
+    def test_extra_sqlite_files_cannot_hide_by_name_or_location(self):
+        for name in ('extra.db', 'STATE.DB', '.hidden.sqlite3', 'nested/extra.sqlite',
+                     '../config/other.db', '../renamed.bin', 'old.db-wal', 'old.sqlite-shm'):
+            with self.subTest(name=name), tempfile.TemporaryDirectory() as directory:
+                root = Path(directory).resolve()
+                self.make_database(root)
+                path = root / 'data' / name
+                path.parent.mkdir(parents=True, exist_ok=True)
+                path.write_bytes(b'SQLite format 3\x00')
+                for module in self.modules:
+                    with self.subTest(module=module.__name__), self.assertRaises(AssertionError):
+                        module.database_path(root)
+                with self.assertRaises(AssertionError):
+                    runtime_review.database_integrity(root)
+
+    def test_expected_sqlite_sidecars_are_not_second_databases(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory).resolve()
+            self.make_database(root)
+            for suffix in ('-wal', '-shm', '-journal'):
+                (root / 'data' / ('smail.db' + suffix)).touch()
+            for module in self.modules:
+                self.assertEqual(module.database_path(root), root / 'data/smail.db')
+
+    def test_sql_mode_rejects_renamed_database_and_orphan_sidecars(self):
+        for name in ('renamed.bin', 'data/.hidden.sqlite', 'data/old.db-wal'):
+            with self.subTest(name=name), tempfile.TemporaryDirectory() as directory:
+                root = Path(directory).resolve()
+                (root / 'data').mkdir()
+                self.assertEqual(sqlite_inventory(root, []), [])
+                (root / name).write_bytes(b'SQLite format 3\x00')
+                with self.assertRaises(AssertionError):
+                    sqlite_inventory(root, [])
+
+    def test_directory_links_are_rejected_before_reading_their_contents(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory).resolve()
+            self.make_database(root)
+            linked = root / 'data' / 'linked'
+            linked.mkdir()
+            original = Path.is_symlink
+            with patch.object(Path, 'is_symlink', lambda path: path == linked or original(path)):
+                with self.assertRaises(AssertionError):
+                    sqlite_inventory(root, [root / 'data/smail.db'])
+
+    def test_snapshot_and_restore_counts_close_their_connections(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory).resolve()
+            self.make_database(root)
+            source = root / 'data/smail.db'
+            target = root / 'snapshot.db'
+            deploy_review.snapshot(source, target)
+            moved = root / 'snapshot-moved.db'
+            target.rename(moved)
+            with closing(sqlite3.connect(moved)) as database:
+                self.assertEqual(database.execute('PRAGMA integrity_check').fetchone(), ('ok',))
+            isolated = root / 'isolated'
+            isolated.mkdir()
+            self.make_database(isolated)
+            self.assertEqual(restore_review.database_counts(isolated), {'smail.db': {'marker': 0}})
+            (isolated / 'data/smail.db').rename(isolated / 'data/moved.db')
+
+
+    def test_deployed_schema_requires_unified_configuration_history(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory).resolve()
+            self.make_database(root)
+            with closing(sqlite3.connect(root / 'data/smail.db')) as database:
+                database.executescript('CREATE TABLE schema_migration (version INTEGER); '
+                                      'INSERT INTO schema_migration VALUES (23); '
+                                      'CREATE TABLE outbound_delivery (state TEXT); '
+                                      "INSERT INTO outbound_delivery VALUES ('pending');")
+                with self.assertRaises(AssertionError):
+                    deploy_review.database_state(root)
+                database.executescript('INSERT INTO schema_migration VALUES (24);')
+                with self.assertRaises(sqlite3.OperationalError):
+                    deploy_review.database_state(root)
+                database.executescript('CREATE TABLE config_records (id TEXT);')
+            self.assertEqual(deploy_review.database_state(root), (24, {'pending': 1}))
+
+
 if __name__ == '__main__':
     unittest.main()

test/tools/database_inventory.py

@@ -0,0 +1,31 @@
+from pathlib import Path
+import stat
+
+
+def sqlite_inventory(root, expected):
+    root = Path(root)
+    assert root.is_absolute() and root.resolve(strict=True) == root
+    assert (root / 'data').is_dir()
+    expected = set(expected)
+    assert all(path.is_relative_to(root) for path in expected)
+    seen = set()
+    pending = [root]
+    while pending:
+        path = pending.pop()
+        assert not path.is_symlink() and not getattr(path, 'is_junction', lambda: False)()
+        mode = path.stat().st_mode
+        if stat.S_ISDIR(mode):
+            pending.extend(path.iterdir())
+            continue
+        assert stat.S_ISREG(mode)
+        name = path.name.lower()
+        if name.endswith(('-wal', '-shm', '-journal')):
+            assert path.with_name(path.name.rsplit('-', 1)[0]) in expected
+            continue
+        with path.open('rb') as stream:
+            header = stream.read(16)
+        if header == b'SQLite format 3\x00' or name.endswith('.db') or '.sqlite' in name:
+            assert path in expected
+            seen.add(path)
+    assert seen == expected
+    return sorted(seen)

test/tools/deploy_review.py

@@ -7,9 +7,12 @@ import os
 from pathlib import Path
 import shutil
 import sqlite3
+from contextlib import closing
 import subprocess
 import time
 
+from database_inventory import sqlite_inventory
+
 
 ROOT = Path('/opt/smail')
 STAGING = Path('/var/tmp/smail-review-20260917')
@@ -27,8 +30,7 @@ def digest(path):
 
 def database_path(root):
     data = root / 'data'
-    files = sorted(path for path in data.rglob('*.db')
-                   if path.is_file() or path.is_symlink())
+    files = sqlite_inventory(root, [data / 'smail.db'])
     expected = data / DATABASE_NAME
     assert files == [expected], '正式验收要求运行目录仅保留 data/smail.db'
     assert expected.is_file() and not expected.is_symlink(), '活动数据库不能是符号链接'
@@ -36,19 +38,31 @@ def database_path(root):
 
 
 def snapshot(source, target):
-    with sqlite3.connect(f'file:{source}?mode=ro', uri=True) as original:
-        with sqlite3.connect(target) as copied:
+    with closing(sqlite3.connect(source.as_uri() + '?mode=ro', uri=True)) as original:
+        with closing(sqlite3.connect(target)) as copied:
             original.backup(copied)
             assert copied.execute('PRAGMA integrity_check').fetchone()[0] == 'ok'
     target.chmod(0o600)
 
 
+
+def database_state(root):
+    database = database_path(root)
+    with closing(sqlite3.connect(database.as_uri() + '?mode=ro', uri=True)) as connection:
+        assert connection.execute('PRAGMA integrity_check').fetchall() == [('ok',)]
+        version = connection.execute('SELECT MAX(version) FROM schema_migration').fetchone()[0]
+        assert version == 24
+        connection.execute('SELECT COUNT(*) FROM config_records').fetchone()
+        states = dict(connection.execute('SELECT state,COUNT(*) FROM outbound_delivery GROUP BY state'))
+    return version, states
+
+
 def wait_ready():
     for _ in range(60):
         result = subprocess.run([
             'curl', '--silent', '--fail', '--max-time', '2', '--noproxy', '*',
             '--cacert', str(ROOT / 'ssl/custom.crt'), '--resolve',
-            'mail.solit.cn:443:127.0.0.1', 'https://mail.solit.cn/api/ping',
+            'web.solit.cn:443:127.0.0.1', 'https://web.solit.cn/api/ping',
         ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
         if result.returncode == 0 and run('systemctl', 'is-active', 'smail.service') == 'active':
             return
@@ -98,12 +112,7 @@ def main():
             assert digest(ROOT / name) == checksum
         subprocess.run(['systemctl', 'start', 'smail.service'], check=True)
         wait_ready()
-        database_path(ROOT)
-        with sqlite3.connect(f'file:{database}?mode=ro', uri=True) as connection:
-            assert connection.execute('PRAGMA integrity_check').fetchone()[0] == 'ok'
-            version = connection.execute('SELECT MAX(version) FROM schema_migration').fetchone()[0]
-            assert version == 23
-            states = dict(connection.execute('SELECT state,COUNT(*) FROM outbound_delivery GROUP BY state'))
+        version, states = database_state(ROOT)
         result = {'backup': str(backup), 'archive_sha256': digest(archive),
                   'artifacts': expected, 'migration': version, 'queue_states': states,
                   'service': run('systemctl', 'is-active', 'smail.service'),

test/tools/internal_smtp_check.py

@@ -5,6 +5,7 @@ import hashlib
 import json
 import smtplib
 import sqlite3
+from contextlib import closing
 import ssl
 import subprocess
 import time
@@ -13,6 +14,8 @@ from email.policy import SMTP
 from email.utils import formatdate
 from pathlib import Path
 
+from database_inventory import sqlite_inventory
+
 
 ROOT = Path('/opt/smail')
 DATABASE_NAME = 'smail.db'
@@ -24,8 +27,7 @@ ATTACHMENT = '中文附件内容，内部收信验收。'.encode('utf-8')
 
 def database_path(root):
     data = root / 'data'
-    files = sorted(path for path in data.rglob('*.db')
-                   if path.is_file() or path.is_symlink())
+    files = sqlite_inventory(root, [data / 'smail.db'])
     expected = data / DATABASE_NAME
     assert files == [expected], '正式验收要求运行目录仅保留 data/smail.db'
     assert expected.is_file() and not expected.is_symlink(), '活动数据库不能是符号链接'
@@ -79,12 +81,12 @@ def main():
     backup = Path('/var/backups') / f'smail-internal-smtp-{RUN}'
     backup.mkdir(mode=0o700)
     database_path_value = database_path(ROOT)
-    with sqlite3.connect(f'file:{database_path_value}?mode=ro', uri=True) as source:
-        with sqlite3.connect(backup / DATABASE_NAME) as target:
+    with closing(sqlite3.connect(database_path_value.as_uri() + '?mode=ro', uri=True)) as source:
+        with closing(sqlite3.connect(backup / DATABASE_NAME)) as target:
             source.backup(target)
             assert target.execute('PRAGMA integrity_check').fetchone()[0] == 'ok'
     (backup / DATABASE_NAME).chmod(0o600)
-    with sqlite3.connect(f'file:{database_path_value}?mode=ro', uri=True) as database:
+    with closing(sqlite3.connect(database_path_value.as_uri() + '?mode=ro', uri=True)) as database:
         database.row_factory = sqlite3.Row
         policy = json.loads((ROOT / 'config/email.json').read_text(encoding='utf-8')).get('policy', {})
         assert policy.get('spamFilterLevel', 0) == 0, 'This fixture requires the current unfiltered policy'

test/tools/restore_review.py

@@ -7,10 +7,13 @@ import os
 from pathlib import Path
 import shutil
 import sqlite3
+from contextlib import closing
 import subprocess
 import tarfile
 import time
 
+from database_inventory import sqlite_inventory
+
 
 DATABASE_NAME = 'smail.db'
 
@@ -21,8 +24,7 @@ def run(*args):
 
 def database_path(root):
     data = root / 'data'
-    files = sorted(path for path in data.rglob('*.db')
-                   if path.is_file() or path.is_symlink())
+    files = sqlite_inventory(root, [data / 'smail.db'])
     expected = data / DATABASE_NAME
     assert files == [expected], '正式验收要求运行目录仅保留 data/smail.db'
     assert expected.is_file() and not expected.is_symlink(), '活动数据库不能是符号链接'
@@ -32,7 +34,7 @@ def database_path(root):
 def database_counts(root):
     counts = {}
     database_path_value = database_path(root)
-    with sqlite3.connect(f'file:{database_path_value}?mode=ro', uri=True) as database:
+    with closing(sqlite3.connect(database_path_value.as_uri() + '?mode=ro', uri=True)) as database:
         assert database.execute('PRAGMA integrity_check').fetchone()[0] == 'ok'
         tables = database.execute("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'").fetchall()
         counts[DATABASE_NAME] = {
@@ -95,7 +97,7 @@ def main():
                 probe = subprocess.run([
                     'nsenter', '--target', pid, '--net', 'curl', '--silent', '--fail', '--max-time', '2',
                     '--noproxy', '*', '--cacert', str(restored / 'ssl/custom.crt'),
-                    '--resolve', 'mail.solit.cn:443:127.0.0.1', 'https://mail.solit.cn/api/ping',
+                    '--resolve', 'web.solit.cn:443:127.0.0.1', 'https://web.solit.cn/api/ping',
                 ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
                 if probe.returncode == 0:
                     ready = True
@@ -124,9 +126,9 @@ with smtplib.SMTP('127.0.0.1',25,timeout=20) as client:
 '''
         subprocess.run(['nsenter', '--target', pid, '--net', 'python3', '-c', probe,
                         str(restored / 'ssl/custom.crt'), stamp], check=True)
-        with sqlite3.connect(f'file:{restored / "data/smail.db"}?mode=ro', uri=True) as database:
+        with closing(sqlite3.connect(f'file:{restored / "data/smail.db"}?mode=ro', uri=True)) as database:
             assert database.execute('SELECT COUNT(*) FROM email WHERE msg_id=?', ('restore-' + stamp + '@review.invalid',)).fetchone()[0] == 1
-        with sqlite3.connect('file:/opt/smail/data/smail.db?mode=ro', uri=True) as database:
+        with closing(sqlite3.connect('file:/opt/smail/data/smail.db?mode=ro', uri=True)) as database:
             assert database.execute('SELECT COUNT(*) FROM email WHERE msg_id=?', ('restore-' + stamp + '@review.invalid',)).fetchone()[0] == 0
         result.update(ready=True, isolated=True, restored_smtp=True, live_pid_unchanged=live_pid == run(
             'systemctl', 'show', 'smail.service', '--property=MainPID', '--value'))

test/tools/runtime_review.py

@@ -13,6 +13,8 @@ import sqlite3
 import subprocess
 import time
 
+from database_inventory import sqlite_inventory
+
 
 def run(*args):
     return subprocess.check_output(args, text=True).strip()
@@ -28,8 +30,7 @@ def database_integrity(root):
     # 验收器不能创建数据库，也不能允许第二个状态库。
     data = root / 'data'
     assert not (data / 'smail-state.db').exists(), '单数据库实例生成了旧状态库'
-    files = sorted(path for path in data.rglob('*.db')
-                   if path.is_file() or path.is_symlink())
+    files = sqlite_inventory(root, [data / 'smail.db'])
     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:
@@ -173,6 +174,9 @@ def main():
         shutil.chown(directory, user='smail', group='smail')
     shutil.copy2(binary, root / 'smail')
     (root / 'smail').chmod(0o755)
+    shutil.copy2(Path(__file__).with_name('database_inventory.py'), target / 'database_inventory.py')
+    (target / 'database_inventory.py').chmod(0o640)
+    shutil.chown(target / 'database_inventory.py', user='root', group='smail')
     shutil.copy2(Path(__file__).with_name('runtime_probe.py'), target / 'runtime_probe.py')
     (target / 'runtime_probe.py').chmod(0o640)
     shutil.chown(target / 'runtime_probe.py', user='root', group='smail')
@@ -351,7 +355,7 @@ def main():
             result['probe'] = json.loads((target / 'probe-result.json').read_text())
         assert probe.returncode == 0, '隔离验收失败，请检查保留的 probe.log'
         assert result['probe']['passed']
-        with sqlite3.connect(f'file:{live / "data/smail.db"}?mode=ro', uri=True) as database:
+        with closing(sqlite3.connect(f'file:{live / "data/smail.db"}?mode=ro', uri=True)) as database:
             message_id = result['probe']['smtp']['message_id']
             assert database.execute('SELECT COUNT(*) FROM email WHERE msg_id=?', (message_id,)).fetchone()[0] == 0
         result['live_message_absent'] = True

test/tools/runtime_sql.py

@@ -7,6 +7,8 @@ import re
 import secrets
 import subprocess
 
+from database_inventory import sqlite_inventory
+
 
 def command(args, sql, env=None):
     result = subprocess.run(args, input=sql, text=True, capture_output=True,
@@ -75,8 +77,7 @@ class SQLDatabase:
 
     def integrity(self, root):
         # 配置历史和业务表必须位于所选 SQL 库，运行目录不能悄悄创建 SQLite。
-        assert not list((root / 'data').rglob('*.db'))
-        assert not list((root / 'data').rglob('*.sqlite*'))
+        sqlite_inventory(root, [])
         count = int(self.query('SELECT COUNT(*) FROM config_records'))
         assert count > 0
         return {'type': self.kind, 'database': self.settings['name'],

test/tools/verify_review.py

@@ -6,9 +6,12 @@ import json
 import os
 from pathlib import Path
 import sqlite3
+from contextlib import closing
 import stat
 import subprocess
 
+from database_inventory import sqlite_inventory
+
 
 DATABASE_NAME = 'smail.db'
 
@@ -19,8 +22,7 @@ def run(*args):
 
 def database_path(root):
     data = root / 'data'
-    files = sorted(path for path in data.rglob('*.db')
-                   if path.is_file() or path.is_symlink())
+    files = sqlite_inventory(root, [data / 'smail.db'])
     expected = data / DATABASE_NAME
     assert files == [expected], '正式验收要求运行目录仅保留 data/smail.db'
     assert expected.is_file() and not expected.is_symlink(), '活动数据库不能是符号链接'
@@ -72,7 +74,7 @@ def main():
     summary['admin_version'] = run(str(root / 'smail-admin'), 'version')
     summary['database_integrity'] = {}
     database = database_path(root)
-    with sqlite3.connect(f'file:{database}?mode=ro', uri=True) as connection:
+    with closing(sqlite3.connect(database.as_uri() + '?mode=ro', uri=True)) as connection:
         integrity = connection.execute('PRAGMA integrity_check').fetchall()
         assert integrity == [('ok',)]
         summary['database_integrity'][DATABASE_NAME] = 'ok'
@@ -83,11 +85,11 @@ def main():
     assert control.st_uid == database.stat().st_uid
     summary['control_socket_mode'] = oct(stat.S_IMODE(control.st_mode))
     curl = ['curl', '--silent', '--show-error', '--fail', '--max-time', '10', '--noproxy', '*',
-            '--cacert', str(root / 'ssl/custom.crt'), '--resolve', 'mail.solit.cn:443:127.0.0.1']
-    run(*curl, 'https://mail.solit.cn/api/ping')
+            '--cacert', str(root / 'ssl/custom.crt'), '--resolve', 'web.solit.cn:443:127.0.0.1']
+    run(*curl, 'https://web.solit.cn/api/ping')
     summary['unauthenticated'] = {}
     for endpoint in ('/api/v1/user/info', '/api/v1/email/list'):
-        response = json.loads(run(*curl, '-X', 'POST', 'https://mail.solit.cn' + endpoint))
+        response = json.loads(run(*curl, '-X', 'POST', 'https://web.solit.cn' + endpoint))
         assert response['errorNo'] == 403 and not response.get('data')
         summary['unauthenticated'][endpoint] = response['errorNo']
     entries = run('journalctl', '-u', 'smail.service', '--since', args.since, '--output=json', '--no-pager')