Setting the file. One moment.
Test Workload Classes · Agent Advisor · aws/agent-toolkit-for-aws · Skills Docs
ContentsBack to the top of the page 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 95
This file
Number 23.79
Position 79 of 81
Type Python
Size 5 KB
Lines 118 scripts/ test_workload_classes.py
Python · 118 lines · 5 KB
13
14 def _text ():
15 return WC_MD .read_text()
16
17
18 def test_file_exists ():
19 assert WC_MD .exists(), WC_MD
20
21
22 def test_first_match_wins_stated ():
23 assert "first match wins" in _text().lower()
24
25
26 def test_rules_present_in_order ():
27 text = _text()
28 positions = [text.find( f "**W { i } **" ) for i in range ( 1 , 7 )]
29 assert all (p != - 1 for p in positions), f "missing rule ids: { positions } "
30 assert positions == sorted (positions), "W1..W6 must appear in order"
31
32
33 def test_batch_never_agentcore ():
34 # AgentCore frames work as sessions with session-scoped billing (even the
35 # Instances compute type, 14d ceiling) — wrong shape for batch schedulers.
36 text = _text()
37 assert "batch" in text
38 assert re.search( r "batch . {0,400} never AgentCore" , text, re. DOTALL | re. IGNORECASE )
39
40
41 def test_light_io_scale_to_zero_default ():
42 assert re.search( r "light_io . {0,300} Lambda" , _text(), re. DOTALL )
43
44
45 def test_existing_cluster_respected ():
46 # W1: an existing EKS/ECS cluster pulls same-class workloads onto it.
47 assert "existing_cluster" in _text()
48
49
50 def test_verdict_tokens_explicit ():
51 """Each workload-classes rule now states the verdict enum token explicitly
52 (e.g., `batch` for W2, `lambda` for W3/W4, `fargate` for W5/W6) alongside
53 the prose label. This prevents the bug where design.json.verdict drifted
54 from workload_class ( BUG -7: ocr had workload_class:batch but verdict:ecs).
55 """
56 text = _text()
57 # W2 → `batch` token
58 assert re.search( r "W2 . * `batch` . * AWS Batch" , text, re. DOTALL )
59 # W3 → `lambda` token
60 assert re.search( r "W3 . * `lambda` . * Lambda" , text, re. DOTALL )
61 # W4 → `lambda` token
62 assert re.search( r "W4 . * `lambda` . * Lambda" , text, re. DOTALL )
63 # W5 → `fargate` token
64 assert re.search( r "W5 . * `fargate` . * Fargate" , text, re. DOTALL )
65 # W6 → `fargate` token
66 assert re.search( r "W6 . * `fargate` . * Fargate" , text, re. DOTALL )
67
68
69 def test_design_md_asserts_verdict_workload_class_consistency ():
70 """design.md's _assert must include the invariant that verdict and
71 workload_class are never contradictory (batch workload → batch verdict unless
72 W1 existing-cluster reuse)."""
73 design_md = WC_MD .parent.parent / "phases" / "design" / "design.md"
74 assert design_md.exists()
75 text = design_md.read_text()
76 # Check for the assertion that verdict matches workload_class rule mapping
77 assert re.search( r "verdict . * workload . * never contradictory" , text, re. DOTALL | re. IGNORECASE )
78
79
80 def test_consolidated_does_not_rewrite_per_unit_verdict ():
81 """Live-test 0715 (consolidate onto ECS): the LLM overwrote every unit.verdict to the
82 superset (ecs), violating design.md's own _assert (non-agent verdict == workload-classes
83 token) and losing the raw verdict the downstream effective-runtime logic needs. design.md
84 must state that consolidation lives only in the platform block, never in units[].verdict."""
85 design_md = WC_MD .parent.parent / "phases" / "design" / "design.md"
86 text = re.sub( r " \s + " , " " , design_md.read_text()) # normalize prose line-wraps
87 assert re.search( r " [ Cc ] onsolidated does NOT rewrite per-unit" , text) or \
88 re.search( r "never overwrites ? . {0,40} verdict . {0,40} superset" , text, re. IGNORECASE ), \
89 "design.md must state consolidated does not rewrite per-unit verdict"
90 assert "effective_runtime" in text and \
91 re.search( r "consolidated . {0,80} platform \. runtime" , text, re. IGNORECASE ), \
92 "design.md must set per-unit effective_runtime to platform.runtime when consolidated"
93
94
95 def test_every_verdict_has_a_service_card_with_serving_notes ():
96 """Codex P1 #6: design.md derives each unit's key_change from the winning runtime's
97 '## Serving & security notes' block. Every non-agent verdict a workload-classes rule
98 can emit (batch/fargate/lambda/ecs/eks) must resolve to a card that HAS that block, or
99 the derivation fabricates. fargate reuses ecs.md."""
100 refs = WC_MD .parent # decision-refs/
101 # (verdict token -> card filename); fargate aliases to ecs.md
102 verdict_cards = {
103 "batch" : "batch.md" ,
104 "lambda" : "lambda.md" ,
105 "ecs" : "ecs.md" ,
106 "eks" : "eks.md" ,
107 "fargate" : "ecs.md" , # ECS-on-Fargate shares the ecs card
108 }
109 for verdict, card in verdict_cards.items():
110 path = refs / card
111 assert path.exists(), f " { verdict } verdict has no service card ( { card } )"
112 assert "## Serving & security notes" in path.read_text(), \
113 f " { card } ( { verdict } ) lacks the '## Serving & security notes' block key_change needs"
114 # design.md must state the fargate->ecs card alias so the loader resolves it.
115 design_md = refs.parent / "phases" / "design" / "design.md"
116 dtext = design_md.read_text()
117 assert re.search( r "`fargate` . {0,40} ` ? ecs \. md` ?| ecs \. md . {0,40} fargate" , dtext, re. IGNORECASE ), \
118 "design.md must state fargate resolves to ecs.md for card loading"