CoreRasurae
b41f536da6
CI / helm (push) Successful in 47s
CI / build (push) Successful in 1m11s
CI / push-validation (push) Successful in 40s
CI / benchmark-publish (push) Has started running
CI / lint (push) Successful in 1m26s
CI / typecheck (push) Successful in 1m55s
CI / e2e_tests (push) Successful in 1m12s
CI / quality (push) Successful in 2m6s
CI / security (push) Successful in 2m9s
CI / benchmark-regression (push) Failing after 1m28s
CI / integration_tests (push) Successful in 3m48s
CI / unit_tests (push) Successful in 7m31s
CI / coverage (push) Has started running
CI / docker (push) Successful in 1m35s
CI / helm (pull_request) Successful in 39s
CI / push-validation (pull_request) Successful in 39s
CI / build (pull_request) Successful in 1m13s
CI / lint (pull_request) Successful in 1m32s
CI / quality (pull_request) Successful in 1m49s
CI / typecheck (pull_request) Successful in 1m48s
CI / security (pull_request) Successful in 1m58s
CI / status-check (push) Has been cancelled
CI / integration_tests (pull_request) Successful in 4m48s
CI / unit_tests (pull_request) Successful in 5m11s
CI / coverage (pull_request) Has started running
CI / docker (pull_request) Successful in 1m44s
CI / status-check (pull_request) Has been cancelled
Add BDD regression tests for issue #4328 automation profile gates Tests verify that complete_strategize() and complete_execute() respect automation profile thresholds and do NOT unconditionally call auto_progress(). Covers 8 built-in profiles: manual, full-auto, supervised, auto, review_before_apply, ci, trusted ISSUES CLOSED: #4328
672 lines
24 KiB
Python
672 lines
24 KiB
Python
"""Helper for Workflow Example 8: Cloud Infrastructure Management.
|
|
|
|
Integration test exercising the ``supervised`` automation profile with:
|
|
- Custom resource type (local/terraform-state) registration (Step 1)
|
|
- Custom skill (local/terraform-ops) with tool composition (Step 1)
|
|
- Supervised automation profile behavior verification
|
|
- Action creation with invariants + plan use (Step 3)
|
|
- Infrastructure analysis with mocked Terraform operations
|
|
- Invariant enforcement verification
|
|
|
|
Spec reference: docs/specification.md, Example 8 (~line 39315)
|
|
Issue: #772
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Shared DB setup
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _setup_db() -> Any:
|
|
"""Create in-memory SQLite, return session factory."""
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from cleveragents.infrastructure.database.models import Base
|
|
|
|
engine = create_engine("sqlite:///:memory:", echo=False)
|
|
Base.metadata.create_all(engine)
|
|
session = sessionmaker(bind=engine, expire_on_commit=False)()
|
|
|
|
class _NoClose:
|
|
"""Keep in-memory SQLite alive across session.close() calls."""
|
|
|
|
__slots__ = ("_s",)
|
|
|
|
def __init__(self, s: object) -> None:
|
|
object.__setattr__(self, "_s", s)
|
|
|
|
def close(self) -> None:
|
|
pass
|
|
|
|
def __getattr__(self, n: str) -> object:
|
|
return getattr(object.__getattribute__(self, "_s"), n)
|
|
|
|
wrapper = _NoClose(session)
|
|
return lambda: wrapper
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixture directory
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_FIXTURES = Path(__file__).resolve().parent / "fixtures" / "wf08"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Mock Terraform operations (deterministic, no real Terraform needed)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _mock_terraform_show(resource_address: str) -> dict[str, Any]:
|
|
"""Mock ``terraform show`` returning resource state from fixture."""
|
|
tfstate_path = _FIXTURES / "terraform.tfstate.json"
|
|
state = json.loads(tfstate_path.read_text())
|
|
for resource in state.get("resources", []):
|
|
addr = f"{resource['type']}.{resource['name']}"
|
|
if addr == resource_address:
|
|
return {
|
|
"address": addr,
|
|
"type": resource["type"],
|
|
"name": resource["name"],
|
|
"values": resource["instances"][0]["attributes"],
|
|
}
|
|
return {"error": f"Resource {resource_address} not found"}
|
|
|
|
|
|
def _mock_cloud_metrics(resource_id: str, metric: str) -> dict[str, Any]:
|
|
"""Mock ``cloud_metrics`` fetching monitoring data from fixture."""
|
|
metrics_path = _FIXTURES / "cloud-metrics.json"
|
|
data = json.loads(metrics_path.read_text())
|
|
for entry in data.get("metrics", []):
|
|
if entry["resource_id"] == resource_id and entry["metric"] == metric:
|
|
return entry
|
|
return {"error": f"No metric {metric} for {resource_id}"}
|
|
|
|
|
|
def _mock_terraform_plan(target: str | None = None) -> dict[str, Any]:
|
|
"""Mock ``terraform plan`` returning a plan summary."""
|
|
changes: list[dict[str, str]] = []
|
|
if target is None or target == "aws_instance.web":
|
|
changes.append(
|
|
{
|
|
"action": "update",
|
|
"resource": "aws_instance.web",
|
|
"detail": "instance_type: m5.2xlarge -> m5.large (right-sizing)",
|
|
}
|
|
)
|
|
return {
|
|
"plan_summary": {
|
|
"add": 0,
|
|
"change": len(changes),
|
|
"destroy": 0,
|
|
},
|
|
"changes": changes,
|
|
}
|
|
|
|
|
|
def _analyze_infrastructure() -> dict[str, Any]:
|
|
"""Run mocked infrastructure analysis producing optimization recommendations.
|
|
|
|
Simulates the LLM-driven analysis of Terraform state and cloud metrics
|
|
to produce cost-saving recommendations.
|
|
"""
|
|
# Step 1: Show current web instance state
|
|
web_state = _mock_terraform_show("aws_instance.web")
|
|
assert "error" not in web_state, f"Failed to show web instance: {web_state}"
|
|
|
|
# Step 2: Get CPU metrics for the web instance
|
|
web_cpu = _mock_cloud_metrics("i-0abc123def456", "CPUUtilization")
|
|
assert "error" not in web_cpu, f"Failed to get web CPU metrics: {web_cpu}"
|
|
|
|
# Step 3: Check API instance (should NOT be recommended for changes — critical)
|
|
api_state = _mock_terraform_show("aws_instance.api")
|
|
assert api_state["values"]["tags"]["critical"] == "true"
|
|
|
|
# Step 4: Generate plan for right-sizing
|
|
plan = _mock_terraform_plan("aws_instance.web")
|
|
|
|
# Step 5: Build optimization recommendations
|
|
recommendations: list[dict[str, Any]] = []
|
|
|
|
# Web instance is over-provisioned: avg CPU 12.3%, peak 28.7% on m5.2xlarge
|
|
if web_cpu["average"] < 30.0 and web_cpu["peak"] < 70.0:
|
|
recommendations.append(
|
|
{
|
|
"resource": "aws_instance.web",
|
|
"type": "right-sizing",
|
|
"current": web_state["values"]["instance_type"],
|
|
"recommended": "m5.large",
|
|
"reason": (
|
|
f"Average CPU {web_cpu['average']}%, "
|
|
f"peak {web_cpu['peak']}% — "
|
|
"well within m5.large capacity"
|
|
),
|
|
"estimated_monthly_savings_usd": 148.0,
|
|
}
|
|
)
|
|
|
|
return {
|
|
"recommendations": recommendations,
|
|
"plan": plan,
|
|
"analyzed_resources": 3,
|
|
"skipped_critical": ["aws_instance.api"],
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 1. Register custom resource type (Spec Step 1)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _register_terraform_resource_type() -> None:
|
|
"""Register local/terraform-state resource type via YAML fixture."""
|
|
from cleveragents.application.services.resource_registry_service import (
|
|
ResourceRegistryService,
|
|
)
|
|
|
|
factory = _setup_db()
|
|
svc = ResourceRegistryService(session_factory=factory)
|
|
|
|
yaml_path = str(_FIXTURES / "terraform-state.yaml")
|
|
spec = svc.register_type(yaml_path)
|
|
|
|
assert spec.name == "local/terraform-state", (
|
|
f"Expected name 'local/terraform-state', got '{spec.name}'"
|
|
)
|
|
assert str(spec.sandbox_strategy) == "copy_on_write", (
|
|
f"Expected sandbox_strategy 'copy_on_write', got '{spec.sandbox_strategy}'"
|
|
)
|
|
assert spec.capabilities.get("read") is True
|
|
assert spec.capabilities.get("write") is True
|
|
assert spec.capabilities.get("sandbox") is True
|
|
assert spec.capabilities.get("checkpoint") is True
|
|
|
|
# Verify cli_args (state-path required, workspace optional)
|
|
cli_args = spec.cli_args
|
|
assert len(cli_args) == 2, f"Expected 2 cli_args, got {len(cli_args)}"
|
|
state_path_arg = next(a for a in cli_args if a.name == "state-path")
|
|
assert state_path_arg.required is True
|
|
workspace_arg = next(a for a in cli_args if a.name == "workspace")
|
|
assert workspace_arg.required is False
|
|
assert workspace_arg.default == "default"
|
|
|
|
# Verify show round-trip
|
|
shown = svc.show_type("local/terraform-state")
|
|
assert shown.name == "local/terraform-state"
|
|
|
|
# Register an instance of this type
|
|
with tempfile.TemporaryDirectory(prefix="wf08-tf-") as tmpdir:
|
|
resource = svc.register_resource(
|
|
type_name="local/terraform-state",
|
|
name="local/prod-tfstate",
|
|
location=tmpdir,
|
|
description="Production Terraform state",
|
|
properties={
|
|
"state-path": f"{tmpdir}/terraform.tfstate",
|
|
"workspace": "production",
|
|
},
|
|
)
|
|
assert resource.name == "local/prod-tfstate"
|
|
assert resource.properties["workspace"] == "production"
|
|
|
|
print("register-terraform-resource-type-ok")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 2. Register skill with tool composition (Spec Step 1)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _register_terraform_skill() -> None:
|
|
"""Register local/terraform-ops skill with 3 tools and include_skills."""
|
|
from cleveragents.application.services.skill_service import SkillService
|
|
from cleveragents.domain.models.core.skill import Skill
|
|
|
|
skill_svc = SkillService()
|
|
|
|
# Build skill from config dict matching spec YAML
|
|
skill_config: dict[str, Any] = {
|
|
"name": "local/terraform-ops",
|
|
"description": "Terraform infrastructure operations",
|
|
"anonymous_tools": [
|
|
{
|
|
"description": "Run terraform plan and return the execution plan",
|
|
"source": "custom",
|
|
"input_schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"target": {
|
|
"type": "string",
|
|
"description": "Specific resource to target",
|
|
},
|
|
},
|
|
"required": [],
|
|
},
|
|
"capability": {
|
|
"writes": False,
|
|
"checkpointable": False,
|
|
},
|
|
},
|
|
{
|
|
"description": "Show current state of a Terraform resource",
|
|
"source": "custom",
|
|
"input_schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"resource_address": {
|
|
"type": "string",
|
|
"description": "Terraform resource address",
|
|
},
|
|
},
|
|
"required": ["resource_address"],
|
|
},
|
|
"capability": {
|
|
"writes": False,
|
|
"checkpointable": False,
|
|
},
|
|
},
|
|
{
|
|
"description": "Fetch cloud monitoring metrics for a resource",
|
|
"source": "custom",
|
|
"input_schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"resource_id": {"type": "string"},
|
|
"metric": {
|
|
"type": "string",
|
|
"description": "e.g., CPUUtilization",
|
|
},
|
|
"period_days": {
|
|
"type": "integer",
|
|
"default": 30,
|
|
},
|
|
},
|
|
"required": ["resource_id", "metric"],
|
|
},
|
|
"capability": {
|
|
"writes": False,
|
|
"checkpointable": False,
|
|
},
|
|
},
|
|
],
|
|
# Skill composition — includes local/file-ops per spec
|
|
"includes": ["local/file-ops"],
|
|
}
|
|
|
|
skill = Skill.from_config(skill_config)
|
|
assert skill.name == "local/terraform-ops", (
|
|
f"Expected name 'local/terraform-ops', got '{skill.name}'"
|
|
)
|
|
assert skill.description == "Terraform infrastructure operations"
|
|
|
|
# Verify 3 anonymous tools (terraform_plan, terraform_show, cloud_metrics)
|
|
assert len(skill.anonymous_tools) == 3, (
|
|
f"Expected 3 anonymous tools, got {len(skill.anonymous_tools)}"
|
|
)
|
|
|
|
# Verify all tools are read-only (writes=False)
|
|
for tool in skill.anonymous_tools:
|
|
cap = tool.capability
|
|
assert cap is not None, "Tool capability must not be None"
|
|
assert cap.writes is False, (
|
|
f"Tool '{tool.description}' should have writes=False"
|
|
)
|
|
assert cap.checkpointable is False, (
|
|
f"Tool '{tool.description}' should have checkpointable=False"
|
|
)
|
|
|
|
# Verify skill composition (includes)
|
|
assert len(skill.includes) == 1, f"Expected 1 include, got {len(skill.includes)}"
|
|
assert skill.includes[0].name == "local/file-ops"
|
|
|
|
# Verify skill_count tracks
|
|
assert skill_svc.skill_count() >= 0
|
|
|
|
print("register-terraform-skill-ok")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 3. Supervised profile behavior (Spec Step 3 automation)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _supervised_profile_behavior() -> None:
|
|
"""Verify supervised profile: gates at strategize->execute and execute->apply."""
|
|
from cleveragents.application.services.automation_profile_service import (
|
|
AutomationProfileService,
|
|
)
|
|
from cleveragents.application.services.plan_lifecycle_service import (
|
|
PlanLifecycleService,
|
|
)
|
|
from cleveragents.config.settings import Settings
|
|
from cleveragents.domain.models.core.plan import (
|
|
PlanPhase,
|
|
ProcessingState,
|
|
ProjectLink,
|
|
)
|
|
|
|
# Verify supervised profile properties
|
|
profile_svc = AutomationProfileService(repo=None)
|
|
supervised = profile_svc.resolve_profile(plan_profile="supervised")
|
|
assert supervised.name == "supervised"
|
|
assert supervised.create_tool == 1.0, (
|
|
f"Expected create_tool=1.0, got {supervised.create_tool}"
|
|
)
|
|
assert supervised.select_tool == 1.0, (
|
|
f"Expected select_tool=1.0, got {supervised.select_tool}"
|
|
)
|
|
assert supervised.safety.require_sandbox is True
|
|
assert supervised.safety.require_checkpoints is True
|
|
|
|
# Verify supervised profile gates plan transitions
|
|
service = PlanLifecycleService(settings=Settings())
|
|
service.create_action(
|
|
name="local/supervised-gate-test",
|
|
description="Test supervised profile gates",
|
|
definition_of_done="Gates verified",
|
|
strategy_actor="local/stub",
|
|
execution_actor="local/stub",
|
|
automation_profile="supervised",
|
|
)
|
|
plan = service.use_action(
|
|
action_name="local/supervised-gate-test",
|
|
project_links=[ProjectLink(project_name="test")],
|
|
)
|
|
pid = plan.identity.plan_id
|
|
|
|
# Strategize already auto-completed by try_auto_run
|
|
# (decompose_task=0.0 in supervised profile)
|
|
plan = service.get_plan(pid)
|
|
|
|
# Supervised: create_tool=1.0 means should_auto_progress is False
|
|
assert plan.phase == PlanPhase.STRATEGIZE
|
|
assert plan.processing_state == ProcessingState.COMPLETE
|
|
assert service.should_auto_progress(plan) is False, (
|
|
"Supervised profile should NOT auto-progress from strategize"
|
|
)
|
|
|
|
# Explicit execute
|
|
service.execute_plan(pid)
|
|
service.start_execute(pid)
|
|
plan = service.complete_execute(pid)
|
|
|
|
# Supervised: select_tool=1.0 means should_auto_progress is False
|
|
assert plan.phase == PlanPhase.EXECUTE
|
|
assert plan.processing_state == ProcessingState.COMPLETE
|
|
assert service.should_auto_progress(plan) is False, (
|
|
"Supervised profile should NOT auto-progress from execute"
|
|
)
|
|
|
|
# Explicit apply works
|
|
service.apply_plan(pid)
|
|
service.start_apply(pid)
|
|
plan = service.complete_apply(pid)
|
|
assert plan.processing_state == ProcessingState.APPLIED
|
|
assert plan.is_terminal
|
|
|
|
print("supervised-profile-behavior-ok")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 4. Action with supervised profile, args, invariants + plan use (Spec Step 3)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _create_infra_optimize_action() -> None:
|
|
"""Create infra-optimize action and verify plan creation with invariants."""
|
|
from cleveragents.application.services.plan_lifecycle_service import (
|
|
PlanLifecycleService,
|
|
)
|
|
from cleveragents.config.settings import Settings
|
|
from cleveragents.domain.models.core.action import (
|
|
ActionArgument,
|
|
ArgumentRequirement,
|
|
ArgumentType,
|
|
)
|
|
from cleveragents.domain.models.core.plan import (
|
|
InvariantSource,
|
|
PlanPhase,
|
|
ProcessingState,
|
|
ProjectLink,
|
|
)
|
|
|
|
service = PlanLifecycleService(settings=Settings())
|
|
action = service.create_action(
|
|
name="local/infra-optimize",
|
|
description="Analyze and optimize cloud infrastructure costs",
|
|
definition_of_done=(
|
|
"- Unused resources identified and removal plans generated\n"
|
|
"- Over-provisioned resources identified with right-sizing recs\n"
|
|
"- terraform validate passes on all changes\n"
|
|
"- terraform plan shows no unexpected resource deletions\n"
|
|
"- Estimated monthly cost savings documented"
|
|
),
|
|
strategy_actor="local/refactoring-strategist",
|
|
execution_actor="local/infra-executor",
|
|
automation_profile="supervised",
|
|
reusable=True,
|
|
arguments=[
|
|
ActionArgument(
|
|
name="optimization_targets",
|
|
arg_type=ArgumentType.STRING,
|
|
requirement=ArgumentRequirement.OPTIONAL,
|
|
description="Comma-separated: compute, storage, all",
|
|
default_value="all",
|
|
),
|
|
ActionArgument(
|
|
name="min_savings_threshold",
|
|
arg_type=ArgumentType.FLOAT,
|
|
requirement=ArgumentRequirement.OPTIONAL,
|
|
description="Min monthly savings (USD) to include",
|
|
default_value="10.0",
|
|
),
|
|
],
|
|
invariants=[
|
|
"Changes to production databases require manual approval",
|
|
(
|
|
"Instance type downgrades must verify"
|
|
" current peak CPU < 70% of new capacity"
|
|
),
|
|
],
|
|
)
|
|
|
|
assert str(action.namespaced_name) == "local/infra-optimize"
|
|
assert action.automation_profile == "supervised"
|
|
assert action.reusable is True
|
|
assert len(action.arguments) == 2
|
|
assert len(action.invariants) == 2
|
|
|
|
# Use the action to create a plan
|
|
plan = service.use_action(
|
|
action_name="local/infra-optimize",
|
|
project_links=[ProjectLink(project_name="local/infra-prod")],
|
|
arguments={
|
|
"optimization_targets": "compute,storage",
|
|
"min_savings_threshold": 25.0,
|
|
},
|
|
)
|
|
assert plan.phase == PlanPhase.STRATEGIZE
|
|
assert plan.state == ProcessingState.COMPLETE
|
|
|
|
# Verify arguments flowed to plan
|
|
assert plan.arguments["optimization_targets"] == "compute,storage"
|
|
assert plan.arguments["min_savings_threshold"] == 25.0
|
|
|
|
# Verify invariants propagated from action
|
|
action_invariants = [
|
|
inv for inv in plan.invariants if inv.source == InvariantSource.ACTION
|
|
]
|
|
assert len(action_invariants) == 2, (
|
|
f"Expected 2 action invariants on plan, got {len(action_invariants)}"
|
|
)
|
|
inv_texts = sorted(inv.text for inv in action_invariants)
|
|
expected = sorted(
|
|
[
|
|
"Changes to production databases require manual approval",
|
|
(
|
|
"Instance type downgrades must verify"
|
|
" current peak CPU < 70% of new capacity"
|
|
),
|
|
]
|
|
)
|
|
assert inv_texts == expected, f"Invariant texts mismatch: {inv_texts}"
|
|
|
|
print("create-infra-optimize-action-ok")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 5. Infrastructure analysis produces optimization recommendations
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _infra_analysis_recommendations() -> None:
|
|
"""Verify mocked infrastructure analysis produces valid recommendations."""
|
|
result = _analyze_infrastructure()
|
|
|
|
# Must have recommendations
|
|
recs = result["recommendations"]
|
|
assert len(recs) >= 1, f"Expected at least 1 recommendation, got {len(recs)}"
|
|
|
|
# First recommendation should be right-sizing the web instance
|
|
web_rec = recs[0]
|
|
assert web_rec["resource"] == "aws_instance.web"
|
|
assert web_rec["type"] == "right-sizing"
|
|
assert web_rec["current"] == "m5.2xlarge"
|
|
assert web_rec["recommended"] == "m5.large"
|
|
assert web_rec["estimated_monthly_savings_usd"] > 0
|
|
|
|
# Plan should show no destroys (safe optimization)
|
|
plan = result["plan"]
|
|
assert plan["plan_summary"]["destroy"] == 0, (
|
|
"Optimization plan should have zero destroys"
|
|
)
|
|
|
|
# Critical resources should be skipped
|
|
assert "aws_instance.api" in result["skipped_critical"]
|
|
|
|
# Total analyzed resources
|
|
assert result["analyzed_resources"] == 3
|
|
|
|
print("infra-analysis-recommendations-ok")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 6. Invariant enforcement
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _invariant_enforcement() -> None:
|
|
"""Verify invariants block critical resource modifications.
|
|
|
|
Simulates the invariant check: "Never delete or modify resources tagged
|
|
with 'critical: true'" — a project-level invariant from the spec.
|
|
"""
|
|
from cleveragents.application.services.plan_lifecycle_service import (
|
|
PlanLifecycleService,
|
|
)
|
|
from cleveragents.config.settings import Settings
|
|
from cleveragents.domain.models.core.plan import (
|
|
InvariantSource,
|
|
ProjectLink,
|
|
)
|
|
|
|
service = PlanLifecycleService(settings=Settings())
|
|
action = service.create_action(
|
|
name="local/invariant-test-infra",
|
|
description="Invariant enforcement test",
|
|
definition_of_done="Invariants block critical changes",
|
|
strategy_actor="local/stub",
|
|
execution_actor="local/stub",
|
|
automation_profile="supervised",
|
|
invariants=[
|
|
"Never delete or modify resources tagged with 'critical: true'",
|
|
"All changes must be backward-compatible with running services",
|
|
"Cost changes must be documented in the PR description",
|
|
],
|
|
)
|
|
assert len(action.invariants) == 3
|
|
|
|
plan = service.use_action(
|
|
action_name="local/invariant-test-infra",
|
|
project_links=[ProjectLink(project_name="local/infra-prod")],
|
|
)
|
|
|
|
# All invariants should flow to the plan
|
|
action_invariants = [
|
|
inv for inv in plan.invariants if inv.source == InvariantSource.ACTION
|
|
]
|
|
assert len(action_invariants) == 3, (
|
|
f"Expected 3 action invariants, got {len(action_invariants)}"
|
|
)
|
|
|
|
# Simulate invariant check: critical resource modification blocked
|
|
api_state = _mock_terraform_show("aws_instance.api")
|
|
is_critical = api_state["values"]["tags"].get("critical") == "true"
|
|
assert is_critical, "API instance must be tagged as critical"
|
|
|
|
# Verify that the invariant text matches what we'd check against
|
|
inv_texts = [inv.text for inv in action_invariants]
|
|
critical_invariant = next((t for t in inv_texts if "critical" in t.lower()), None)
|
|
assert critical_invariant is not None, (
|
|
"Must have invariant referencing critical resources"
|
|
)
|
|
|
|
# Mock enforcement: if resource is critical, block the change
|
|
change_blocked = is_critical and critical_invariant is not None
|
|
assert change_blocked, (
|
|
"Critical resource modification should be blocked by invariant"
|
|
)
|
|
|
|
# Non-critical resource should NOT be blocked
|
|
web_state = _mock_terraform_show("aws_instance.web")
|
|
web_is_critical = web_state["values"]["tags"].get("critical") == "true"
|
|
assert not web_is_critical, "Web instance should NOT be critical"
|
|
web_blocked = web_is_critical and critical_invariant is not None
|
|
assert not web_blocked, "Non-critical resource modification should NOT be blocked"
|
|
|
|
print("invariant-enforcement-ok")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Command dispatch
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_COMMANDS: dict[str, Any] = {
|
|
"register-terraform-resource-type": _register_terraform_resource_type,
|
|
"register-terraform-skill": _register_terraform_skill,
|
|
"supervised-profile-behavior": _supervised_profile_behavior,
|
|
"create-infra-optimize-action": _create_infra_optimize_action,
|
|
"infra-analysis-recommendations": _infra_analysis_recommendations,
|
|
"invariant-enforcement": _invariant_enforcement,
|
|
}
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) < 2:
|
|
print(f"Usage: {sys.argv[0]} <command>")
|
|
print(f"Commands: {', '.join(sorted(_COMMANDS))}")
|
|
sys.exit(1)
|
|
|
|
cmd = sys.argv[1]
|
|
fn = _COMMANDS.get(cmd)
|
|
if fn is None:
|
|
print(f"Unknown command: {cmd}")
|
|
sys.exit(1)
|
|
|
|
try:
|
|
fn()
|
|
except Exception as exc:
|
|
print(f"FAIL [{cmd}]: {type(exc).__name__}: {exc}")
|
|
import traceback
|
|
|
|
traceback.print_exc()
|
|
sys.exit(1)
|