import tempfile import unittest import os import threading from pathlib import Path from uuid import uuid4 from archive_clients.state import ( ClientStore, CommandConflict, FileOperationConflict, JobConflict, ) class ClientStoreTests(unittest.TestCase): def test_database_connections_serialize_local_writers(self): """Concurrent jobs share one daemon DB without SQLite lock failures.""" with tempfile.TemporaryDirectory() as directory: database = Path(directory) / "state.db" first = ClientStore(database) second = ClientStore(database) first.initialize() barrier = threading.Barrier(2) failures: list[Exception] = [] def write(store, command_id): try: barrier.wait() store.accept_command(command_id, '{"kind":"job"}', '{"ok":true}') except Exception as exc: # pragma: no cover - assertion below failures.append(exc) left = threading.Thread(target=write, args=(first, "command-1")) right = threading.Thread(target=write, args=(second, "command-2")) left.start() right.start() left.join(1) right.join(1) self.assertFalse(left.is_alive()) self.assertFalse(right.is_alive()) self.assertEqual(failures, []) self.assertEqual(len(first.list_accepted_commands()), 2) with first._connect() as connection: self.assertEqual( connection.execute("PRAGMA busy_timeout").fetchone()[0], 30000, ) def test_command_acceptance_is_durable_and_content_addressed(self): with tempfile.TemporaryDirectory() as directory: database = Path(directory) / "state.db" store = ClientStore(database) store.initialize() self.assertEqual(os.stat(database).st_mode & 0o777, 0o600) command_id = str(uuid4()) first = store.accept_command(command_id, '{"b":2,"a":1}', '{"ok":true}') duplicate = ClientStore(database).accept_command( command_id, '{"a":1,"b":2}', '{"ok":false}' ) self.assertFalse(first.duplicate) self.assertTrue(duplicate.duplicate) self.assertEqual(duplicate.acknowledgement_json, '{"ok":true}') with self.assertRaises(CommandConflict): store.accept_command(command_id, '{"a":2}', '{"ok":true}') def test_job_definition_is_immutable_while_cursor_advances(self): with tempfile.TemporaryDirectory() as directory: store = ClientStore(Path(directory) / "state.db") store.initialize() store.save_job( "job-1", '{"jobId":"job-1"}', "JOB_STATE_WAITING", 1, 2, False, ) store.save_job( "job-1", '{"jobId":"job-1"}', "JOB_STATE_RUNNING", 2, 3, False, ) self.assertEqual( store.job_snapshot_rows(["job-1"])[0]["revision"], 2 ) with self.assertRaisesRegex(JobConflict, "backwards"): store.save_job( "job-1", '{"jobId":"job-1"}', "JOB_STATE_WAITING", 1, 2, False, ) with self.assertRaises(JobConflict): store.save_job( "job-1", '{"jobId":"other"}', "JOB_STATE_RUNNING", 2, 3, False, ) def test_route_attempt_nonce_and_updates_survive_duplicate_commands(self): with tempfile.TemporaryDirectory() as directory: store = ClientStore(Path(directory) / "state.db") store.initialize() command_id = str(uuid4()) store.accept_command(command_id, '{"route":"one"}', '{"ok":true}') first = store.begin_route_attempt( command_id, "route-1", '{"routeId":"route-1"}', "nonce-1" ) duplicate = store.begin_route_attempt( command_id, "route-1", '{"routeId":"route-1"}', "new-nonce" ) self.assertEqual(first["nonce"], "nonce-1") self.assertEqual(duplicate["nonce"], "nonce-1") store.record_route_update( command_id, 1, "provisioning", '{"sequence":1}' ) store.record_route_ownership(command_id, True, True) store.record_route_update( command_id, 1, "provisioning", '{"sequence":1}' ) store.record_route_update( command_id, 2, "ready", '{"sequence":2}' ) self.assertEqual(store.get_route_attempt(command_id)["state"], "ready") self.assertEqual(store.get_route_attempt(command_id)["created_folder"], 1) self.assertEqual( [row["sequence"] for row in store.route_update_rows(command_id)], [1, 2], ) with self.assertRaises(CommandConflict): store.record_route_update( command_id, 2, "ready", '{"sequence":2,"changed":true}' ) def test_file_operation_journal_is_idempotent_and_content_addressed(self): with tempfile.TemporaryDirectory() as directory: store = ClientStore(Path(directory) / "state.db") store.initialize() first = store.begin_file_operation( "operation-1", "job-1", '{"destination":"a","source":"b"}' ) repeated = store.begin_file_operation( "operation-1", "job-1", '{"source":"b","destination":"a"}' ) self.assertEqual(first["state"], "intent") self.assertEqual(repeated["state"], "intent") completed = store.complete_file_operation( "operation-1", '{"method":1}' ) duplicate = store.complete_file_operation( "operation-1", '{"method":1}' ) self.assertEqual(completed["state"], "completed") self.assertEqual(duplicate["result_json"], '{"method":1}') with self.assertRaises(FileOperationConflict): store.begin_file_operation( "operation-1", "job-1", '{"source":"other"}' ) with self.assertRaises(FileOperationConflict): store.complete_file_operation( "operation-1", '{"method":2}' ) def test_job_artifacts_are_immutable_and_survive_restart(self): with tempfile.TemporaryDirectory() as directory: database = Path(directory) / "state.db" store = ClientStore(database) store.initialize() store.save_job( "job-1", '{"jobId":"job-1"}', "JOB_STATE_RUNNING", 1, 0, False, ) store.put_job_artifact( "job-1", "baseline", {"selected": [1, 3]} ) reopened = ClientStore(database) self.assertEqual( reopened.get_job_artifact("job-1", "baseline")["value"], {"selected": [1, 3]}, ) with self.assertRaises(JobConflict): reopened.put_job_artifact( "job-1", "baseline", {"selected": [2]} ) def test_reconciliation_retires_only_stale_job_leases(self): with tempfile.TemporaryDirectory() as directory: store = ClientStore(Path(directory) / "state.db") store.initialize() stale = store.accept_command( "stale", '{"executeStep":{"jobId":"job-1"}}', '{"status":"COMMAND_ACK_STATUS_ACCEPTED"}', ) store.accept_command( "other", '{"executeStep":{"jobId":"job-2"}}', '{"status":"COMMAND_ACK_STATUS_ACCEPTED"}', ) self.assertEqual(stale.state, "accepted") store.save_job( "job-1", '{"jobId":"job-1"}', "JOB_STATE_RUNNING", 3, 3, False, ) store.reconcile_job( job_id="job-1", definition_json='{"jobId":"job-1"}', state="JOB_STATE_QUEUED", revision=0, last_event_sequence=0, committed=False, superseded_command_ids=["stale"], ) self.assertFalse(store.is_command_active("stale")) self.assertTrue(store.is_command_active("other")) row = store.job_snapshot_rows(["job-1"])[0] self.assertEqual(row["last_event_sequence"], 0) self.assertEqual(row["state"], "JOB_STATE_QUEUED") def test_reconciliation_acknowledgement_and_state_are_atomic(self): with tempfile.TemporaryDirectory() as directory: store = ClientStore(Path(directory) / "state.db") store.initialize() store.accept_command( "stale", '{"executeStep":{"jobId":"job-1"}}', '{"status":"COMMAND_ACK_STATUS_ACCEPTED"}', ) accepted = store.accept_reconcile_command( "reconcile-1", '{"reconcileJob":{"authoritativeJob":{}}}', '{"status":"COMMAND_ACK_STATUS_ACCEPTED"}', job_id="job-1", definition_json='{"jobId":"job-1"}', state="JOB_STATE_QUEUED", revision=0, last_event_sequence=0, committed=False, superseded_command_ids=["stale"], ) self.assertFalse(accepted.duplicate) self.assertEqual(accepted.state, "accepted") self.assertTrue(store.is_command_active("reconcile-1")) self.assertFalse(store.is_command_active("stale")) self.assertEqual( store.job_snapshot_rows(["job-1"])[0]["last_event_sequence"], 0 ) duplicate = store.accept_reconcile_command( "reconcile-1", '{"reconcileJob":{"authoritativeJob":{}}}', '{"status":"COMMAND_ACK_STATUS_ACCEPTED"}', job_id="job-1", definition_json='{"jobId":"job-1"}', state="JOB_STATE_QUEUED", revision=0, last_event_sequence=0, committed=False, superseded_command_ids=["stale"], ) self.assertTrue(duplicate.duplicate) if __name__ == "__main__": unittest.main()