diff --git a/test/tools/deploy_review.py b/test/tools/deploy_review.py index 764adf2..4db8152 100644 --- a/test/tools/deploy_review.py +++ b/test/tools/deploy_review.py @@ -1,4 +1,4 @@ -"""仅在 47 服务器备份并升级 Smail，保留双库和失败证据。""" +"""仅在 47 服务器备份并升级 Smail，保留单库和失败证据。""" import argparse import hashlib @@ -13,7 +13,7 @@ import time ROOT = Path('/opt/smail') STAGING = Path('/var/tmp/smail-review-20260917') -DATABASES = ('smail.db', 'smail-state.db') +DATABASE_NAME = 'smail.db' def run(*args): @@ -25,6 +25,16 @@ def digest(path): return hashlib.file_digest(stream, 'sha256').hexdigest() +def database_path(root): + data = root / 'data' + files = sorted(path for path in data.rglob('*.db') + if path.is_file() or path.is_symlink()) + expected = data / DATABASE_NAME + assert files == [expected], '正式验收要求运行目录仅保留 data/smail.db' + assert expected.is_file() and not expected.is_symlink(), '活动数据库不能是符号链接' + return expected + + def snapshot(source, target): with sqlite3.connect(f'file:{source}?mode=ro', uri=True) as original: with sqlite3.connect(target) as copied: @@ -65,6 +75,7 @@ def main(): assert header[:5] == b'\x7fELF\x02' and header[18:20] == b'\x3e\x00' assert digest(path) == checksum assert not (ROOT / name).is_symlink() + database = database_path(ROOT) backup = Path('/var/backups') / time.strftime('smail-before-outbox-%Y%m%d-%H%M%S') backup.mkdir(mode=0o700) @@ -76,10 +87,8 @@ def main(): subprocess.run(['tar', '-czf', str(archive), '-C', '/opt', 'smail'], check=True) subprocess.run(['tar', '-dzf', str(archive), '-C', '/opt'], check=True) archive.chmod(0o600) - for name in DATABASES: - source = ROOT / 'data' / name - assert source.resolve() == source and not source.is_symlink() - snapshot(source, backup / name) + assert database.resolve() == database and not database.is_symlink() + snapshot(database, backup / DATABASE_NAME) for name in expected: shutil.copy2(ROOT / name, backup / name) installed = True @@ -89,13 +98,12 @@ def main(): assert digest(ROOT / name) == checksum subprocess.run(['systemctl', 'start', 'smail.service'], check=True) wait_ready() - for name in DATABASES: - with sqlite3.connect(f'file:{ROOT / "data" / name}?mode=ro', uri=True) as database: - assert database.execute('PRAGMA integrity_check').fetchone()[0] == 'ok' - with sqlite3.connect(f'file:{ROOT / "data/smail.db"}?mode=ro', uri=True) as database: - version = database.execute('SELECT MAX(version) FROM schema_migration').fetchone()[0] + 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(database.execute('SELECT state,COUNT(*) FROM outbound_delivery GROUP BY state')) + states = dict(connection.execute('SELECT state,COUNT(*) FROM outbound_delivery GROUP BY state')) result = {'backup': str(backup), 'archive_sha256': digest(archive), 'artifacts': expected, 'migration': version, 'queue_states': states, 'service': run('systemctl', 'is-active', 'smail.service'), @@ -106,10 +114,10 @@ def main(): except BaseException: if installed: subprocess.run(['systemctl', 'stop', 'smail.service'], check=True) - for name in DATABASES: - snapshot(ROOT / 'data' / name, backup / ('failed-' + name)) - snapshot(backup / name, ROOT / 'data' / name) - shutil.chown(ROOT / 'data' / name, user='smail', group='smail') + database = database_path(ROOT) + snapshot(database, backup / ('failed-' + DATABASE_NAME)) + snapshot(backup / DATABASE_NAME, database) + shutil.chown(database, user='smail', group='smail') for name in expected: shutil.copy2(backup / name, ROOT / name) subprocess.run(['systemctl', 'start', 'smail.service'], check=True) diff --git a/test/tools/internal_smtp_check.py b/test/tools/internal_smtp_check.py index fc7749d..373a427 100644 --- a/test/tools/internal_smtp_check.py +++ b/test/tools/internal_smtp_check.py @@ -15,12 +15,23 @@ from pathlib import Path ROOT = Path('/opt/smail') +DATABASE_NAME = 'smail.db' RUN = time.strftime('%Y%m%d-%H%M%S') DOMAIN = 'solit.cn' BODY = '内部25端口中文正文验收：邮件必须真实入库，并且只属于指定收件人。' 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()) + expected = data / DATABASE_NAME + assert files == [expected], '正式验收要求运行目录仅保留 data/smail.db' + assert expected.is_file() and not expected.is_symlink(), '活动数据库不能是符号链接' + return expected + + def connect(): client = smtplib.SMTP('127.0.0.1', 25, timeout=30) client.ehlo('internal-check.invalid') @@ -64,16 +75,16 @@ def deliver(label, recipients, multipart=False): def main(): assert ROOT.is_dir() and Path('/etc/systemd/system/smail.service').is_file() assert subprocess.check_output(['systemctl', 'is-active', 'smail.service'], text=True).strip() == 'active' - # 只读连接真实业务库，备份两份数据库，不清除前一轮故障证据。 + # 只读连接真实业务库，备份唯一数据库，不清除前一轮故障证据。 backup = Path('/var/backups') / f'smail-internal-smtp-{RUN}' backup.mkdir(mode=0o700) - for name in ('smail.db', 'smail-state.db'): - with sqlite3.connect(f'file:{ROOT / "data" / name}?mode=ro', uri=True) as source: - with sqlite3.connect(backup / name) as target: - source.backup(target) - assert target.execute('PRAGMA integrity_check').fetchone()[0] == 'ok' - (backup / name).chmod(0o600) - with sqlite3.connect(f'file:{ROOT / "data/smail.db"}?mode=ro', uri=True) as database: + 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: + 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: 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' diff --git a/test/tools/restore_review.py b/test/tools/restore_review.py index f4b5c44..9c79ba5 100644 --- a/test/tools/restore_review.py +++ b/test/tools/restore_review.py @@ -12,20 +12,33 @@ import tarfile import time +DATABASE_NAME = 'smail.db' + + def run(*args): return subprocess.check_output(args, text=True).strip() +def database_path(root): + data = root / 'data' + files = sorted(path for path in data.rglob('*.db') + if path.is_file() or path.is_symlink()) + expected = data / DATABASE_NAME + assert files == [expected], '正式验收要求运行目录仅保留 data/smail.db' + assert expected.is_file() and not expected.is_symlink(), '活动数据库不能是符号链接' + return expected + + def database_counts(root): counts = {} - for name in ('smail.db', 'smail-state.db'): - with sqlite3.connect(f'file:{root / "data" / name}?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[name] = { - table: database.execute('SELECT COUNT(*) FROM "' + table.replace('"', '""') + '"').fetchone()[0] - for (table,) in tables - } + database_path_value = database_path(root) + with sqlite3.connect(f'file:{database_path_value}?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] = { + table: database.execute('SELECT COUNT(*) FROM "' + table.replace('"', '""') + '"').fetchone()[0] + for (table,) in tables + } return counts diff --git a/test/tools/verify_review.py b/test/tools/verify_review.py index c2f6ae1..d3394a3 100644 --- a/test/tools/verify_review.py +++ b/test/tools/verify_review.py @@ -10,10 +10,23 @@ import stat import subprocess +DATABASE_NAME = 'smail.db' + + def run(*args): return subprocess.check_output(args, text=True).strip() +def database_path(root): + data = root / 'data' + files = sorted(path for path in data.rglob('*.db') + if path.is_file() or path.is_symlink()) + expected = data / DATABASE_NAME + assert files == [expected], '正式验收要求运行目录仅保留 data/smail.db' + assert expected.is_file() and not expected.is_symlink(), '活动数据库不能是符号链接' + return expected + + def main(): parser = argparse.ArgumentParser() parser.add_argument('--results', nargs='+', default=[]) @@ -58,17 +71,16 @@ def main(): summary[name + '_sha256'] = actual summary['admin_version'] = run(str(root / 'smail-admin'), 'version') summary['database_integrity'] = {} - for name in ('smail.db', 'smail-state.db'): - with sqlite3.connect(f'file:{root / "data" / name}?mode=ro', uri=True) as database: - integrity = database.execute('PRAGMA integrity_check').fetchall() - assert integrity == [('ok',)] - summary['database_integrity'][name] = 'ok' - with sqlite3.connect(f'file:{root / "data/smail.db"}?mode=ro', uri=True) as database: + database = database_path(root) + with sqlite3.connect(f'file:{database}?mode=ro', uri=True) as connection: + 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['queue_states'] = dict(database.execute('SELECT state,COUNT(*) FROM outbound_delivery GROUP BY state')) + 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 - assert control.st_uid == (root / 'data/smail.db').stat().st_uid + 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']