83 lines
5.3 KiB
Python
83 lines
5.3 KiB
Python
|
"""Bounded root outbox scans retain poison rows and revisit fixed snapshots."""
|
||
|
|
import json
|
||
|
|
from pathlib import Path
|
||
|
|
import sqlite3
|
||
|
|
import unittest
|
||
|
|
|
||
|
|
|
||
|
|
class FileOutboxScan(unittest.TestCase):
|
||
|
|
def test_history_scan_and_ack_are_origin_and_bot_scoped(self):
|
||
|
|
source = Path(__file__).resolve().parents[1] / 'crates/supervisor/src/runner_files.py'
|
||
|
|
module = {}
|
||
|
|
exec(compile(source.read_text(), str(source), 'exec'), module)
|
||
|
|
current = {'computer_id': 'replacement', 'bot_id': 'a', 'generation': 2}
|
||
|
|
origin = {**current, 'computer_id': 'original', 'generation': 1}
|
||
|
|
peer = {**origin, 'bot_id': 'b'}
|
||
|
|
with sqlite3.connect(':memory:') as db:
|
||
|
|
module['file_journal_schema'](db)
|
||
|
|
db.executemany('INSERT INTO file_outbox(identity,payload) VALUES (?,?)',
|
||
|
|
[(module['file_identity'](origin), '{bad')] * 100 +
|
||
|
|
[(module['file_identity'](origin), '{"retained":true}'),
|
||
|
|
(module['file_identity'](peer), '{"peer":true}'),
|
||
|
|
(module['file_identity'](current), '{"current":true}')])
|
||
|
|
def call(cursor, action='file_history_events', scope=current, **extra):
|
||
|
|
body = {'origin_computer_id': 'original', 'origin_generation': 1, 'cursor': cursor, **extra}
|
||
|
|
return module['file_history_events'](db, {'identity': scope, 'action': action, 'input': json.dumps(body)})
|
||
|
|
first = call({})
|
||
|
|
self.assertEqual(len(first['events']), 100)
|
||
|
|
self.assertTrue(all(e['event'] == 'file.invalid' for e in first['events']))
|
||
|
|
second = call(json.loads(first['stdout']))
|
||
|
|
self.assertEqual([e['id'] for e in second['events']], [101])
|
||
|
|
call({'ids': [101, 102, 103]}, 'ack_file_history_events')
|
||
|
|
self.assertEqual([r[0] for r in db.execute('SELECT id FROM file_outbox ORDER BY id')], list(range(1, 101)) + [102, 103])
|
||
|
|
self.assertEqual(call({}, scope={**current, 'bot_id': 'other'})['events'], [])
|
||
|
|
for extra in [{'origin_generation': 3}, {'origin_generation': True}, {'bot_id': 'b'}, {'resolve': True}]:
|
||
|
|
with self.assertRaises(ValueError):
|
||
|
|
call({}, **extra)
|
||
|
|
with self.assertRaises(ValueError):
|
||
|
|
call({'ids': [102], 'bot_id': 'b'}, 'ack_file_history_events')
|
||
|
|
|
||
|
|
def test_poison_page_does_not_block_later_receipts_or_discard_failures(self):
|
||
|
|
source = Path(__file__).resolve().parents[1] / 'crates/supervisor/src/runner_files.py'
|
||
|
|
module = {}
|
||
|
|
exec(compile(source.read_text(), str(source), 'exec'), module)
|
||
|
|
identity = {'computer_id': 'fixture', 'bot_id': 'scan', 'generation': 1}
|
||
|
|
encoded = module['file_identity'](identity)
|
||
|
|
with sqlite3.connect(':memory:') as db:
|
||
|
|
module['file_journal_schema'](db)
|
||
|
|
db.executemany('INSERT INTO file_outbox(identity,payload) VALUES (?,?)',
|
||
|
|
[(encoded, '{broken')] * 100 + [(encoded, '{"valid":true}')])
|
||
|
|
def scan(body, scope=identity, action='file_events'):
|
||
|
|
return module['file_events'](db, {'identity': scope, 'action': action, 'input': json.dumps(body)})
|
||
|
|
first = scan({})
|
||
|
|
self.assertEqual(len(first['events']), 100)
|
||
|
|
self.assertTrue(all(e['event'] == 'file.invalid' for e in first['events']))
|
||
|
|
self.assertNotIn('broken', json.dumps(first))
|
||
|
|
cursor = json.loads(first['stdout'])
|
||
|
|
self.assertEqual(cursor, {'after_id': 100, 'through_id': 101})
|
||
|
|
db.execute('INSERT INTO file_outbox(identity,payload) VALUES (?,?)', (encoded, '{"new":true}'))
|
||
|
|
second = scan(cursor)
|
||
|
|
self.assertEqual([e['id'] for e in second['events']], [101])
|
||
|
|
self.assertEqual(json.loads(second['stdout']), {'after_id': 0, 'through_id': 0})
|
||
|
|
scan({'ids': [101]}, action='ack_file_events')
|
||
|
|
self.assertEqual(db.execute('SELECT count(*) FROM file_outbox').fetchone()[0], 101)
|
||
|
|
repeated = scan({})
|
||
|
|
self.assertEqual([e['id'] for e in repeated['events']], list(range(1, 101)))
|
||
|
|
self.assertEqual([e['id'] for e in scan(json.loads(repeated['stdout']))['events']], [102])
|
||
|
|
self.assertEqual(scan(cursor, {**identity, 'bot_id': 'other'})['events'], [])
|
||
|
|
for bad in [{'after_id': True, 'through_id': 1}, {'after_id': 2, 'through_id': 1},
|
||
|
|
{'after_id': 0}, {'after_id': 0, 'through_id': 2**63}]:
|
||
|
|
with self.assertRaises(ValueError):
|
||
|
|
scan(bad)
|
||
|
|
# Unicode escaping must not make one row exceed the response budget
|
||
|
|
# and permanently prevent its cursor from advancing.
|
||
|
|
db.execute('UPDATE file_outbox SET payload=? WHERE id=1', (json.dumps({'text': '\u0001' * 30000}, ensure_ascii=False),))
|
||
|
|
self.assertEqual(scan({})['events'][0]['event'], 'file.invalid')
|
||
|
|
for corrupt in [b'blob', '[' * 2000 + '0' + ']' * 2000, 'x' * (1024 * 1024), '{}\0trailing', '{"x":NaN}', '{"x":1e999}', '{"x":' + '9' * 1000 + '}', r'{"x":"\ud800"}']:
|
||
|
|
db.execute('UPDATE file_outbox SET payload=? WHERE id=1', (corrupt,))
|
||
|
|
self.assertEqual(scan({})['events'][0]['event'], 'file.invalid')
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
unittest.main()
|