test_persistent_stdio.mojo (6812B)
1 from std.collections import List 2 from safe_tempdir import SafeTempDir 3 from std.testing import TestSuite, assert_equal, assert_true 4 5 from fixture_assertions import load_scenario_request_json 6 from hyf_runtime.startup import ( 7 RuntimeStartupContext, 8 RuntimeStartupInput, 9 resolve_startup_context, 10 ) 11 from hyf_stdio.server import MAX_FRAME_BYTES, run_stdio_session 12 13 14 def _context(temp_dir: String) raises -> RuntimeStartupContext: 15 return resolve_startup_context( 16 RuntimeStartupInput( 17 env_paths_profile="repo_local", 18 env_repo_local_base_root=temp_dir, 19 user_home="/home/unused", 20 argv=List[String](), 21 ) 22 ) 23 24 25 def test_persistent_session_processes_multiple_frames() raises: 26 with SafeTempDir() as temp_dir: 27 var context = _context(temp_dir) 28 var frames = List[String]() 29 frames.append(load_scenario_request_json("scenarios/status_ok.json")) 30 frames.append( 31 load_scenario_request_json("scenarios/capabilities_ok.json") 32 ) 33 frames.append(load_scenario_request_json("scenarios/status_ok.json")) 34 var responses = run_stdio_session(frames, context) 35 assert_equal(len(responses), 3) 36 for response in responses: 37 assert_true(response.find('"ok":true') >= 0) 38 39 40 def test_session_recovers_from_malformed_frame() raises: 41 with SafeTempDir() as temp_dir: 42 var context = _context(temp_dir) 43 var frames = List[String]() 44 frames.append(load_scenario_request_json("scenarios/status_ok.json")) 45 frames.append("{not valid json") 46 frames.append(load_scenario_request_json("scenarios/status_ok.json")) 47 var responses = run_stdio_session(frames, context) 48 assert_equal(len(responses), 3) 49 assert_true(responses[0].find('"ok":true') >= 0) 50 assert_true(responses[1].find("invalid_request") >= 0) 51 assert_true(responses[2].find('"ok":true') >= 0) 52 53 54 def main() raises: 55 TestSuite.discover_tests[__functions_in_module()]().run() 56 57 58 def test_session_preserves_order_and_recovers_after_malformed() raises: 59 with SafeTempDir() as temp_dir: 60 var context = _context(temp_dir) 61 var frames = List[String]() 62 frames.append(load_scenario_request_json("scenarios/status_ok.json")) 63 frames.append("{bad json") 64 frames.append( 65 load_scenario_request_json("scenarios/capabilities_ok.json") 66 ) 67 var responses = run_stdio_session(frames, context) 68 assert_equal(len(responses), 3) 69 assert_true(responses[0].find('"ok":true') >= 0) 70 assert_true(responses[1].find("invalid_request") >= 0) 71 assert_true(responses[2].find('"ok":true') >= 0) 72 assert_true(responses[2].find("business_capabilities") >= 0) 73 74 75 def test_session_rejects_oversized_frame() raises: 76 with SafeTempDir() as temp_dir: 77 var context = _context(temp_dir) 78 var huge = String("") 79 for _ in range(MAX_FRAME_BYTES + 1): 80 huge += "x" 81 assert_true(huge.byte_length() == MAX_FRAME_BYTES + 1) 82 var frames = List[String]() 83 frames.append(huge) 84 frames.append(load_scenario_request_json("scenarios/status_ok.json")) 85 var responses = run_stdio_session(frames, context) 86 assert_true(responses[0].find("size limit") >= 0) 87 assert_true(responses[1].find('"ok":true') >= 0) 88 89 90 def test_session_accepts_frame_at_configured_limit() raises: 91 with SafeTempDir() as temp_dir: 92 var context = _context(temp_dir) 93 var at_limit = String("") 94 for _ in range(MAX_FRAME_BYTES): 95 at_limit += "x" 96 assert_true(at_limit.byte_length() == MAX_FRAME_BYTES) 97 var frames = List[String]() 98 frames.append(at_limit) 99 frames.append(load_scenario_request_json("scenarios/status_ok.json")) 100 var responses = run_stdio_session(frames, context) 101 assert_true(responses[0].find("size limit") < 0) 102 assert_true(responses[1].find('"ok":true') >= 0) 103 104 105 def test_farm_update_operation_is_gated_until_enabled() raises: 106 with SafeTempDir() as temp_dir: 107 var context = _context(temp_dir) 108 var request = ( 109 '{"version":1,"request_id":"farm-op-1","capability":"farm_update.interpret",' 110 '"input":{"source":{"source_id":"s1","revision":"r1",' 111 '"text":"Got about 80 lb of Roma tomatoes.",' 112 '"source_time":"2026-09-21T09:00:00-07:00","timezone":"America/Vancouver",' 113 '"actor_id":"farm-1","farm_id":"farm-1"}}}' 114 ) 115 var frames = List[String]() 116 frames.append(request) 117 var responses = run_stdio_session(frames, context) 118 assert_true(responses[0].find("capability_disabled") >= 0) 119 120 121 def test_buyer_match_operation_is_gated_until_enabled() raises: 122 with SafeTempDir() as temp_dir: 123 var context = _context(temp_dir) 124 var request = ( 125 '{"version":1,"request_id":"match-op-1","capability":"buyer_request.match",' 126 '"input":{"need":{"need_id":"n1"},"snapshots":[]}}' 127 ) 128 var frames = List[String]() 129 frames.append(request) 130 var responses = run_stdio_session(frames, context) 131 assert_true(responses[0].find("capability_disabled") >= 0) 132 133 134 from fixture_loader import load_fixture_json_file as _load_fixture_json 135 from std.pathlib import Path as _Path, _dir_of_current_file as _dir_of 136 137 138 def test_new_operation_wire_frames_are_gated() raises: 139 var wire_dir = _dir_of() / "fixtures" / "hyf_v1_jev" / "wire" 140 var manifest = _load_fixture_json(wire_dir / "manifest.json") 141 assert_equal(manifest["spec_id"].string_value(), "hyf_v1_jev") 142 with SafeTempDir() as temp_dir: 143 var context = _context(temp_dir) 144 for name in manifest["scenarios"].array_items(): 145 var scenario = _load_fixture_json(wire_dir / name.string_value()) 146 var frames = List[String]() 147 frames.append(scenario["request"].string_value()) 148 var responses = run_stdio_session(frames, context) 149 assert_true( 150 responses[0].find( 151 scenario["expected_error_code"].string_value() 152 ) 153 >= 0 154 ) 155 156 157 def test_long_session_processes_many_frames_in_order() raises: 158 with SafeTempDir() as temp_dir: 159 var context = _context(temp_dir) 160 var frames = List[String]() 161 for index in range(50): 162 frames.append( 163 load_scenario_request_json("scenarios/status_ok.json") 164 ) 165 var responses = run_stdio_session(frames, context) 166 assert_equal(len(responses), 50) 167 for response in responses: 168 assert_true(response.find('"ok":true') >= 0) 169 assert_true(response.find("status-fixture-1") >= 0)