86 lines
4.9 KiB
Python
86 lines
4.9 KiB
Python
"""联合故障工具的范围、阶段与失败判定回归，不连接真实服务。"""
|
|
|
|
import http.client
|
|
import json
|
|
from pathlib import Path
|
|
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_sql_switch_fault as fault
|
|
from runtime_sql import SQLFixture
|
|
|
|
|
|
class RuntimeSQLSwitchFaultTest(unittest.TestCase):
|
|
def test_every_thread_must_be_stopped(self):
|
|
task = Mock()
|
|
for state, expected in (('S', False), ('R', False), ('T', True), ('t', True)):
|
|
task.__truediv__ = Mock(return_value=Mock(read_text=Mock(return_value='State:\t' + state + ' (state)\n')))
|
|
with patch.object(Path, 'iterdir', return_value=iter([task])):
|
|
self.assertEqual(fault.stopped_tasks('123'), expected)
|
|
|
|
def test_formal_scope_rejected_before_process_or_database_access(self):
|
|
with patch.object(fault.subprocess, 'Popen') as process, patch.object(fault, 'service_state') as state:
|
|
with self.assertRaises(AssertionError):
|
|
fault.verify_switch_fault(Path('/opt/smail'), 'smail.service', 'unused', {},
|
|
SQLFixture('mysql'), SQLFixture('mysql'), 'before-root')
|
|
process.assert_not_called()
|
|
state.assert_not_called()
|
|
|
|
def test_boundary_requires_active_directory_phase_and_expected_database(self):
|
|
source, target = SQLFixture('mysql'), SQLFixture('mysql')
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
archive = root / 'data/transaction/one'
|
|
archive.mkdir(parents=True)
|
|
(root / 'config').mkdir()
|
|
trace = root / 'trace.log'
|
|
manifest = archive / 'manifest.json'
|
|
for phase, name in (('before-root', source.settings['name']), ('after-root', target.settings['name'])):
|
|
(root / 'config/config.json').write_text(json.dumps({'database': {'name': name}}))
|
|
manifest.write_text(json.dumps({'phase': 'sections_committed'}))
|
|
for path, expected in (('/opt/smail/data/transaction/one/candidate', False),
|
|
('/opt/smail/config', True), ('/opt/smail/config-other', False)):
|
|
trace.write_text('renameat(4<' + path + '>, ".config.json-a", 4<' + path + '>, "config.json"')
|
|
self.assertEqual(bool(fault.boundary(root, trace, phase, source, target)), expected)
|
|
trace.write_text('renameat(4</opt/smail/config>, ".config.json-a", 4</opt/smail/config>, "config.json"')
|
|
manifest.write_text(json.dumps({'phase': 'root_committed'}))
|
|
self.assertIsNone(fault.boundary(root, trace, phase, source, target))
|
|
|
|
def test_committed_requires_matching_nonempty_mail_and_preserved_history(self):
|
|
source, target = Mock(), Mock()
|
|
for mail, history in (([[], []], [['a'], ['a']]),
|
|
([['a'], ['b']], [['a'], ['a']]),
|
|
([['a'], ['a']], [[], []]),
|
|
([['a'], ['a']], [['a'], ['b']])):
|
|
with patch.object(fault, 'table_rows', side_effect=mail), \
|
|
patch.object(fault, 'history_rows', side_effect=history):
|
|
with self.assertRaises(AssertionError):
|
|
fault.assert_committed(source, target)
|
|
with patch.object(fault, 'table_rows', side_effect=[['mail'], ['mail']]), \
|
|
patch.object(fault, 'history_rows', side_effect=[['old'], ['old', 'new']]):
|
|
self.assertEqual(fault.assert_committed(source, target)['committed_history_rows'], 2)
|
|
|
|
def test_timeout_and_regular_response_cannot_pass_as_killed_process(self):
|
|
for outcome in (None, TimeoutError(), http.client.RemoteDisconnected(), ConnectionResetError()):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory) / 'instance'
|
|
client = Mock()
|
|
client.success.return_value = {'migration': {'id': 'test-id'}, 'token': 'not-for-logs'}
|
|
client.request.side_effect = outcome
|
|
with patch.object(fault, 'login', return_value=(client, None)), \
|
|
patch.object(fault, 'clear_pending_restart', return_value=1), \
|
|
patch.object(fault, 'wait_until'):
|
|
if isinstance(outcome, (http.client.RemoteDisconnected, ConnectionResetError)):
|
|
self.assertTrue(fault.prepare(root, 'unused', SQLFixture('mysql'))['request_disconnected'])
|
|
else:
|
|
with self.assertRaises((AssertionError, TimeoutError)):
|
|
fault.prepare(root, 'unused', SQLFixture('mysql'))
|
|
self.assertEqual((root.parent / 'switch-ready').read_bytes(), b'')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|