112 lines
5.2 KiB
Python
112 lines
5.2 KiB
Python
"""迁移验收工具的只读、范围和进程证据边界测试，不访问云服务器。"""
|
|
|
|
from contextlib import closing
|
|
import json
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'tools'))
|
|
import runtime_migration as migration
|
|
|
|
|
|
class RuntimeMigrationTest(unittest.TestCase):
|
|
def test_pending_restart_is_completed_before_migration(self):
|
|
client = Mock()
|
|
client.success.side_effect = [
|
|
{'activeGeneration': 5, 'pendingRestart': ['database']}, {'status': 'accepted'},
|
|
]
|
|
with patch.object(migration, 'wait_ready') as wait:
|
|
self.assertEqual(migration.clear_pending_restart(client), 5)
|
|
wait.assert_called_once_with(client, 5)
|
|
self.assertEqual(client.success.call_args_list[1].args, ('/api/v1/system/restart',))
|
|
|
|
def test_no_pending_change_does_not_restart(self):
|
|
client = Mock()
|
|
client.success.return_value = {'activeGeneration': 5, 'pendingRestart': []}
|
|
with patch.object(migration, 'wait_ready') as wait:
|
|
self.assertEqual(migration.clear_pending_restart(client), 5)
|
|
wait.assert_not_called()
|
|
self.assertEqual(client.success.call_count, 1)
|
|
|
|
def test_missing_database_is_not_created(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
missing = Path(directory) / 'missing.db'
|
|
with self.assertRaises(sqlite3.OperationalError):
|
|
migration.database_fingerprint(missing)
|
|
self.assertFalse(missing.exists())
|
|
|
|
def test_fingerprint_detects_changes(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / 'source.db'
|
|
with closing(sqlite3.connect(path)) as database:
|
|
database.execute('CREATE TABLE item (value TEXT)')
|
|
database.commit()
|
|
first = migration.database_fingerprint(path)
|
|
self.assertEqual(first, migration.database_fingerprint(path))
|
|
with closing(sqlite3.connect(path)) as database:
|
|
database.execute("INSERT INTO item VALUES ('changed')")
|
|
database.commit()
|
|
self.assertNotEqual(first, migration.database_fingerprint(path))
|
|
|
|
def test_history_requires_payload_and_digest_equality(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
(root / 'data').mkdir()
|
|
for name in ('smail.db', 'migrated.db'):
|
|
with closing(sqlite3.connect(root / 'data' / name)) as database:
|
|
database.execute('CREATE TABLE config_records (kind TEXT, id TEXT, payload BLOB, digest TEXT)')
|
|
database.execute('INSERT INTO config_records VALUES (?, ?, ?, ?)', ('backup', 'one', b'payload', 'digest'))
|
|
database.commit()
|
|
self.assertEqual(migration.verify_history_preserved(root)['source_records'], 1)
|
|
with closing(sqlite3.connect(root / 'data/migrated.db')) as database:
|
|
database.execute("UPDATE config_records SET payload = 'changed'")
|
|
database.commit()
|
|
with self.assertRaises(AssertionError):
|
|
migration.verify_history_preserved(root)
|
|
|
|
def test_integrity_distinguishes_active_and_offline_files(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
(root / 'config').mkdir()
|
|
(root / 'data').mkdir()
|
|
(root / 'config/config.json').write_text(json.dumps({
|
|
'database': {'type': 'sqlite', 'localRef': 'migrated.db'},
|
|
}))
|
|
with closing(sqlite3.connect(root / 'data/migrated.db')) as database:
|
|
database.execute('CREATE TABLE config_records (id INTEGER)')
|
|
database.execute('INSERT INTO config_records VALUES (1)')
|
|
database.commit()
|
|
(root / 'data/smail.db').touch()
|
|
(root / 'data/occupied.db').touch()
|
|
result = migration.migration_integrity(root)
|
|
self.assertEqual(result['active_database'], 'migrated.db')
|
|
self.assertEqual(result['config_records'], 1)
|
|
(root / 'data/smail-state.db').touch()
|
|
with self.assertRaises(AssertionError):
|
|
migration.migration_integrity(root)
|
|
|
|
def test_system_service_is_rejected_before_command(self):
|
|
with patch.object(migration.subprocess, 'check_output') as command:
|
|
for name in ('smail.service', 'mysql.service', 'smail-runtime-review-anything'):
|
|
with self.assertRaises(AssertionError):
|
|
migration.service_state(name)
|
|
command.assert_not_called()
|
|
|
|
def test_formal_root_rejected_before_command(self):
|
|
with patch.object(migration.subprocess, 'check_output') as command:
|
|
with self.assertRaises(AssertionError):
|
|
migration.verify_migration(Path('/opt/smail'), 'smail.service', 'not-a-secret', {})
|
|
command.assert_not_called()
|
|
|
|
def test_process_proof_rejects_zero_pid(self):
|
|
with self.assertRaises(AssertionError):
|
|
migration.process_uses_target(Path('/unused'), {'MainPID': '0'})
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|