1import unittest
2
3from native_maintenance import maintenance_locks
4
5
6class Maintenance(unittest.TestCase):
7 def history(self, path='m6-collaboration/synthetic.one', host='192.168.77.2', status='0xc0000055'):
8 events = [{'connection': 1, 'opened': True, 'peer': [host, 1]}]
9 def exchange(command, status='0x0', **fields):
10 message = len(events)
11 events.extend([
12 {'connection': 1, 'direction': 'request', 'command': command, 'message': message, **fields},
13 {'connection': 1, 'direction': 'response', 'command': command, 'message': message, 'status': status, 'file_id': 'file'},
14 ])
15 exchange(5, path=path)
16 for phase, result in [('maintenance-held', status), ('maintenance-released', '0x0')]:
17 events.append({'control': {'phase': phase}})
18 exchange(10, result, file_id='file', locks=[(0xffffeffc, 4096, 0x12)])
19 return events
20
21 def test_section_denied_then_acquired(self):
22 self.assertEqual(len(maintenance_locks(self.history())), 2)
23
24 def test_toc_compaction_does_not_count(self):
25 with self.assertRaises(AssertionError):
26 maintenance_locks(self.history(path='m6-collaboration/Open Notebook.onetoc2'))
27
28 def test_rust_maintenance_does_not_count(self):
29 with self.assertRaises(AssertionError):
30 maintenance_locks(self.history(host='10.0.2.2'))
31
32 def test_unrelated_error_does_not_prove_exclusion(self):
33 with self.assertRaises(AssertionError):
34 maintenance_locks(self.history(status='0xc0000001'))
35
36 def test_delayed_response_uses_request_phase(self):
37 events = self.history()
38 events.insert(-1, {'control': {'phase': 'maintenance-resumed'}})
39 self.assertEqual(len(maintenance_locks(events)), 2)
40 events = self.history()
41 events.insert(-2, {'control': {'phase': 'maintenance-resumed'}})
42 with self.assertRaises(AssertionError):
43 maintenance_locks(events)
44
45 def test_shared_or_short_lock_is_not_maintenance(self):
46 for lock in [(0xffffeffc, 4096, 0x11), (0xfffffffc, 1, 0x12)]:
47 events = self.history()
48 for event in events:
49 if 'locks' in event: event['locks'] = [lock]
50 with self.assertRaises(AssertionError): maintenance_locks(events)
51
52
53if __name__ == '__main__': unittest.main()