import json import importlib.util import pathlib import subprocess import sys import types import unittest from unittest.mock import patch ROOT = pathlib.Path(__file__).parents[1] class TransactionDurabilityTests(unittest.TestCase): def run_store_fixture(self, mode="success", transactions=None): transactions = transactions or [ { "id": "42-1000-1", "test_id": 42, "timestamp": 10, "instrument_id": 7, "order_name": "grid_order-1", "side": 1, "amount": 1, "position": 1, "price": 100, "pnl": 0, "equity": 100, "average_price": 100, "type": 2, "fee": 0.1, }, { "id": "42-1000-2", "test_id": 42, "timestamp": 11, "instrument_id": 7, "order_name": "profit_order", "side": 2, "amount": 1, "position": 0, "price": 101, "pnl": 1, "equity": 101, "average_price": 0, "type": 2, "fee": 0.1, }, ] source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2Store.php'))}; " "class FakeDb { public $rows=[]; public $mode; function __construct($mode){$this->mode=$mode;} " "function mysqli_update($sql,$params){ if($this->mode==='insert-failure')return false; " "$v=$params['values'];$this->rows[(string)$v[0]]=$v;return true;} " "function mysqli_fetch_all($sql,$params){ if(str_contains($sql,'transaction_id')){" "$ids=array_keys($this->rows);if($this->mode==='mismatch')array_pop($ids);" "return array_map(fn($id)=>['transaction_id'=>$id],$ids);} return [];}} " "$store=(new ReflectionClass('ResultsV2Store'))->newInstanceWithoutConstructor();" "$p=(new ReflectionClass('ResultsV2Store'))->getProperty('resultsDb');$p->setAccessible(true);" f"$db=new FakeDb({json.dumps(mode)});$p->setValue($store,$db);" f"$tx=json_decode({json.dumps(json.dumps(transactions))},true);" "$result=$store->reconcileTransactions(42,$tx);echo json_encode([$result,count($db->rows)]);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) return json.loads(result.stdout) def test_reconciliation_is_idempotent_and_verifies_the_full_identity_set(self): self.assertEqual([True, 2], self.run_store_fixture()) self.assertEqual([False, 2], self.run_store_fixture("mismatch")) self.assertEqual([False, 0], self.run_store_fixture("insert-failure")) def test_duplicate_identity_preflight_fails_before_any_write(self): duplicate = [{"id": "42-1", "test_id": 42}, {"id": "42-1", "test_id": 42}] self.assertEqual([False, 0], self.run_store_fixture(transactions=duplicate)) def test_consumer_and_schema_use_the_same_stable_idempotency_key(self): handler = (ROOT / "beeDataHandler" / "beeDataHandler.py").read_text(encoding="utf-8") transaction_handler = handler.split("def upsert_order_to_db", 1)[0] migration = (ROOT / "deploy/migrations/20260913_bee_transaction_durability.sql").read_text(encoding="utf-8") self.assertIn("ON DUPLICATE KEY UPDATE", handler) self.assertIn("return True", handler) self.assertIn("SIGNAL SQLSTATE '45000'", migration) self.assertIn( "UNIQUE INDEX `uq_bee_transaction_identity` (`test_id`, `transaction_id`)", migration, ) self.assertIn("GROUP BY `test_id`, `transaction_id`", migration) self.assertNotIn("`test_id` = VALUES(`test_id`)", transaction_handler) def test_store_upsert_preserves_transaction_ownership(self): store = (ROOT / "classes" / "ResultsV2Store.php").read_text(encoding="utf-8") self.assertIn("ON DUPLICATE KEY UPDATE", store) self.assertNotIn("`test_id` = VALUES(`test_id`)", store) def test_migration_is_restart_safe_and_verifies_exact_index_shape(self): migration = (ROOT / "deploy/migrations/20260913_bee_transaction_durability.sql").read_text(encoding="utf-8") self.assertIn("DROP PROCEDURE IF EXISTS `apply_bee_transaction_identity`", migration) self.assertIn("INFORMATION_SCHEMA`.`STATISTICS", migration) self.assertIn("SUB_PART", migration) self.assertIn("index_columns <> 'test_id,transaction_id'", migration) self.assertGreaterEqual(migration.count("index_column_count <> 2"), 2) self.assertGreaterEqual(migration.count("index_prefix_count <> 0"), 2) self.assertIn("transaction index shape mismatch", migration) self.assertIn("transaction index verification failed", migration) def test_consumer_duplicate_delivery_and_unavailable_pool_are_safe(self): module_path = ROOT / "beeDataHandler" / "beeDataHandler.py" saved_module = sys.modules.get("ConnectionPool") sys.modules["ConnectionPool"] = types.SimpleNamespace(ConnectionPool=object) try: spec = importlib.util.spec_from_file_location("bee_transaction_handler", module_path) handler = importlib.util.module_from_spec(spec) with patch("logging.basicConfig"): spec.loader.exec_module(handler) finally: if saved_module is None: del sys.modules["ConnectionPool"] else: sys.modules["ConnectionPool"] = saved_module class Cursor: def __init__(self): self.calls = [] def execute(self, query, params): self.calls.append((query, params)) class Connection: def __init__(self): self.cursor_value = Cursor() self.commits = 0 def cursor(self): return self.cursor_value def commit(self): self.commits += 1 class Pool: def __init__(self, available=True): self.connection = Connection() self.available = available self.released = [] def get_connection(self): if not self.available: raise handler.Error("unavailable") return self.connection def release_connection(self, connection): self.released.append(connection) transaction = { "id": "42-1000-1", "test_id": 42, "timestamp": 10, "instrument_id": 7, "order_name": "grid_order-1", "side": 1, "amount": 1, "position": 1, "price": 100, "pnl": 0, "equity": 100, "average_price": 100, "type": 2, "fee": 0.1, } pool = Pool() self.assertTrue(handler.add_transaction_to_db(pool, transaction)) self.assertTrue(handler.add_transaction_to_db(pool, transaction)) self.assertEqual(2, pool.connection.commits) self.assertEqual(2, len(pool.connection.cursor_value.calls)) self.assertTrue(all(call[1][0] == transaction["id"] for call in pool.connection.cursor_value.calls)) self.assertTrue(all("ON DUPLICATE KEY UPDATE" in call[0] for call in pool.connection.cursor_value.calls)) unavailable = Pool(available=False) self.assertFalse(handler.add_transaction_to_db(unavailable, transaction)) self.assertEqual([], unavailable.released) def test_terminal_completion_is_gated_on_durable_reconciliation(self): observer = (ROOT / "classes/ResultsV2Observer.php").read_text(encoding="utf-8") self.assertIn("reconcileTransactions($testId, $bufferedTransactions)", observer) self.assertIn("transaction_persistence_incomplete", observer) def test_transient_buffer_failure_can_be_reconciled_by_exact_redelivery(self): source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2TelemetryBuffer.php'))}; " "class RetryRedis { public $keys=[];public $sets=[];public $fail=true;" "function exists($k){return isset($this->keys[$k])||isset($this->sets[$k]);}" "function set($k,$v){$this->keys[$k]=$v;return 'OK';}" "function sadd($k,$v){$n=!isset($this->sets[$k][$v]);$this->sets[$k][$v]=true;return $n?1:0;}" "function srem($k,$v){unset($this->sets[$k][$v]);return 1;}" "function smembers($k){return array_keys($this->sets[$k]??[]);}" "function jsonset($k,$p,$v){if($this->fail)return false;$this->keys[$k]=$v;return 'OK';}" "function jsonget($k){return $this->keys[$k]??null;}" "function jsondel($k,$p){unset($this->keys[$k]);return 1;}" "function del($k){unset($this->keys[$k],$this->sets[$k]);return 1;}}" "$r=new RetryRedis();$b=new ResultsV2TelemetryBuffer(42,$r);" "$tx=['id'=>'42-1','test_id'=>42,'timestamp'=>1];$first=$b->record($tx);" "$r->fail=false;$retry=$b->record($tx);$b2=new ResultsV2TelemetryBuffer(42,$r);" "echo json_encode([$first,$retry,$b2->transactions(),$b2->incompleteReasons()]);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) first, retry, transactions, reasons = json.loads(result.stdout) self.assertFalse(first) self.assertTrue(retry) self.assertEqual([{"id": "42-1", "test_id": 42, "timestamp": 1}], transactions) self.assertEqual([], reasons) def test_transaction_redelivery_preserves_other_incomplete_reasons(self): source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2TelemetryBuffer.php'))}; " "class RetryRedis { public $keys=[];public $sets=[];public $fail=true;" "function exists($k){return isset($this->keys[$k])||isset($this->sets[$k]);}" "function set($k,$v){$this->keys[$k]=$v;return 'OK';}" "function sadd($k,$v){$n=!isset($this->sets[$k][$v]);$this->sets[$k][$v]=true;return $n?1:0;}" "function srem($k,$v){unset($this->sets[$k][$v]);return 1;}" "function smembers($k){return array_keys($this->sets[$k]??[]);}" "function jsonset($k,$p,$v){if($this->fail)return false;$this->keys[$k]=$v;return 'OK';}" "function jsonget($k){return $this->keys[$k]??null;}" "function jsondel($k,$p){unset($this->keys[$k]);return 1;}" "function del($k){unset($this->keys[$k],$this->sets[$k]);return 1;}}" "$r=new RetryRedis();$b=new ResultsV2TelemetryBuffer(42,$r);" "$tx=['id'=>'42-1','test_id'=>42,'timestamp'=>1];$b->record($tx);" "$b->addIncompleteReason('equity_write_failed');$r->fail=false;$b->record($tx);" "$b2=new ResultsV2TelemetryBuffer(42,$r);" "echo json_encode([$r->exists('results-v2:telemetry-incomplete:42'),$b2->incompleteReasons()]);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) incomplete, reasons = json.loads(result.stdout) self.assertTrue(incomplete) self.assertEqual(["equity_write_failed"], reasons) if __name__ == "__main__": unittest.main()