"""Real PostgreSQL concurrency and transaction regressions for miner control.""" import unittest from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timezone from uuid import UUID import psycopg from psycopg.types.json import Jsonb from app.agent_commands.models import AgentCommandResultRequest from app.agent_commands.repository import AgentCommandRepository from app.agent_commands.service import AgentCommandService from app.config import settings from app.miner_control.exceptions import MinerControlConflict from app.miner_control.repository import MinerControlRepository from app.miner_control.service import MinerControlService SITE_ID = UUID("00000000-0000-4000-8000-000000001510") MINER_ID = UUID("00000000-0000-4000-8000-000000001511") USER_ID = UUID("00000000-0000-4000-8000-000000001512") OTHER_MINER_ID = UUID("00000000-0000-4000-8000-000000001513") AGENT_ID = "control-test-agent" def connect(): return psycopg.connect( host=settings.postgres_host, port=settings.postgres_port, dbname=settings.postgres_db, user=settings.postgres_user, password=settings.postgres_password, connect_timeout=5, ) def result_payload(accepted=True): return { "schema_version": 1, "command_accepted": accepted, "verification_attempted": False, "verification_status": "unconfirmed" if accepted else "unavailable", "observed_service_state": None, "observed_performance_mode": None, "failure_code": None if accepted else "write_rejected", } class MinerControlPostgresTests(unittest.TestCase): @classmethod def setUpClass(cls): connection = connect() try: with connection.cursor() as cur: cur.execute( "INSERT INTO sites (id, name) VALUES (%s, 'Control Test Site') " "ON CONFLICT (id) DO NOTHING", (SITE_ID,) ) cur.execute( """ INSERT INTO users (id, username, display_name, password_hash, role_id) SELECT %s, 'control-test-user', 'Control Test User', 'unused', id FROM roles WHERE name = 'Admin' ON CONFLICT (id) DO NOTHING """, (USER_ID,) ) cur.execute( """ INSERT INTO miners ( id, site_id, asset_tag, current_ip, status, api3_supported, configured_performance_mode, first_seen_at, last_seen_at ) VALUES (%s, %s, 'CONTROL-TEST', '192.0.2.10', 'online', true, 'normal', NULL, NULL) ON CONFLICT (id) DO UPDATE SET site_id = EXCLUDED.site_id, current_ip = EXCLUDED.current_ip, status = EXCLUDED.status, api3_supported = true """, (MINER_ID, SITE_ID) ) cur.execute( """ INSERT INTO minerops_site_agents ( agent_id, site_id, site_name, last_payload ) VALUES (%s, %s, 'Control Test Site', %s) ON CONFLICT (agent_id) DO UPDATE SET site_id = EXCLUDED.site_id, last_payload = EXCLUDED.last_payload """, (AGENT_ID, SITE_ID, Jsonb({"control_available": True})) ) connection.commit() finally: connection.close() @classmethod def tearDownClass(cls): connection = connect() try: with connection.cursor() as cur: cur.execute("DELETE FROM audit_logs WHERE entity_id = %s", (MINER_ID,)) cur.execute( "DELETE FROM minerops_agent_command_queue " "WHERE request_payload ->> 'miner_id' IN (%s, %s)", (str(MINER_ID), str(OTHER_MINER_ID)), ) cur.execute("DELETE FROM minerops_site_agents WHERE agent_id = %s", (AGENT_ID,)) cur.execute("DELETE FROM miners WHERE id = %s", (MINER_ID,)) cur.execute("DELETE FROM users WHERE id = %s", (USER_ID,)) cur.execute("DELETE FROM sites WHERE id = %s", (SITE_ID,)) connection.commit() finally: connection.close() def setUp(self): connection = connect() try: with connection.cursor() as cur: cur.execute("DELETE FROM audit_logs WHERE entity_id = %s", (MINER_ID,)) cur.execute( "DELETE FROM minerops_agent_command_queue " "WHERE request_payload ->> 'miner_id' IN (%s, %s)", (str(MINER_ID), str(OTHER_MINER_ID)), ) connection.commit() finally: connection.close() def insert_command(self, command_id, miner_id, status="pending", created=None): connection = connect() try: with connection.cursor() as cur: cur.execute( """ INSERT INTO minerops_agent_command_queue ( id, agent_id, command_type, command_status, request_payload, created_at ) VALUES (%s, %s, 'miner_control', %s, %s, COALESCE(%s, now())) """, (command_id, AGENT_ID, status, Jsonb({"schema_version": 1, "miner_id": str(miner_id), "action": "suspend", "requested_value": None}), created), ) connection.commit() finally: connection.close() def test_two_concurrent_claimers_never_receive_same_command(self): self.insert_command("concurrent-one", MINER_ID) with ThreadPoolExecutor(max_workers=2) as pool: results = list(pool.map( lambda _value: AgentCommandRepository(connect).claim_pending(AGENT_ID), range(2), )) claimed = [row["command_id"] for result in results for row in result] self.assertEqual(claimed, ["concurrent-one"]) def test_skip_locked_claims_next_available_row(self): self.insert_command( "a-locked", MINER_ID, created=datetime(2026, 8, 24, 10, tzinfo=timezone.utc), ) self.insert_command( "b-available", OTHER_MINER_ID, created=datetime(2026, 8, 24, 11, tzinfo=timezone.utc), ) locker = connect() try: with locker.cursor() as cur: cur.execute( "SELECT id FROM minerops_agent_command_queue " "WHERE id = 'a-locked' FOR UPDATE" ) claimed = AgentCommandRepository(connect).claim_pending(AGENT_ID) self.assertEqual([row["command_id"] for row in claimed], ["b-available"]) finally: locker.rollback() locker.close() def test_result_binding_claim_requirement_and_terminal_immutability(self): self.insert_command("result-one", MINER_ID, status="claimed") request = AgentCommandResultRequest( command_id="result-one", status="completed", result_payload=result_payload() ) service = AgentCommandService(AgentCommandRepository(connect)) with self.assertRaises(Exception): service.submit_result("wrong-agent", request) service.submit_result(AGENT_ID, request) with self.assertRaises(Exception): service.submit_result(AGENT_ID, request) self.insert_command("pending-result", OTHER_MINER_ID, status="pending") pending = AgentCommandResultRequest( command_id="pending-result", status="failed", result_payload=result_payload(False) ) with self.assertRaises(Exception): service.submit_result(AGENT_ID, pending) def test_atomic_queue_audit_duplicate_live_and_completed_history(self): service = MinerControlService(MinerControlRepository(connect)) first = service.queue_control( MINER_ID, "suspend", requested_value=None, actor_user_id=USER_ID, actor_username="control-test-user", ) with self.assertRaises(MinerControlConflict): service.queue_control( MINER_ID, "suspend", requested_value=None, actor_user_id=USER_ID, actor_username="control-test-user", ) connection = connect() try: with connection.cursor() as cur: cur.execute( "SELECT count(*) FROM audit_logs WHERE entity_id = %s", (MINER_ID,) ) self.assertEqual(cur.fetchone()[0], 1) cur.execute( "UPDATE minerops_agent_command_queue SET command_status = 'completed', " "completed_at = now() WHERE id = %s", (first.operation_id,) ) connection.commit() finally: connection.close() second = service.queue_control( MINER_ID, "suspend", requested_value=None, actor_user_id=USER_ID, actor_username="control-test-user", ) self.assertNotEqual(first.operation_id, second.operation_id) def test_audit_failure_rolls_back_queue_row(self): service = MinerControlService(MinerControlRepository(connect)) operation_count_before = self._operation_count() with self.assertRaises(Exception): service.queue_control( MINER_ID, "suspend", requested_value=None, actor_user_id=UUID("00000000-0000-4000-8000-000000009999"), actor_username="missing-user", ) self.assertEqual(self._operation_count(), operation_count_before) def _operation_count(self): connection = connect() try: with connection.cursor() as cur: cur.execute( "SELECT count(*) FROM minerops_agent_command_queue " "WHERE request_payload ->> 'miner_id' = %s", (str(MINER_ID),) ) return cur.fetchone()[0] finally: connection.close() if __name__ == "__main__": unittest.main()