import json import pathlib import subprocess import unittest ROOT = pathlib.Path(__file__).parents[1] class ResultsV2RuntimeContractTests(unittest.TestCase): def test_observer_records_actual_runtime_equity_without_mutating_order_state(self): runtime_state = { "id": 42, "timestamp": 120, "equity": 10.0, "position": 2.0, "average_price": 100.0, "mark_price": 125.0, } source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2Observer.php'))}; " f"$state = json_decode({json.dumps(json.dumps(runtime_state))}, true); " "$before = json_encode($state); " "$point = ResultsV2Observer::equityPoint($state); " "echo json_encode([$point, $before === json_encode($state)]);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) point, unchanged = json.loads(result.stdout) self.assertEqual(42, point["test_id"]) self.assertEqual(120, point["timestamp"]) self.assertEqual(10.0, point["wallet_balance"]) self.assertEqual(0.004, point["unrealized_pnl"]) self.assertEqual(10.004, point["equity"]) self.assertTrue(unchanged) def test_update_and_stop_observe_real_timeline_before_the_existing_terminal_transition(self): update = (ROOT / "api" / "private" / "update.php").read_text(encoding="utf-8") stop = (ROOT / "api" / "private" / "stop.php").read_text(encoding="utf-8") self.assertIn("ResultsV2Observer::fromEnvironment((int)$test_id", update) self.assertIn("ResultsV2Observer::fromEnvironment((int)$test_id", stop) self.assertGreater(update.index("$results_observer->observe($test_data);"), update.index("$order->fillThis")) natural_start = update.index("if($is_done) {", update.index("$guard_data_unavailable")) natural_terminal = update.index("$results_observer->persistTerminal($terminal_data", natural_start) self.assertGreater(natural_terminal, update.index("$order->ClosePosition", natural_start)) self.assertLess(natural_terminal, update.index("$test->stop();", natural_start)) self.assertIn("$terminal_data = $test->get();", update) self.assertIn("$results_observer->persistTerminal($terminal_data, 'end_reached', true);", update) self.assertIn("$results_observer->persistTerminal($persisted_test, $recovered_reason", update) self.assertIn("$results_observer->persistTerminal($test_data, $recovered_reason", update) self.assertGreaterEqual(update.count("$results_observer->persistTerminal"), 3) self.assertGreater(stop.index("$results_observer->persistTerminal"), stop.index("$order->ClosePosition")) self.assertLess(stop.index("$results_observer->persistTerminal"), stop.index("$test->stop(true);")) self.assertIn("$terminal_data = $test->get();", stop) self.assertIn("$results_observer->persistTerminal($terminal_data, 'stopped', false);", stop) def test_existing_indicator_endpoints_pass_their_calculated_values_to_the_observer(self): materialized = {"strategy_revision_id": 7, "indicator": {"period": 20, "z": 1}} snapshot_hash_source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; " f"echo ResultsV2::snapshotHash(json_decode({json.dumps(json.dumps(materialized))}, true));" ) snapshot_hash = subprocess.run(["php", "-r", snapshot_hash_source], capture_output=True, text=True, check=True).stdout source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; " f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2Observer.php'))}; " f"$materialized = json_decode({json.dumps(json.dumps(materialized))}, true); " f"$points = ResultsV2Observer::indicatorSnapshot(42, 'sma', ['period' => 20, 'z' => 1], 7, {json.dumps(snapshot_hash)}, " "$materialized, null, 'signal_evaluation', 'price', 'instrument_price', ['rule' => 'sma'], 120, 101.5, ['close' => 102]); " "echo json_encode($points);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) points = json.loads(result.stdout) self.assertEqual(1, len(points)) self.assertEqual("sma", points[0]["series_type"]) self.assertEqual(102, points[0]["price_context"]["close"]) self.assertEqual(7, points[0]["strategy_revision_id"]) reordered_source = source.replace("['period' => 20]", "['z' => 1, 'period' => 20]") canonical_source = source.replace("['period' => 20]", "['period' => 20, 'z' => 1]") reordered = json.loads(subprocess.run(["php", "-r", reordered_source], capture_output=True, text=True, check=True).stdout) canonical = json.loads(subprocess.run(["php", "-r", canonical_source], capture_output=True, text=True, check=True).stdout) self.assertEqual(reordered[0]["series_id"], canonical[0]["series_id"]) for endpoint in ("get_sma.php", "get_ema.php", "get_supertrend.php"): source = (ROOT / "api" / "ind" / endpoint).read_text(encoding="utf-8") self.assertIn("ResultsV2Observer::fromEnvironment((int)$test_id", source) self.assertIn("$results_observer->observeIndicator", source) def test_indicator_parameters_are_required_from_canonical_frozen_indicator_snapshot(self): materialized = {"strategy_revision_id": 7, "indicator": {"period": 20}} source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; " f"$materialized = json_decode({json.dumps(json.dumps(materialized))}, true); " "$hash = ResultsV2::snapshotHash($materialized); " "$ok = ResultsV2::indicatorSnapshot(42, 'sma:hash', 'sma', ['period' => 20], 7, $hash, $materialized, null, 'signal_evaluation', 'price', 'instrument_price', ['rule' => 'sma'], [[ 'timestamp' => 1, 'value' => 2 ]]); " "$tampered = false; try { ResultsV2::indicatorSnapshot(42, 'sma:hash', 'sma', ['period' => 21], 7, $hash, $materialized, null, 'signal_evaluation', 'price', 'instrument_price', ['rule' => 'sma'], [[ 'timestamp' => 1, 'value' => 2 ]]); } catch (InvalidArgumentException $e) { $tampered = true; } " "$missing = false; try { ResultsV2::indicatorSnapshot(42, 'sma:hash', 'sma', ['period' => 20], 7, ResultsV2::snapshotHash(['other' => 1]), ['other' => 1], null, 'signal_evaluation', 'price', 'instrument_price', ['rule' => 'sma'], [[ 'timestamp' => 1, 'value' => 2 ]]); } catch (InvalidArgumentException $e) { $missing = true; } " "echo json_encode([count($ok), $tampered, $missing]);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) self.assertEqual([1, True, True], json.loads(result.stdout)) def test_interface_runtime_snapshot_phase_setup_is_the_frozen_indicator_contract(self): snapshot = {"instrument_id": 92, "compounding": {}, "phases": [{"phase": 1, "setup": {"grid_type_use_ind": 1, "grid_type_ind_type": 4, "grid_type_ind_id": 64, "indicator_parameters": {"sma": {"period": 20, "candle_size": 5}}}}]} source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; require {json.dumps(str(ROOT / 'classes' / 'ResultsV2Observer.php'))}; " f"$snapshot=json_decode({json.dumps(json.dumps(snapshot))}, true); $hash=ResultsV2::snapshotHash($snapshot); " "$test=['strategy_revision_id'=>7,'strategy_snapshot'=>$snapshot,'strategy_snapshot_hash'=>$hash]; " "$points=ResultsV2::indicatorSnapshot(42,'sma:hash','sma',['period'=>20,'candle_size'=>5],7,$hash,$snapshot,1,'signal_evaluation','price','instrument_price',['rule'=>'sma'],[['timestamp'=>1,'value'=>2]]); " "$bad=false; try { ResultsV2::indicatorSnapshot(42,'sma:hash','sma',['period'=>21,'candle_size'=>5],7,$hash,$snapshot,1,'signal_evaluation','price','instrument_price',['rule'=>'sma'],[['timestamp'=>1,'value'=>2]]); } catch (InvalidArgumentException $e) {$bad=true;} " "echo json_encode([count($points),ResultsV2Observer::indicatorEvidenceRequired($test),$bad]);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) self.assertEqual([1, True, True], json.loads(result.stdout)) def test_legacy_terminal_is_partial_even_with_buffered_equity_and_transactions(self): state = {"id": 42, "start_equity": 10.0, "equity": 11.0, "fee": 0.1, "position": 0.0, "average_price": 0.0, "timestamp": 180} points = [{"timestamp": 60, "wallet_balance": 10.0, "unrealized_pnl": 0.0}, {"timestamp": 180, "wallet_balance": 11.0, "unrealized_pnl": 0.0}] transactions = [{"timestamp": 60, "position": 1.0, "pnl": 0.0}, {"timestamp": 180, "position": 0.0, "pnl": 1.0}] source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; require {json.dumps(str(ROOT / 'classes' / 'ResultsV2Observer.php'))}; " f"$state = json_decode({json.dumps(json.dumps(state))}, true); $points = json_decode({json.dumps(json.dumps(points))}, true); $tx = json_decode({json.dumps(json.dumps(transactions))}, true); " "$legacy = ResultsV2Observer::terminalSummary($state, $points, $tx, true, 'end_reached', true, 2, 4); " "$bound = $state + ['strategy_revision_id' => 7, 'strategy_snapshot' => ['indicator' => ['period' => 20]]]; $bound['strategy_snapshot_hash'] = ResultsV2::snapshotHash($bound['strategy_snapshot']); " "$complete = ResultsV2Observer::terminalSummary($bound, $points, $tx, true, 'end_reached', true, 2, 4); echo json_encode([$legacy, $complete]);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) legacy, complete = json.loads(result.stdout) self.assertEqual("legacy_partial", legacy["data_completeness"]) self.assertIsNone(legacy["max_drawdown"]) self.assertEqual("complete", complete["data_completeness"]) def test_interface_runtime_snapshot_fields_are_immutable_and_require_indicator_evidence(self): state = {"id": 42, "start_equity": 10.0, "fee": 0.1, "position": 0.0, "average_price": 0.0, "timestamp": 180} snapshot = {"instrument_id": 92, "compounding": {}, "phases": [{"phase": 1, "setup": {"grid_type_use_ind": 1, "grid_type_ind_type": 4, "grid_type_ind_id": 64, "indicator_parameters": {"sma": {"period": 20, "candle_size": 5}}}}]} points = [{"timestamp": 60, "wallet_balance": 10.0, "unrealized_pnl": 0.0}, {"timestamp": 180, "wallet_balance": 11.0, "unrealized_pnl": 0.0}] transactions = [{"timestamp": 60, "position": 1.0, "pnl": 0.0}, {"timestamp": 180, "position": 0.0, "pnl": 1.0}] source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; require {json.dumps(str(ROOT / 'classes' / 'ResultsV2Observer.php'))}; " f"$state=json_decode({json.dumps(json.dumps(state))}, true); $snapshot=json_decode({json.dumps(json.dumps(snapshot))}, true); " f"$points=json_decode({json.dumps(json.dumps(points))}, true); $tx=json_decode({json.dumps(json.dumps(transactions))}, true); " "$canonical=function($value) use (&$canonical){if(!is_array($value))return $value;if(array_is_list($value))return array_map($canonical,$value);ksort($value);foreach($value as $key=>$item)$value[$key]=$canonical($item);return $value;}; " "$state += ['strategy_revision_id'=>7,'runtime_snapshot_json'=>json_encode($snapshot),'runtime_content_hash'=>hash('sha256', json_encode($canonical($snapshot), JSON_UNESCAPED_SLASHES))]; " "$summary=ResultsV2Observer::terminalSummary($state,$points,$tx,true,'end_reached',true,2,4); " "class ResultsV2TelemetryBuffer { function __construct($id){} function addIncompleteReason($reason){} function markEvidence($kind){return true;} } " "class FakeResultsStore { public $points=[]; function persistIndicatorPoints($points){$this->points=$points;return true;} } " "$observer=(new ReflectionClass(ResultsV2Observer::class))->newInstanceWithoutConstructor(); $store=new FakeResultsStore(); $property=new ReflectionProperty(ResultsV2Observer::class,'store'); $property->setAccessible(true); $property->setValue($observer,$store); " "$observed=$observer->observeIndicator($state,'sma',['period'=>20,'candle_size'=>5],101.5,['close'=>102],'signal_evaluation','price','instrument_price',['indicator'=>'sma']); " "echo json_encode([$summary['data_completeness'],ResultsV2Observer::indicatorEvidenceRequired($state),$observed,count($store->points),$store->points[0]['strategy_snapshot_hash'],$state['runtime_content_hash']]);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) values = json.loads(result.stdout) self.assertEqual(["complete", True, True, 1], values[:4]) self.assertEqual(values[4], values[5]) def test_supertrend_snapshot_aliases_match_runtime_indicator_parameters(self): snapshot = { "phases": [{ "setup": { "indicator_parameters": { "supertrend": { "supertrend_atr_period": 20, "supertrend_candle_size": 15, "supertrend_multiplier": 16, "supertrend_deviation": 0, "supertrend_use_direction": 0, "supertrend_use_value": 0, } } } }] } source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; " f"$snapshot=json_decode({json.dumps(json.dumps(snapshot))}, true); " "echo json_encode(ResultsV2::frozenIndicatorParameters($snapshot,'supertrend'));" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) self.assertEqual( {"atr_length": 20, "candle_size": 15, "multiplier": 16}, json.loads(result.stdout), ) def test_indicator_evidence_requirement_follows_frozen_snapshot_and_allows_explicit_empty_indicators(self): source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; require {json.dumps(str(ROOT / 'classes' / 'ResultsV2Observer.php'))}; " "$with = ['strategy_revision_id' => 7, 'strategy_snapshot' => ['indicator' => ['period' => 20]]]; $with['strategy_snapshot_hash'] = ResultsV2::snapshotHash($with['strategy_snapshot']); " "$empty = ['strategy_revision_id' => 7, 'strategy_snapshot' => ['indicators' => []]]; $empty['strategy_snapshot_hash'] = ResultsV2::snapshotHash($empty['strategy_snapshot']); " "echo json_encode([ResultsV2Observer::indicatorEvidenceRequired($with), ResultsV2Observer::indicatorEvidenceRequired($empty)]);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) self.assertEqual([True, False], json.loads(result.stdout)) observer = (ROOT / "classes" / "ResultsV2Observer.php").read_text(encoding="utf-8") self.assertIn("indicator_evidence_missing", observer) def test_terminal_summary_is_idempotent_and_never_claims_complete_without_a_transaction_buffer(self): state = { "id": 42, "start_equity": 10.0, "equity": 11.0, "fee": 0.1, "position": 0.0, "average_price": 0.0, "timestamp": 180, } points = [{"timestamp": 60, "wallet_balance": 10.0, "unrealized_pnl": 0.0}, {"timestamp": 180, "wallet_balance": 11.0, "unrealized_pnl": 0.0}] transactions = [{"timestamp": 60, "position": 1.0, "pnl": 0.0}, {"timestamp": 180, "position": 0.0, "pnl": 1.0}] source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; " f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2Observer.php'))}; " f"$state = json_decode({json.dumps(json.dumps(state))}, true); " "$state['strategy_revision_id'] = 7; $state['strategy_snapshot'] = ['indicator' => ['period' => 20]]; $state['strategy_snapshot_hash'] = ResultsV2::snapshotHash($state['strategy_snapshot']); " f"$points = json_decode({json.dumps(json.dumps(points))}, true); " f"$transactions = json_decode({json.dumps(json.dumps(transactions))}, true); " "$complete = ResultsV2Observer::terminalSummary($state, $points, $transactions, true, 'end_reached', true); " "$repeat = ResultsV2Observer::terminalSummary($state, $points, $transactions, true, 'end_reached', true); " "$partial = ResultsV2Observer::terminalSummary($state, $points, [], false, 'end_reached', true); " "echo json_encode([$complete, $repeat, $partial]);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) complete, repeat, partial = json.loads(result.stdout) self.assertEqual(complete, repeat) self.assertEqual("complete", complete["data_completeness"]) self.assertEqual(1, complete["closed_trades"]) self.assertEqual("partial", partial["data_completeness"]) self.assertEqual(0, partial["closed_trades"]) transaction_log = (ROOT / "classes" / "TransactionLog.php").read_text(encoding="utf-8") self.assertIn("ResultsV2TelemetryBuffer", transaction_log) self.assertIn("->record($transaction_data)", transaction_log) self.assertIn("if (!$this->telemetryBuffer->record($transaction_data))", transaction_log) self.assertIn("$.results_v2_telemetry_incomplete", transaction_log) observer = (ROOT / "classes" / "ResultsV2Observer.php").read_text(encoding="utf-8") self.assertIn("public function observe(array $testData, bool $markTelemetry = true)", observer) self.assertIn("$this->observe($testData, false);", observer) self.assertIn("empty($testData['results_v2_telemetry_incomplete'])", observer) buffer = (ROOT / "classes" / "ResultsV2TelemetryBuffer.php").read_text(encoding="utf-8") self.assertIn("if ($this->redis->exists($this->incompleteKey())) return null;", buffer) self.assertIn("$this->markIncomplete();", buffer) def test_real_order_count_and_wall_clock_terminal_runtime_are_buffered_idempotently(self): order = (ROOT / "classes" / "Order.php").read_text(encoding="utf-8") start = (ROOT / "api" / "private" / "start.php").read_text(encoding="utf-8") observer = (ROOT / "classes" / "ResultsV2Observer.php").read_text(encoding="utf-8") buffer = (ROOT / "classes" / "ResultsV2TelemetryBuffer.php").read_text(encoding="utf-8") self.assertIn("->recordOrder($order_id)", order) self.assertIn("ResultsV2TelemetryBuffer($test_id)", start) self.assertIn("->mark();", start) self.assertIn("terminalRuntimeSeconds", buffer) self.assertIn("ordersCount", buffer) self.assertIn("'orders_count' => $ordersCount", observer) self.assertIn("'runtime_seconds' => $runtimeSeconds", observer) self.assertNotIn("end_time'] ?? 0) - (int)($testData['start_time']", observer) def test_terminal_observer_bounds_equity_and_cannot_block_the_stop_lifecycle(self): observer = (ROOT / "classes" / "ResultsV2Observer.php").read_text(encoding="utf-8") store = (ROOT / "classes" / "ResultsV2Store.php").read_text(encoding="utf-8") self.assertIn("getTerminalEquityPoints($testId)", observer) self.assertNotIn("$this->store->getEquityPoints($testId, null)", observer) self.assertIn("public function getTerminalEquityPoints", store) self.assertIn("public function getTerminalTransactions", store) self.assertIn("COUNT(*) AS `point_count`", store) self.assertIn("equity_points_truncated", observer) self.assertIn("transactions_truncated", observer) source = ( "class ResultsV2TelemetryBuffer { " "public static $reasons = []; " "function __construct($testId) {} function mark() {} function markEvidence($name) {} " "function addIncompleteReason($reason) { self::$reasons[] = $reason; } " "function transactions() { return []; } function ordersCount() { return 0; } " "function terminalRuntimeSeconds() { return 0; } function hasEvidence($name) { return true; } " "function incompleteReasons() { return self::$reasons; } " "} " "class FailingTerminalStore { " "function persistEquityPoints($testId, $points) { return true; } " "function getTerminalEquityPoints($testId) { throw new RuntimeException('simulated results failure'); } " "} " f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2.php'))}; " f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2Observer.php'))}; " "$observer = (new ReflectionClass('ResultsV2Observer'))->newInstanceWithoutConstructor(); " "$property = (new ReflectionClass('ResultsV2Observer'))->getProperty('store'); " "$property->setAccessible(true); $property->setValue($observer, new FailingTerminalStore()); " "$saved = $observer->persistTerminal(['id' => 42, 'equity' => 100, 'timestamp' => 1], 'stopped', false); " "echo json_encode([$saved, ResultsV2TelemetryBuffer::$reasons]);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) saved, reasons = json.loads(result.stdout) self.assertFalse(saved) self.assertIn("terminal_summary_failed", reasons) def test_evidence_write_failures_have_durable_reasons_and_force_partial(self): observer = (ROOT / "classes" / "ResultsV2Observer.php").read_text(encoding="utf-8") buffer = (ROOT / "classes" / "ResultsV2TelemetryBuffer.php").read_text(encoding="utf-8") for reason in ( "equity_point_write_failed", "indicator_write_failed", "indicator_provenance_missing", "transaction_write_failed", "terminal_summary_write_failed", "results_store_unavailable", ): self.assertIn(reason, observer + buffer) self.assertIn("incompleteReasons", buffer) self.assertIn("'incomplete_reasons' => $incompleteReasons", observer) def test_failed_transaction_buffer_write_produces_partial_terminal_evidence(self): source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2TelemetryBuffer.php'))}; " "class FakeRedis { " "public $keys = []; public $sets = []; " "function set($key, $value) { $this->keys[$key] = $value; return 'OK'; } " "function exists($key) { return isset($this->keys[$key]) || isset($this->sets[$key]); } " "function sadd($key, $value) { $new = !isset($this->sets[$key][$value]); $this->sets[$key][$value] = true; return $new ? 1 : 0; } " "function srem($key, $value) { unset($this->sets[$key][$value]); return 1; } " "function smembers($key) { return array_keys($this->sets[$key] ?? []); } " "function jsonset($key, $path, $value) { return false; } " "function jsonget($key) { return null; } " "function jsondel($key, $path) { return 1; } " "function del($key) { unset($this->keys[$key], $this->sets[$key]); return 1; } " "} " "$buffer = new ResultsV2TelemetryBuffer(42, new FakeRedis()); " "$saved = $buffer->record(['id' => '42-1', 'timestamp' => 60]); " "echo json_encode([$saved, $buffer->transactions()]);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) saved, transactions = json.loads(result.stdout) self.assertFalse(saved) self.assertIsNone(transactions) def test_order_count_runtime_and_incomplete_recovery_are_idempotent(self): source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2TelemetryBuffer.php'))}; " "class RuntimeRedis { public $keys = []; public $sets = []; " "function set($k,$v){$this->keys[$k]=$v;return 'OK';} function get($k){return $this->keys[$k]??null;} " "function exists($k){return isset($this->keys[$k])||isset($this->sets[$k]);} " "function sadd($k,$v){$n=!isset($this->sets[$k][$v]);$this->sets[$k][$v]=true;return $n?1:0;} " "function smembers($k){return array_keys($this->sets[$k]??[]);} function srem($k,$v){unset($this->sets[$k][$v]);return 1;} " "function jsonset($k,$p,$v){$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;} } " "$redis=new RuntimeRedis();$buffer=new ResultsV2TelemetryBuffer(42,$redis);$buffer->mark();" "$buffer->recordOrder('o1');$buffer->recordOrder('o1');$buffer->recordOrder('o2');" "$runtime1=$buffer->terminalRuntimeSeconds();$runtime2=$buffer->terminalRuntimeSeconds();" "$buffer->addIncompleteReason('indicator_provenance_missing');$buffer->record(['id'=>'t1','timestamp'=>1]);" "echo json_encode([$buffer->ordersCount(),$runtime1,$runtime2,$buffer->incompleteReasons(),$buffer->transactions()]);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) orders, runtime1, runtime2, reasons, transactions = json.loads(result.stdout) self.assertEqual(2, orders) self.assertEqual(runtime1, runtime2) self.assertEqual(["indicator_provenance_missing"], reasons) self.assertEqual(1, len(transactions)) def test_order_and_evidence_redis_failures_are_not_masked_and_duplicates_are_idempotent(self): source = ( f"require {json.dumps(str(ROOT / 'classes' / 'ResultsV2TelemetryBuffer.php'))}; " "class FakeRedis { public $keys=[]; public $sets=[]; public $fail=false; " "function exists($k){return isset($this->keys[$k])||isset($this->sets[$k]);} function set($k,$v){$this->keys[$k]=$v;return 'OK';} function get($k){return $this->keys[$k]??null;} " "function sadd($k,$v){if($this->fail)return 0; $new=!isset($this->sets[$k][$v]);$this->sets[$k][$v]=true;return $new?1:0;} function smembers($k){return array_keys($this->sets[$k]??[]);} " "function jsonset($k,$p,$v){return 'OK';} function srem($k,$v){unset($this->sets[$k][$v]);return 1;} function del($k){unset($this->keys[$k],$this->sets[$k]);return 1;} " "} $r=new FakeRedis(); $b=new ResultsV2TelemetryBuffer(42,$r); $first=$b->recordOrder('o1'); $dup=$b->recordOrder('o1'); $r->fail=true; $failed=$b->recordOrder('o2'); $evidence=$b->markEvidence('equity'); echo json_encode([$first,$dup,$failed,$evidence,$b->ordersCount(),$b->incompleteReasons()]);" ) result = subprocess.run(["php", "-r", source], capture_output=True, text=True) self.assertEqual(0, result.returncode, result.stderr) first, duplicate, failed, evidence, count, reasons = json.loads(result.stdout) self.assertTrue(first) self.assertTrue(duplicate) self.assertFalse(failed) self.assertFalse(evidence) self.assertEqual(1, count) self.assertIn("order_write_failed", reasons) self.assertIn("equity_evidence_write_failed", reasons) if __name__ == "__main__": unittest.main()