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
— line 80
This file
- Number
- 23.78
- Position
- 78 of 81
- Type
- Python
- Size
- 8 KB
- Lines
- 274
scripts/test_verify_model_path.py
Python·274 lines·8 KB
15 "schema_version": 2,
16 "region": "us-east-1",
17 "primary_unit": "support-agent",
18 "workloads": [
19 {
20 "workload_id": "support-agent",
21 "source": {
22 "provider": "anthropic",
23 "model_ids": ["claude-3-7-sonnet-latest"],
24 "sdk": "anthropic",
25 "api_surface": "messages",
26 "source_paths": ["src/agent.py"],
27 },
28 "requirements": {
29 "priority": "balanced",
30 "critical_features": ["tool_use"],
31 **(requirements or {}),
32 },
33 "detected_features": [],
34 }
35 ],
36 }
37
38
39class FakeMessages:
40 def __init__(self, response=None, error=None):
41 self.calls = []
42 self.response = response or type(
43 "Response", (), {"model": "anthropic.claude-sonnet-5"}
44 )()
45 self.error = error
46
47 def create(self, **kwargs):
48 self.calls.append(kwargs)
49 if self.error:
50 raise self.error
51 return self.response
52
53
54class FakeMantleClient:
55 def __init__(self, messages=None):
56 self.messages = messages or FakeMessages()
57
58
59class FakeRuntimeClient:
60 def __init__(self):
61 self.converse_calls = []
62 self.invoke_calls = []
63
64 def converse(self, **kwargs):
65 self.converse_calls.append(kwargs)
66 return {"modelId": kwargs["modelId"]}
67
68 def invoke_model(self, **kwargs):
69 self.invoke_calls.append(kwargs)
70 return {}
71
72
73def _verify(requirements=None, **kwargs):
74 recommendation = model_recommendation.recommend(_input(requirements))
75 return verify_model_path.verify_recommendation(
76 recommendation, now=NOW, **kwargs
77 )
78
79
80def test_mantle_probe_uses_recommended_clean_id():
81 client = FakeMantleClient()
82 result = _verify(mantle_client_factory=lambda region: client)
83 verification = result["workloads"]["support-agent"]
84
85 assert verification["status"] == "passed"
86 assert verification["response_model_id"] == "anthropic.claude-sonnet-5"
87 assert client.messages.calls[0]["model"] == "anthropic.claude-sonnet-5"
88 assert client.messages.calls[0]["max_tokens"] == 8
89
90
91def test_converse_probe_uses_resolved_cris_without_substitution():
92 client = FakeRuntimeClient()
93 result = _verify(
94 {
95 "governance": ["guardrails"],
96 "data_residency": "global_allowed",
97 },
98 runtime_client_factory=lambda region: client,
99 )
100 verification = result["workloads"]["support-agent"]
101
102 assert verification["status"] == "passed"
103 assert (
104 client.converse_calls[0]["modelId"]
105 == "global.anthropic.claude-sonnet-5"
106 )
107
108
109def test_invoke_probe_uses_native_bedrock_body():
110 client = FakeRuntimeClient()
111 result = _verify(
112 {
113 "requires_native_payload": True,
114 "data_residency": "global_allowed",
115 },
116 runtime_client_factory=lambda region: client,
117 )
118 verification = result["workloads"]["support-agent"]
119 body = json.loads(client.invoke_calls[0]["body"])
120
121 assert verification["status"] == "passed"
122 assert body["anthropic_version"] == "bedrock-2023-05-31"
123 assert body["messages"][0]["content"] == verify_model_path.PROMPT
124
125
126def test_unresolved_runtime_cris_does_not_create_client():
127 calls = []
128 result = _verify(
129 {"governance": ["guardrails"]},
130 runtime_client_factory=lambda region: calls.append(region),
131 )
132 verification = result["workloads"]["support-agent"]
133
134 assert verification["status"] == "needs_resolution"
135 assert verification["invocation_model_id"] is None
136 assert calls == []
137
138
139def test_probe_failure_is_structured():
140 client = FakeMantleClient(FakeMessages(error=PermissionError("denied")))
141 result = _verify(mantle_client_factory=lambda region: client)
142 verification = result["workloads"]["support-agent"]
143
144 assert verification["status"] == "failed"
145 assert verification["error"] == {
146 "type": "PermissionError",
147 "message": "denied",
148 }
149
150
151def test_decision_required_is_not_probed():
152 result = _verify(
153 {
154 "preserve_messages_api": True,
155 "governance": ["guardrails"],
156 }
157 )
158
159 assert result["workloads"]["support-agent"]["status"] == "not_applicable"
160
161
162def test_verification_output_matches_schema():
163 result = _verify(mantle_client_factory=lambda region: FakeMantleClient())
164 schema = json.loads(
165 (
166 pathlib.Path(verify_model_path.__file__).parent
167 / "schemas"
168 / "model-verification.json"
169 ).read_text()
170 )
171
172 jsonschema.validate(
173 result,
174 schema,
175 format_checker=jsonschema.FormatChecker(),
176 )
177
178
179# --- OpenAI provider verification (mocked, no network) ---------------------
180
181
182class FakeResponses:
183 def __init__(self, response=None, error=None):
184 self.calls = []
185 self.response = response or type("Resp", (), {"model": "openai.gpt-5.6-sol"})()
186 self.error = error
187
188 def create(self, **kwargs):
189 self.calls.append(kwargs)
190 if self.error:
191 raise self.error
192 return self.response
193
194
195class FakeOpenAIClient:
196 def __init__(self, responses=None):
197 self.responses = responses or FakeResponses()
198
199
200def _openai_input(requirements=None):
201 return {
202 "schema_version": 2,
203 "region": "us-east-2",
204 "primary_unit": "openai-svc",
205 "workloads": [
206 {
207 "workload_id": "openai-svc",
208 "source": {
209 "provider": "openai",
210 "model_ids": ["gpt-5.4"],
211 "sdk": "openai",
212 "api_surface": "responses",
213 "source_paths": ["src/app.py"],
214 },
215 "requirements": {
216 "priority": "balanced",
217 "critical_features": [],
218 **(requirements or {}),
219 },
220 "detected_features": [],
221 }
222 ],
223 }
224
225
226def _verify_openai(requirements=None, **kwargs):
227 recommendation = model_recommendation.recommend(_openai_input(requirements))
228 return verify_model_path.verify_recommendation(recommendation, now=NOW, **kwargs)
229
230
231def test_mantle_responses_probe_calls_responses_create_with_exact_id():
232 client = FakeOpenAIClient()
233 result = _verify_openai(
234 {"api_continuity": "required"},
235 openai_responses_client_factory=lambda region: client,
236 )
237 verification = result["workloads"]["openai-svc"]
238
239 assert verification["status"] == "passed"
240 assert verification["api_path"] == "mantle_openai_responses"
241 assert client.responses.calls[0]["model"] == "openai.gpt-5.6-sol"
242 assert client.responses.calls[0]["input"]
243 assert verification["response_model_id"] == "openai.gpt-5.6-sol"
244
245
246def test_mantle_responses_failure_does_not_substitute_model():
247 client = FakeOpenAIClient(FakeResponses(error=RuntimeError("access denied")))
248 result = _verify_openai(
249 {"api_continuity": "required"},
250 openai_responses_client_factory=lambda region: client,
251 )
252 verification = result["workloads"]["openai-svc"]
253
254 assert verification["status"] == "failed"
255 assert verification["invocation_model_id"] == "openai.gpt-5.6-sol"
256 # Only the exact selected model was ever probed.
257 assert [c["model"] for c in client.responses.calls] == ["openai.gpt-5.6-sol"]
258
259
260def test_openai_decision_required_is_not_probed():
261 called = {"n": 0}
262
263 def factory(region):
264 called["n"] += 1
265 return FakeOpenAIClient()
266
267 result = _verify_openai(
268 {"api_continuity": "required", "governance": ["guardrails"]},
269 openai_responses_client_factory=factory,
270 )
271 verification = result["workloads"]["openai-svc"]
272
273 assert verification["status"] == "not_applicable"
274 assert called["n"] == 0 # no client constructed for a non-selected recommendation