a2a-source-edit: write tests/test_customer_integration_engineer.py
This commit is contained in:
a2a-cloud
234
tests/test_customer_integration_engineer.py
Normal file
234
tests/test_customer_integration_engineer.py
Normal file
@@ -0,0 +1,234 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from a2a_pack import LocalRunContext, LocalWorkspaceClient, NoAuth, WorkspaceAccess, WorkspaceMode
|
||||
|
||||
from agent import (
|
||||
APPROVAL_PHRASE,
|
||||
AuthMethod,
|
||||
ContractArtifact,
|
||||
ContractFixture,
|
||||
CustomerIntegrationEngineer,
|
||||
DesignIntegrationRequest,
|
||||
EndpointTest,
|
||||
FieldMapping,
|
||||
GenerateConnectorRequest,
|
||||
IntegrationRequirements,
|
||||
PrepareHandoffRequest,
|
||||
RunAcceptanceRequest,
|
||||
SuccessCriterion,
|
||||
SystemSpec,
|
||||
ValidateContractRequest,
|
||||
_plan_digest,
|
||||
redact_obj,
|
||||
)
|
||||
|
||||
|
||||
def workspace():
|
||||
return LocalWorkspaceClient(
|
||||
{},
|
||||
access=WorkspaceAccess.dynamic(
|
||||
max_files=256,
|
||||
allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
|
||||
require_reason=False,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def ctx(ws=None):
|
||||
return LocalRunContext(auth=NoAuth(), workspace=ws or workspace(), task_id="test")
|
||||
|
||||
|
||||
def requirements() -> IntegrationRequirements:
|
||||
return IntegrationRequirements(
|
||||
integration_id="webhook-demo",
|
||||
title="Synthetic webhook integration",
|
||||
systems=[SystemSpec(name="Source", kind="webhook", base_url="https://hooks.example.com/events")],
|
||||
auth_methods=[AuthMethod(system_name="Source", auth_type="hmac_signature", secret_ref="WEBHOOK_SECRET", header_name="X-Signature")],
|
||||
data_contracts=[
|
||||
{
|
||||
"name": "event",
|
||||
"direction": "inbound",
|
||||
"fields": [FieldMapping(source="event.id", target="id"), FieldMapping(source="event.email", target="email", pii=True)],
|
||||
"example_payload": {"event": {"id": "evt_1", "email": "a@example.com"}},
|
||||
}
|
||||
],
|
||||
rate_limits=[{"system_name": "Source", "max_requests": 60, "per_seconds": 60}],
|
||||
environments=[{"name": "sandbox", "purpose": "sandbox", "endpoint_refs": ["API_BASE_URL"], "observability_ref": "OBSERVABILITY_REF"}],
|
||||
compliance_constraints=[{"name": "PII minimization", "category": "pii", "requirement": "Mask email in logs."}],
|
||||
success_criteria=[SuccessCriterion(name="Webhook accepted", measurement="2xx response", target="100% for fixtures")],
|
||||
external_host_allowlist=["hooks.example.com"],
|
||||
dry_run=True,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_synthetic_webhook_outputs_expected_files():
|
||||
agent = CustomerIntegrationEngineer()
|
||||
ws = workspace()
|
||||
run_ctx = ctx(ws)
|
||||
req = requirements()
|
||||
|
||||
design = await agent.design_integration(run_ctx, DesignIntegrationRequest(requirements=req))
|
||||
connector = await agent.generate_connector(run_ctx, GenerateConnectorRequest(requirements=req))
|
||||
contract = await agent.validate_contract(
|
||||
run_ctx,
|
||||
ValidateContractRequest(
|
||||
integration_id=req.integration_id,
|
||||
contracts=[
|
||||
ContractArtifact(
|
||||
name="webhook",
|
||||
kind="webhook",
|
||||
schema_doc={"event_type": "customer.created"},
|
||||
signature_header="X-Signature",
|
||||
secret_ref="WEBHOOK_SECRET",
|
||||
fixtures=[ContractFixture(name="ok", payload={"event": "customer.created"})],
|
||||
)
|
||||
],
|
||||
),
|
||||
)
|
||||
acceptance = await agent.run_acceptance(
|
||||
run_ctx,
|
||||
RunAcceptanceRequest(
|
||||
integration_id=req.integration_id,
|
||||
endpoint_tests=[EndpointTest(name="dry-run webhook", url="https://hooks.example.com/events")],
|
||||
contract_fixtures=[],
|
||||
external_host_allowlist=["hooks.example.com"],
|
||||
dry_run=True,
|
||||
),
|
||||
)
|
||||
handoff = await agent.prepare_handoff(
|
||||
run_ctx,
|
||||
PrepareHandoffRequest(
|
||||
integration_id=req.integration_id,
|
||||
title=req.title,
|
||||
rollout_steps=["Enable webhook in sandbox"],
|
||||
rollback_steps=["Disable webhook"],
|
||||
monitoring_checks=["Alert on 5xx rate > 1%"],
|
||||
),
|
||||
)
|
||||
|
||||
expected = {
|
||||
"outputs/integrations/webhook-demo/architecture.md",
|
||||
"outputs/integrations/webhook-demo/mapping.json",
|
||||
"outputs/integrations/webhook-demo/connector/python/connector.py",
|
||||
"outputs/integrations/webhook-demo/connector/python/test_connector.py",
|
||||
"outputs/integrations/webhook-demo/contract-report.json",
|
||||
"outputs/integrations/webhook-demo/acceptance-report.json",
|
||||
"outputs/integrations/webhook-demo/handoff.md",
|
||||
}
|
||||
assert expected.issubset(set(ws.iter_paths()))
|
||||
assert design.status == connector.status == contract.status == acceptance.status == handoff.status == "ok"
|
||||
assert acceptance.network_calls_made == 0
|
||||
assert "outputs__integrations__webhook-demo__architecture.md" in run_ctx.artifacts
|
||||
assert "architecture.md" not in run_ctx.artifacts
|
||||
|
||||
|
||||
def test_auth_redaction_and_raw_secret_rejection():
|
||||
redacted = redact_obj({"Authorization": "Bearer super-secret-token", "nested": {"api_key": "sk-test123456789"}})
|
||||
assert "super-secret-token" not in str(redacted)
|
||||
assert "sk-test" not in str(redacted)
|
||||
with pytest.raises(ValueError):
|
||||
AuthMethod(system_name="API", auth_type="bearer_token", secret_ref="sk-live-raw-secret")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_schema_drift_fails_json_schema_fixture():
|
||||
agent = CustomerIntegrationEngineer()
|
||||
result = await agent.validate_contract(
|
||||
ctx(),
|
||||
ValidateContractRequest(
|
||||
integration_id="schema-demo",
|
||||
contracts=[
|
||||
ContractArtifact(
|
||||
name="customer",
|
||||
kind="json_schema",
|
||||
schema_doc={"type": "object", "required": ["id"], "properties": {"id": {"type": "string"}}, "additionalProperties": False},
|
||||
fixtures=[ContractFixture(name="missing id", payload={"email": "a@example.com"}, should_pass=True)],
|
||||
)
|
||||
],
|
||||
),
|
||||
)
|
||||
assert result.status == "failed"
|
||||
assert result.drift_detected is True
|
||||
assert result.checks[0].status == "fail"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ssrf_blocks_private_and_unallowlisted_hosts():
|
||||
agent = CustomerIntegrationEngineer()
|
||||
result = await agent.run_acceptance(
|
||||
ctx(),
|
||||
RunAcceptanceRequest(
|
||||
integration_id="ssrf-demo",
|
||||
endpoint_tests=[
|
||||
EndpointTest(name="private", url="https://127.0.0.1/hook"),
|
||||
EndpointTest(name="unlisted", url="https://evil.example/hook"),
|
||||
],
|
||||
external_host_allowlist=["api.example.com"],
|
||||
dry_run=True,
|
||||
),
|
||||
)
|
||||
assert result.status == "blocked"
|
||||
assert all(step.status == "blocked" for step in result.step_results)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_approval_bound_to_plan_digest():
|
||||
agent = CustomerIntegrationEngineer()
|
||||
request = PrepareHandoffRequest(
|
||||
integration_id="approval-demo",
|
||||
title="Approval demo",
|
||||
rollout_steps=["create release branch"],
|
||||
rollback_steps=["revert release branch"],
|
||||
monitoring_checks=["check error rate"],
|
||||
delivery_actions=["open pull request"],
|
||||
approval_phrase=APPROVAL_PHRASE,
|
||||
approve_delivery=True,
|
||||
plan_digest="wrong",
|
||||
)
|
||||
blocked = await agent.prepare_handoff(ctx(), request)
|
||||
assert blocked.status == "blocked"
|
||||
digest = _plan_digest({
|
||||
"integration_id": request.integration_id,
|
||||
"rollout_steps": request.rollout_steps,
|
||||
"rollback_steps": request.rollback_steps,
|
||||
"monitoring_checks": request.monitoring_checks,
|
||||
"delivery_actions": request.delivery_actions,
|
||||
})
|
||||
approved = await agent.prepare_handoff(ctx(), request.model_copy(update={"plan_digest": digest}))
|
||||
assert approved.status == "ok"
|
||||
assert approved.delivery_executed is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_idempotency_stable_in_acceptance_dry_run():
|
||||
agent = CustomerIntegrationEngineer()
|
||||
request = RunAcceptanceRequest(
|
||||
integration_id="idempotency-demo",
|
||||
endpoint_tests=[EndpointTest(name="hook", url="https://hooks.example.com/events", request_json={"x": 1})],
|
||||
external_host_allowlist=["hooks.example.com"],
|
||||
dry_run=True,
|
||||
)
|
||||
first = await agent.run_acceptance(ctx(), request)
|
||||
second = await agent.run_acceptance(ctx(), request)
|
||||
assert first.step_results[0].idempotency_key == second.step_results[0].idempotency_key
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_retry_metadata_without_network_by_default():
|
||||
agent = CustomerIntegrationEngineer()
|
||||
result = await agent.run_acceptance(
|
||||
ctx(),
|
||||
RunAcceptanceRequest(
|
||||
integration_id="retry-demo",
|
||||
endpoint_tests=[EndpointTest(name="safe", url="https://hooks.example.com/events")],
|
||||
external_host_allowlist=["hooks.example.com"],
|
||||
max_retries=3,
|
||||
allow_network=False,
|
||||
dry_run=True,
|
||||
),
|
||||
)
|
||||
assert result.step_results[0].status == "skipped"
|
||||
assert result.step_results[0].attempts == 0
|
||||
assert result.network_calls_made == 0
|
||||
Reference in New Issue
Block a user