Setting the file. One moment. Add Capabilities
81
Creating Amazon Aurora Db Cluster With Instances
104
Routing Traffic With Route53 And CloudFront
Resilience Program Design
Creating API Gateway Stage
def test_context_and_output_limits_filter_candidates
— line 138
This file
- Number
- 23.72
- Position
- 72 of 81
- Type
- Python
- Size
- 13 KB
- Lines
- 421
scripts/test_model_recommendation.py
Python·421 lines·13 KB
15 "refusal_handling",
16 "sampling_parameters",
17 "tokenizer_rebaseline",
18}
19
20
21def _workload(**overrides):
22 workload = {
23 "workload_id": "support-agent",
24 "source": {
25 "provider": "anthropic",
26 "model_ids": ["claude-3-7-sonnet-latest"],
27 "sdk": "anthropic",
28 "api_surface": "messages",
29 "source_paths": ["src/agent.py"],
30 },
31 "requirements": {
32 "priority": "balanced",
33 "critical_features": ["tool_use"],
34 },
35 "detected_features": [],
36 }
37 for key, value in overrides.items():
38 if key in {"source", "requirements"}:
39 workload[key].update(value)
40 else:
41 workload[key] = value
42 return workload
43
44
45def _input(workload=None):
46 return {
47 "schema_version": 2,
48 "region": "us-east-1",
49 "primary_unit": "support-agent",
50 "workloads": [workload or _workload()],
51 }
52
53
54def _recommend(workload=None):
55 return model_recommendation.recommend(_input(workload))["workloads"][
56 "support-agent"
57 ]
58
59
60def _codes(items):
61 return {item["code"] for item in items}
62
63
64def test_default_anthropic_messages_prefers_mantle_and_sonnet5():
65 # Sonnet 5 (launched 2026-06-30) supports Mantle Messages, so the balanced
66 # priority no longer has to escalate to Opus for Messages continuity.
67 rec = _recommend()
68
69 assert rec["decision_status"] == "recommended"
70 assert rec["primary_model"] == "anthropic.claude-sonnet-5"
71 assert rec["api_path"] == "mantle_messages"
72 assert rec["invocation_model_id"] == "anthropic.claude-sonnet-5"
73 assert rec["model_identity"]["path_model_id"] == rec["primary_model"]
74 assert rec["verification"]["probe_status"] == "not_run"
75
76
77def test_cost_priority_on_messages_uses_clean_mantle_haiku_id():
78 rec = _recommend(_workload(requirements={"priority": "cost"}))
79
80 assert rec["api_path"] == "mantle_messages"
81 assert rec["primary_model"] == "anthropic.claude-haiku-4-5"
82 assert rec["model_identity"]["requires_cris"] is False
83
84
85def test_governance_selects_runtime_converse_when_messages_is_a_preference():
86 rec = _recommend(
87 _workload(
88 requirements={"governance": ["guardrails", "invocation_logging"]}
89 )
90 )
91
92 assert rec["decision_status"] == "recommended"
93 assert rec["api_path"] == "runtime_converse"
94 assert rec["primary_model"] == "anthropic.claude-sonnet-5"
95
96
97def test_required_messages_and_governance_require_user_decision():
98 rec = _recommend(
99 _workload(
100 requirements={
101 "preserve_messages_api": True,
102 "governance": ["guardrails"],
103 }
104 )
105 )
106
107 assert rec["decision_status"] == "decision_required"
108 assert rec["primary_model"] is None
109 assert rec["api_path"] is None
110 assert {option["api_path"] for option in rec["decision_options"]} == {
111 "mantle_messages",
112 "runtime_converse",
113 }
114 assert "model_path_decision_required" in _codes(rec["blocks"])
115
116
117def test_conflicting_paths_without_catalog_candidates_are_rejected():
118 workload = _workload(
119 requirements={
120 "preserve_messages_api": True,
121 "governance": ["guardrails"],
122 "min_context_tokens": 2000000,
123 }
124 )
125
126 with pytest.raises(ValueError, match="no candidate for every conflicting"):
127 _recommend(workload)
128
129
130def test_native_payload_selects_runtime_invoke():
131 rec = _recommend(
132 _workload(requirements={"requires_native_payload": True})
133 )
134
135 assert rec["api_path"] == "runtime_invoke"
136
137
138def test_context_and_output_limits_filter_candidates():
139 rec = _recommend(
140 _workload(
141 requirements={
142 "priority": "cost",
143 "min_context_tokens": 1000000,
144 "expected_output_tokens": 100000,
145 }
146 )
147 )
148
149 assert rec["model_identity"]["model_key"] == "claude_sonnet_5"
150 assert all(
151 option["model_key"] != "claude_haiku_4_5"
152 for option in rec["alternatives"]
153 )
154
155
156def test_thinking_requirement_filters_haiku():
157 rec = _recommend(
158 _workload(
159 requirements={
160 "priority": "cost",
161 "thinking_enabled": True,
162 }
163 )
164 )
165
166 assert rec["model_identity"]["model_key"] == "claude_sonnet_5"
167
168
169def test_source_version_hop_requires_explicit_feature_scan():
170 rec = _recommend()
171
172 assert rec["source_analysis"] == {
173 "detected_version": "3.7",
174 "target_version": "5.0",
175 "version_changed": True,
176 }
177 assert "claude_version_hop" in _codes(rec["migration_deltas"])
178 assert "version_scan_incomplete" in _codes(rec["blocks"])
179 assert {
180 feature
181 for feature, status in rec["feature_assessment"].items()
182 if status == "unknown"
183 } == VERSION_SCAN_FEATURES
184
185
186def test_explicit_absent_status_clears_version_scan_block():
187 workload = _workload(
188 feature_status={feature: "absent" for feature in VERSION_SCAN_FEATURES}
189 )
190 rec = _recommend(workload)
191
192 assert "version_scan_incomplete" not in _codes(rec["blocks"])
193
194
195def test_detected_version_features_emit_blocks_and_tuning():
196 rec = _recommend(
197 _workload(
198 detected_features=[
199 "budget_tokens",
200 "sampling_parameters",
201 "assistant_prefill",
202 "refusal_handling",
203 "tokenizer_rebaseline",
204 "max_tokens_headroom",
205 ]
206 )
207 )
208
209 assert {
210 "budget_tokens_removed",
211 "sampling_parameters_removed",
212 "assistant_prefill_removed",
213 }.issubset(_codes(rec["blocks"]))
214 assert {
215 "refusal_handling",
216 "tokenizer_rebaseline",
217 "max_tokens_headroom",
218 }.issubset(_codes(rec["tuning"]))
219
220
221@pytest.mark.parametrize(
222 "requirements,expected_path",
223 [
224 ({"preserve_messages_api": True}, "mantle_messages"),
225 ({"governance": ["guardrails"]}, "runtime_converse"),
226 ],
227)
228def test_structured_output_uses_portable_forced_tool_guidance(
229 requirements, expected_path
230):
231 rec = _recommend(
232 _workload(
233 requirements=requirements,
234 detected_features=["structured_output"],
235 )
236 )
237
238 assert rec["api_path"] == expected_path
239 assert "structured_output_portable_pattern" in _codes(rec["blocks"])
240 assert "structured_output" in _codes(rec["migration_deltas"])
241 finding = next(
242 item
243 for item in rec["blocks"]
244 if item["code"] == "structured_output_portable_pattern"
245 )
246 assert "forced tool without strict" in finding["remediation"]
247
248
249def test_structured_output_and_citations_emit_conflict():
250 rec = _recommend(
251 _workload(detected_features=["structured_output", "citations"])
252 )
253
254 assert "structured_output_citations_conflict" in _codes(rec["blocks"])
255
256
257def test_agent_features_produce_architecture_and_trajectory_requirements():
258 rec = _recommend(
259 _workload(
260 requirements={"critical_features": ["agentic", "tool_use"]},
261 detected_features=["agent_infra", "server_tools"],
262 )
263 )
264
265 assert rec["evaluation"]["mode"] == "trajectory"
266 assert {item["feature"] for item in rec["architecture_impacts"]} == {
267 "agent_infra",
268 "server_tools",
269 }
270 assert rec["compatibility"]["rearchitecture"] == [
271 "agent_infra",
272 "server_tools",
273 ]
274
275
276@pytest.mark.parametrize(
277 "requirements,expected",
278 [
279 ({"data_residency": "unknown"}, None),
280 (
281 {"data_residency": "global_allowed"},
282 "global.anthropic.claude-sonnet-5",
283 ),
284 (
285 {"data_residency": "geo_required", "cris_geography": "eu"},
286 "eu.anthropic.claude-sonnet-5",
287 ),
288 (
289 {
290 "inference_profile_id": "arn:aws:bedrock:us-east-1:123:inference-profile/custom"
291 },
292 "arn:aws:bedrock:us-east-1:123:inference-profile/custom",
293 ),
294 ],
295)
296def test_runtime_cris_resolution(requirements, expected):
297 requirements["governance"] = ["guardrails"]
298 rec = _recommend(_workload(requirements=requirements))
299
300 assert rec["api_path"] == "runtime_converse"
301 assert rec["invocation_model_id"] == expected
302 assert rec["verification"]["invocation_model_id"] == expected
303
304
305# The intended treatment of every provider the input schema allows, written here rather than derived
306# from the code — a test that reads its expectation out of the constant it is checking cannot catch a
307# drift in that constant. "anthropic" means a real recommendation from a module that understands the
308# source; "generic" means a recommendation that carries `provider_module_pending` and stays
309# provisional; "openai" has its own module and its own test suite.
310PROVIDER_TREATMENT = {
311 "anthropic": "anthropic",
312 "none": "anthropic", # no detected provider -> Bedrock-native pool
313 "unknown": "anthropic",
314 "openai": "openai",
315 "azure_openai": "generic",
316 "google_genai": "generic",
317 "bedrock": "generic",
318}
319
320
321def test_every_schema_provider_routes_and_is_labelled_honestly():
322 """Every provider the schema allows must route to a real module and be labelled honestly.
323
324 The orchestrator sends everything non-OpenAI to the Anthropic module, which then classifies the
325 source against its own ANTHROPIC_POOL. Two places, one rule — so this pins the rule itself, and
326 fails if either side moves or if a provider is added to the schema without deciding its
327 treatment."""
328 scripts = pathlib.Path(model_recommendation.__file__).parent
329 schema = json.loads(
330 (scripts / "schemas" / "model-recommendation-input.json").read_text()
331 )
332 # `source` is a $ref into $defs, so the enum is read from there rather than inline.
333 assert schema["properties"]["workloads"]["items"]["properties"]["source"] == {
334 "$ref": "#/$defs/source"
335 }
336 providers = schema["$defs"]["source"]["properties"]["provider"]["enum"]
337 assert set(providers) == set(PROVIDER_TREATMENT), (
338 "a provider was added to the schema without deciding how it is treated"
339 )
340 assert anthropic_model_recommendation.ANTHROPIC_POOL == {
341 p for p, t in PROVIDER_TREATMENT.items() if t == "anthropic"
342 }
343
344 for provider, treatment in PROVIDER_TREATMENT.items():
345 module = "openai" if treatment == "openai" else "anthropic"
346 assert model_recommendation._provider_module(provider) == module, provider
347 if module == "openai":
348 continue
349 rec = _recommend(_workload(source={"provider": provider}))
350 assert rec["provider_module"] == treatment, provider
351 assert ("provider_module_pending" in _codes(rec["blocks"])) == (
352 treatment == "generic"
353 ), provider
354
355
356def test_future_provider_is_explicitly_provisional():
357 # OpenAI now has a real provider module; Azure OpenAI remains the pending case.
358 rec = _recommend(
359 _workload(
360 source={
361 "provider": "azure_openai",
362 "model_ids": ["gpt-5.4"],
363 "sdk": "azure_openai",
364 "api_surface": "responses",
365 }
366 )
367 )
368
369 assert rec["provider_module"] == "generic"
370 assert "provider_module_pending" in _codes(rec["blocks"])
371
372
373def test_duplicate_workload_ids_are_rejected():
374 data = _input()
375 data["workloads"].append(_workload())
376
377 with pytest.raises(ValueError, match="duplicate workload_id"):
378 model_recommendation.recommend(data)
379
380
381def test_input_and_output_match_schemas():
382 scripts = pathlib.Path(model_recommendation.__file__).parent
383 input_data = _input()
384 result = model_recommendation.recommend(input_data)
385
386 input_schema = json.loads(
387 (scripts / "schemas" / "model-recommendation-input.json").read_text()
388 )
389 output_schema = json.loads(
390 (scripts / "schemas" / "model-recommendation.json").read_text()
391 )
392 jsonschema.validate(input_data, input_schema)
393 jsonschema.validate(result, output_schema)
394
395
396def test_catalog_records_path_specific_ids_and_limits():
397 catalog = model_recommendation.load_catalog()
398
399 assert catalog["verified_at"] == "2026-07-21"
400 assert catalog["verified_region"] == "us-east-1"
401 assert (
402 catalog["models"]["claude_sonnet_5"]["paths"]["mantle_messages"][
403 "available"
404 ]
405 is True
406 )
407 assert (
408 catalog["models"]["claude_haiku_4_5"]["paths"]["mantle_messages"][
409 "model_id"
410 ]
411 == "anthropic.claude-haiku-4-5"
412 )
413 assert (
414 catalog["models"]["claude_haiku_4_5"]["paths"]["runtime_converse"][
415 "model_id"
416 ]
417 == "anthropic.claude-haiku-4-5-20251001-v1:0"
418 )
419 assert (
420 catalog["models"]["claude_opus_4_8"]["output_token_ceiling"] == 128000
421 )