| 1 | import copy |
| 2 | import json |
| 3 | from pathlib import Path |
| 4 | import sys |
| 5 | import tempfile |
| 6 | import unittest |
| 7 | from concurrent_rust import running_clients, verify |
| 8 | |
| 9 | |
| 10 | class OracleTests(unittest.TestCase): |
| 11 | def test_timeout_and_environment_reach_every_subprocess(self): |
| 12 | with tempfile.TemporaryDirectory() as directory: |
| 13 | output = Path(directory) |
| 14 | executable = output / 'client' |
| 15 | executable.write_text(f'#!{sys.executable}\nimport json, os, sys, time\nprint(json.dumps(dict(os.environ)), flush=True)\nwhile not os.path.exists(sys.argv[5]): time.sleep(.001)\n') |
| 16 | executable.chmod(0o700) |
| 17 | with running_clients(output, 'unused.one', 2, 1, 1, 7, timeout=600.25, |
| 18 | executable=executable, environment={'KEPT': 'value'}): |
| 19 | (output / 'start').touch() |
| 20 | for actor in ['w0', 'w1', 'r0']: |
| 21 | environment = json.loads((output / f'{actor}.jsonl').read_text()) |
| 22 | self.assertEqual(environment['ONESTORE_CLIENT_TIMEOUT_MS'], '600250') |
| 23 | self.assertEqual(environment['KEPT'], 'value') |
| 24 | self.assertEqual(json.loads((output / 'clients.json').read_text())['timeout_ms'], 600250) |
| 25 | |
| 26 | def test_distinct_reader_binary_is_launched_and_recorded(self): |
| 27 | with tempfile.TemporaryDirectory() as directory: |
| 28 | output = Path(directory) |
| 29 | for mode in ('writer', 'reader'): |
| 30 | executable = output / mode |
| 31 | executable.write_text(f'#!{sys.executable}\nimport os, sys, time\nprint({mode!r}, flush=True)\nwhile not os.path.exists(sys.argv[5]): time.sleep(.001)\n') |
| 32 | executable.chmod(0o700) |
| 33 | with running_clients(output, 'unused.one', 2, 1, 1, 7, |
| 34 | executable=output / 'writer', reader_executable=output / 'reader'): |
| 35 | (output / 'start').touch() |
| 36 | self.assertEqual((output / 'w0.jsonl').read_text().strip(), 'writer') |
| 37 | self.assertEqual((output / 'r0.jsonl').read_text().strip(), 'reader') |
| 38 | import hashlib |
| 39 | manifest = json.loads((output / 'clients.json').read_text()) |
| 40 | self.assertEqual(manifest['reader_sha256'], hashlib.sha256((output / 'reader').read_bytes()).hexdigest()) |
| 41 | |
| 42 | def setUp(self): |
| 43 | self.logs = { |
| 44 | 'w0': [{'event': 'ready'}, {'event': 'commit', 'source_transaction': 1, |
| 45 | 'operation': 0, 'token': ' [w0:0]', 'started_us': 10, 'finished_us': 30}, {'event': 'done'}], |
| 46 | 'w1': [{'event': 'ready'}, {'event': 'commit', 'source_transaction': 2, |
| 47 | 'operation': 0, 'token': ' [w1:0]', 'started_us': 20, 'finished_us': 40}, {'event': 'done'}], |
| 48 | 'r0': [{'event': 'ready'}, {'event': 'read', 'transaction': 3, |
| 49 | 'text': 'Concurrent edits: [w0:0] [w1:0]', 'started_us': 50, 'finished_us': 60}, {'event': 'done'}], |
| 50 | } |
| 51 | |
| 52 | def test_valid_history_and_deliberate_corruption(self): |
| 53 | self.assertEqual(verify(self.logs, 1, 2, 1)['commits'], 2) |
| 54 | for field, value in [('text', 'Concurrent edits: [w1:0]'), ('transaction', 2), ('finished_us', 19)]: |
| 55 | logs = copy.deepcopy(self.logs) |
| 56 | logs['r0'][1][field] = value |
| 57 | with self.assertRaises(AssertionError): |
| 58 | verify(logs, 1, 2, 1) |
| 59 | |
| 60 | def test_duplicate_snapshot_and_missing_acknowledgement(self): |
| 61 | logs = copy.deepcopy(self.logs) |
| 62 | logs['w1'][1]['source_transaction'] = 1 |
| 63 | with self.assertRaises(AssertionError): |
| 64 | verify(logs, 1, 2, 1) |
| 65 | logs['w1'].pop(1) |
| 66 | with self.assertRaises(AssertionError): |
| 67 | verify(logs, 1, 2, 1) |
| 68 | |
| 69 | def test_length_changing_unicode_history_and_wrong_intent(self): |
| 70 | previous = 'Concurrent edits:' |
| 71 | for i in range(2): |
| 72 | events = self.logs[f'w{i}'] |
| 73 | event = events[1] |
| 74 | event['attempt'] = 1 |
| 75 | replacement = ' café 🦀' + event['token'] |
| 76 | events.insert(1, {'event': 'intent', 'attempt': 1, 'operation': 0, |
| 77 | 'source_transaction': i + 1, 'before': previous, 'token': event['token'], |
| 78 | 'range': [17, len(previous.encode('utf-16-le')) // 2], 'replacement': replacement}) |
| 79 | previous = 'Concurrent edits:' + replacement |
| 80 | self.logs['r0'][1]['text'] = previous |
| 81 | self.assertEqual(verify(self.logs, 1, 2, 1, edit=True)['final_text'], previous) |
| 82 | self.logs['w1'][1]['range'][0] = 24 |
| 83 | with self.assertRaises(UnicodeDecodeError): |
| 84 | verify(self.logs, 1, 2, 1, edit=True) |
| 85 | |
| 86 | def test_serial_calls_do_not_count_as_concurrency(self): |
| 87 | self.logs['w1'][1]['started_us'] = 31 |
| 88 | with self.assertRaisesRegex(AssertionError, 'never overlapped'): |
| 89 | verify(self.logs, 1, 2, 1) |
| 90 | |
| 91 | |
| 92 | if __name__ == '__main__': |
| 93 | unittest.main() |