Setting the file. One moment.
Test Validate Result · LLM To Bedrock · aws/agent-toolkit-for-aws · Skills Docs
Back to the top of the page 81
Creating Amazon Aurora Db Cluster With Instances
104
Routing Traffic With Route53 And CloudFront
Resilience Program Design
Creating API Gateway Stage
— line 168
This file
Number 30.33
Position 33 of 35
Type Python
Size 13 KB
Lines 352 scripts/ test_validate_result.py
Python · 352 lines · 13 KB
).parent
/
"schemas"
14
15
16 def schema (name):
17 return json.loads(( SCHEMAS / f " { name } .json" ).read_text())
18
19
20 def validate (name, data):
21 return vr.jsonschema.Draft202012Validator(schema(name)).is_valid(data)
22
23
24 # ---------- golden payloads (verbatim shapes from the agent docs) ----------
25
26 GOLDEN_ANALYSIS = {
27 "summary" : "LangChain + langchain-openai detected. 2 files need modification." ,
28 "source_code_path" : "/repo" , "migration_plan_path" : "/repo/.migration/0610-1200" ,
29 "app_language" : "Python" , "ai_framework" : "LangChain" ,
30 "ai_framework_version" : "langchain==0.1.14" , "source_provider" : "openai" ,
31 "source_models" : [ "gpt-4o" ],
32 "target_models" : [ "gpt-4o -> us.anthropic.claude-sonnet-4-6" ],
33 "same_model_family" : False , "bedrock_provider_available" : True ,
34 "prompt_locations" : [ "app.py:42 : SYSTEM_PROMPT constant" ],
35 "prompt_patterns" : "hardcoded" ,
36 "special_patterns" : { "streaming" : True , "function_calling" : False ,
37 "embeddings" : False , "vision" : False },
38 "code_change_sites" : 2 ,
39 "files_to_modify" : [ "app.py: replace ChatOpenAI with ChatBedrockConverse" ],
40 "dependencies_to_replace" : [ "langchain-openai -> langchain-aws" ],
41 "log_files_found" : "none" , "errors" : "none" , "behavior_deltas" : [],
42 "source_baseline_available" : False ,
43 }
44
45 GOLDEN_INGESTION = {
46 "summary" : "Built golden dataset with 9 cases from the code template." ,
47 "golden_dataset_path" : "/repo/.saws-migrate/golden-dataset/prompts.jsonl" ,
48 "prompt_template_path" : "/repo/.saws-migrate/golden-dataset/templates/prompt_template.txt" ,
49 "total_golden_cases" : 9 , "golden_from_logs" : 0 , "golden_from_user" : 0 ,
50 "golden_from_code_confirmed" : 9 , "vision_test_images" : 0 , "log_format" : "none" ,
51 "coverage_level" : "code-confirmed" , "use_case_type" : "text-only" ,
52 "gaps" : [ "No production traffic data" ], "pii_detected" : False ,
53 "pii_action" : "not-applicable" , "errors" : "none" ,
54 }
55
56 GOLDEN_INGESTION_ZERO = { ** GOLDEN_INGESTION ,
57 "summary" : "No LLM call sites — zero golden cases." ,
58 "golden_dataset_path" : "" , "prompt_template_path" : "" , "total_golden_cases" : 0 ,
59 "golden_from_code_confirmed" : 0 , "coverage_level" : "none" ,
60 "use_case_type" : "unknown" ,
61 "gaps" : [ "No LLM call sites in source" ]}
62
63 GOLDEN_EVAL = {
64 "eval_report_path" : "/repo/.saws-migrate/eval-results/" ,
65 "pass_rate" : 0.89 , "total_cases" : 9 , "failures" : 1 ,
66 "notes" : "live_source_baseline_used_model: \n 1 prompt needs manual review." ,
67 "live_source_baseline" : True , "judge_model" : "claude-opus-4-8" ,
68 "source_baseline_quality" : "good" ,
69 }
70
71 GOLDEN_EVAL_ZERO = { ** GOLDEN_EVAL , "pass_rate" : 1.0 , "total_cases" : 0 ,
72 "failures" : 0 , "notes" : "no_golden_cases: true" ,
73 "live_source_baseline" : False , "source_baseline_quality" : "unknown" }
74
75 SHA = "a" * 40
76
77 GOLDEN_REWRITE = {
78 "branch_name" : "bedrock-migration" ,
79 "files_changed" : [ "app.py" , "pyproject.toml" ],
80 "dependencies_updated" : [ "langchain-openai -> langchain-aws" ],
81 "notes" : "5 tests generated, 5/5 passing in clean checkout." ,
82 "behavior_delta_decisions" : [
83 { "delta_type" : "temperature-range-mismatch" , "location" : "app.py:95" ,
84 "resolution_chosen" : "range_narrowed_1" , "source" : "user_question" }
85 ],
86 "baseline_parent_sha" : SHA , "branch_tip_sha" : "b" * 40 ,
87 }
88
89 GOLDEN_DELTA_DECISIONS = [
90 { "delta_type" : "temperature-range-mismatch" , "location" : "app.py:95" ,
91 "resolution_chosen" : "range_narrowed_1" , "source" : "user_question" },
92 ]
93
94
95 @pytest.mark.parametrize ( "name,payload" , [
96 ( "analysis" , GOLDEN_ANALYSIS ),
97 ( "ingestion" , GOLDEN_INGESTION ),
98 ( "ingestion" , GOLDEN_INGESTION_ZERO ),
99 ( "eval" , GOLDEN_EVAL ),
100 ( "eval" , GOLDEN_EVAL_ZERO ),
101 ( "rewrite" , GOLDEN_REWRITE ),
102 ( "delta-decisions" , GOLDEN_DELTA_DECISIONS ),
103 ( "delta-decisions" , []),
104 ])
105 def test_golden_payloads_validate (name, payload):
106 assert validate(name, payload), f "golden { name } payload must validate"
107
108
109 # ---------- required enforcement (parameterized over each schema's required) ----------
110
111 def required_fields (name):
112 s = schema(name)
113 branch = s[ "oneOf" ][ 0 ] if "oneOf" in s else s
114 return branch[ "required" ]
115
116 GOLDENS = { "analysis" : GOLDEN_ANALYSIS , "ingestion" : GOLDEN_INGESTION ,
117 "eval" : GOLDEN_EVAL , "rewrite" : GOLDEN_REWRITE }
118
119
120 @pytest.mark.parametrize ( "name" , list ( GOLDENS ))
121 def test_empty_object_fails (name):
122 assert not validate(name, {})
123
124
125 @pytest.mark.parametrize ( "name,field" , [
126 (name, field) for name in GOLDENS for field in required_fields(name)
127 ])
128 def test_each_required_field_enforced (name, field):
129 # Regression for the dropped-`required` bug AND the loose-JS gaps:
130 # evaluator missing live_source_baseline/judge_model/source_baseline_quality,
131 # analyzer missing any typed field, must all be rejected.
132 payload = copy.deepcopy( GOLDENS [name])
133 del payload[field]
134 assert not validate(name, payload), f " { name } without { field } must fail"
135
136
137 def test_special_patterns_requires_all_four_booleans ():
138 payload = copy.deepcopy( GOLDEN_ANALYSIS )
139 del payload[ "special_patterns" ][ "vision" ]
140 assert not validate( "analysis" , payload)
141
142
143 # ---------- control states (per schema) ----------
144
145 @pytest.mark.parametrize ( "name,reason" , [
146 ( "analysis" , "model_access" ), ( "analysis" , "model_unresolvable" ),
147 ( "analysis" , "assess_output_missing" ),
148 ( "eval" , "model_access" ), ( "eval" , "source_key_auth" ), ( "eval" , "model_unresolvable" ),
149 ( "rewrite" , "model_access" ), ( "rewrite" , "source_key_auth" ), ( "rewrite" , "model_unresolvable" ),
150 ])
151 def test_blocked_reasons_in_own_schema (name, reason):
152 assert validate(name, { "blocked" : { "reason" : reason, "detail" : "x" }})
153
154
155 @pytest.mark.parametrize ( "name,reason" , [
156 ( "eval" , "assess_output_missing" ), # analyzer-only reason
157 ( "rewrite" , "assess_output_missing" ),
158 ( "analysis" , "source_key_auth" ), # eval/rewrite-only reason
159 ])
160 def test_blocked_reason_from_other_schema_fails (name, reason):
161 assert not validate(name, { "blocked" : { "reason" : reason, "detail" : "x" }})
162
163
164 def test_ingestion_has_no_blocked_branch ():
165 assert not validate( "ingestion" , { "blocked" : { "reason" : "model_access" , "detail" : "x" }})
166
167
168 def test_blocked_without_detail_fails ():
169 assert not validate( "analysis" , { "blocked" : { "reason" : "model_access" }})
170
171
172 def test_payload_mixed_with_blocked_fails ():
173 payload = { ** GOLDEN_EVAL , "blocked" : { "reason" : "model_access" , "detail" : "x" }}
174 assert not validate( "eval" , payload)
175
176
177 def test_partial_only_in_eval ():
178 partial = { "partial" : { "completed" : 7 , "total" : 20 , "reason" : "throttled" }}
179 assert validate( "eval" , partial)
180 assert not validate( "analysis" , partial)
181 assert not validate( "rewrite" , partial)
182 assert not validate( "ingestion" , partial)
183
184
185 def test_rewrite_missing_sha_fields_fails ():
186 for field in ( "baseline_parent_sha" , "branch_tip_sha" ):
187 payload = copy.deepcopy( GOLDEN_REWRITE )
188 del payload[field]
189 assert not validate( "rewrite" , payload)
190
191
192 def test_delta_decisions_entry_missing_resolution_fails ():
193 bad = [{ "delta_type" : "x" , "location" : "a.py:1" , "source" : "user_question" }]
194 assert not validate( "delta-decisions" , bad)
195 assert not validate( "delta-decisions" , { "not" : "an array" })
196
197
198 # ---------- CLI behavior ----------
199
200 def write (tmp_path, name, data):
201 f = tmp_path / name
202 f.write_text(json.dumps(data))
203 return str (f)
204
205
206 def test_cli_control_lines (tmp_path, capsys):
207 assert vr.main([ "--schema" , "eval" , write(tmp_path, "ok.json" , GOLDEN_EVAL )]) == 0
208 assert "RESULT=valid CONTROL=ok" in capsys.readouterr().out
209
210 blocked = { "blocked" : { "reason" : "model_access" , "detail" : "enable in console" }}
211 assert vr.main([ "--schema" , "eval" , write(tmp_path, "b.json" , blocked)]) == 0
212 assert "CONTROL=blocked REASON=model_access" in capsys.readouterr().out
213
214 partial = { "partial" : { "completed" : 7 , "total" : 20 , "reason" : "throttled" }}
215 assert vr.main([ "--schema" , "eval" , write(tmp_path, "p.json" , partial)]) == 0
216 assert "CONTROL=partial COMPLETED=7 TOTAL=20" in capsys.readouterr().out
217
218
219 def test_cli_invalid_reports_field_paths (tmp_path, capsys):
220 payload = copy.deepcopy( GOLDEN_EVAL )
221 del payload[ "judge_model" ]
222 assert vr.main([ "--schema" , "eval" , write(tmp_path, "bad.json" , payload)]) == 1
223 out = capsys.readouterr().out
224 assert "RESULT=invalid" in out
225 assert "judge_model" in out
226
227
228 def test_cli_missing_file_exits_2 (capsys):
229 assert vr.main([ "--schema" , "eval" , "/nonexistent/x.json" ]) == 2
230
231
232 def test_cli_delta_decisions_always_control_ok (tmp_path, capsys):
233 assert vr.main([ "--schema" , "delta-decisions" , write(tmp_path, "d.json" , [])]) == 0
234 assert "CONTROL=ok" in capsys.readouterr().out
235
236
237 # ---------- run-context comparison ----------
238
239 RUN_CONTEXT = {
240 "repo_root" : "/repo" , "migration_dir" : "/repo/.migration/0610-1200" ,
241 "region" : "us-east-1" , "aws_profile" : "" , "aws_account" : "123456789012" ,
242 "repo_head_sha" : SHA , "repo_branch" : "main" , "repo_dirty_sha256" : "" ,
243 "target_models" : [{ "source_model" : "gpt-4o" ,
244 "aws_model_id" : "us.anthropic.claude-sonnet-4-6" ,
245 "use_case" : "primary" }],
246 "resolved_model_overrides" : {},
247 "source_provider" : "openai" , "source_baseline_available" : True ,
248 "source_key_sha256" : "c" * 64 ,
249 "log_files" : [{ "path" : "logs/traces.jsonl" , "sha256" : "d" * 64 }],
250 "max_golden_cases" : 200 , "assess_design_sha256" : "e" * 64 ,
251 "report_date_suffix" : "2026-06-10" ,
252 "schema_version" : 1 , "plugin_version" : "1.0.1" ,
253 }
254
255
256 def test_run_context_identical_matches (tmp_path, capsys):
257 s = write(tmp_path, "saved.json" , RUN_CONTEXT )
258 c = write(tmp_path, "current.json" , RUN_CONTEXT )
259 assert vr.main([ "--check-run-context" , s, "--current" , c]) == 0
260 assert "RUN_CONTEXT=match" in capsys.readouterr().out
261
262
263 MUTATIONS = {
264 "region" : "eu-west-1" , "aws_profile" : "prod" , "aws_account" : "999999999999" ,
265 "migration_dir" : "/repo/.migration/0611-0900" , "repo_head_sha" : "f" * 40 ,
266 "repo_branch" : "develop" , "repo_dirty_sha256" : "1" * 64 ,
267 "source_provider" : "google" , "source_key_sha256" : "0" * 64 ,
268 "max_golden_cases" : 50 , "assess_design_sha256" : "2" * 64 ,
269 "schema_version" : 2 , "plugin_version" : "1.0.2" ,
270 }
271
272
273 @pytest.mark.parametrize ( "field" , list ( MUTATIONS ))
274 def test_single_field_mismatch_named (tmp_path, capsys, field):
275 cur = copy.deepcopy( RUN_CONTEXT )
276 cur[field] = MUTATIONS [field]
277 s = write(tmp_path, "saved.json" , RUN_CONTEXT )
278 c = write(tmp_path, "current.json" , cur)
279 assert vr.main([ "--check-run-context" , s, "--current" , c]) == 1
280 out = capsys.readouterr().out
281 assert "RUN_CONTEXT=mismatch" in out
282 assert f "MISMATCH $. { field } " in out
283
284
285 def test_nested_mismatch_reported (tmp_path, capsys):
286 cur = copy.deepcopy( RUN_CONTEXT )
287 cur[ "log_files" ][ 0 ][ "sha256" ] = "9" * 64
288 s = write(tmp_path, "s.json" , RUN_CONTEXT )
289 c = write(tmp_path, "c.json" , cur)
290 assert vr.main([ "--check-run-context" , s, "--current" , c]) == 1
291 assert "log_files" in capsys.readouterr().out
292
293
294 def test_key_hash_mismatch_never_prints_hashes (tmp_path, capsys):
295 cur = copy.deepcopy( RUN_CONTEXT )
296 cur[ "source_key_sha256" ] = "0" * 64
297 s = write(tmp_path, "s.json" , RUN_CONTEXT )
298 c = write(tmp_path, "c.json" , cur)
299 vr.main([ "--check-run-context" , s, "--current" , c])
300 out = capsys.readouterr().out
301 assert "MISMATCH $.source_key_sha256 differs" in out
302 assert "c" * 64 not in out and "0" * 64 not in out
303
304
305 def test_report_date_suffix_excluded_from_comparison (tmp_path, capsys):
306 cur = copy.deepcopy( RUN_CONTEXT )
307 cur[ "report_date_suffix" ] = "2026-06-11"
308 s = write(tmp_path, "s.json" , RUN_CONTEXT )
309 c = write(tmp_path, "c.json" , cur)
310 assert vr.main([ "--check-run-context" , s, "--current" , c]) == 0
311
312
313 def test_unknown_extra_key_is_mismatch (tmp_path, capsys):
314 cur = copy.deepcopy( RUN_CONTEXT )
315 cur[ "future_field" ] = "surprise"
316 s = write(tmp_path, "s.json" , RUN_CONTEXT )
317 c = write(tmp_path, "c.json" , cur)
318 assert vr.main([ "--check-run-context" , s, "--current" , c]) == 1
319 assert "future_field" in capsys.readouterr().out
320
321
322 def test_run_context_missing_file_exits_2 (tmp_path, capsys):
323 c = write(tmp_path, "c.json" , RUN_CONTEXT )
324 assert vr.main([ "--check-run-context" , "/nonexistent.json" , "--current" , c]) == 2
325
326
327 def test_analysis_valid_without_optional_fields ():
328 assert validate( "analysis" , GOLDEN_ANALYSIS )
329
330
331 def test_empty_object_mismatch_detected (tmp_path, capsys):
332 """Empty dict in saved vs absent key in current must be a mismatch."""
333 saved = copy.deepcopy( RUN_CONTEXT )
334 saved[ "resolved_model_overrides" ] = {}
335 cur = copy.deepcopy( RUN_CONTEXT )
336 cur.pop( "resolved_model_overrides" , None )
337 s = write(tmp_path, "s.json" , saved)
338 c = write(tmp_path, "c.json" , cur)
339 assert vr.main([ "--check-run-context" , s, "--current" , c]) == 1
340 assert "resolved_model_overrides" in capsys.readouterr().out
341
342
343 def test_empty_array_mismatch_detected (tmp_path, capsys):
344 """Empty list in saved vs absent key in current must be a mismatch."""
345 saved = copy.deepcopy( RUN_CONTEXT )
346 saved[ "log_files" ] = []
347 cur = copy.deepcopy( RUN_CONTEXT )
348 cur.pop( "log_files" , None )
349 s = write(tmp_path, "s.json" , saved)
350 c = write(tmp_path, "c.json" , cur)
351 assert vr.main([ "--check-run-context" , s, "--current" , c]) == 1
352 assert "log_files" in capsys.readouterr().out