bobseares-2/customer-integration-engineer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

a2a-source-edit: write tests/test_customer_integration_engineer.py

Browse Source
This commit is contained in:
a2a-cloud
2026-07-13 11:50:06 +00:00
parent daca401b0f
commit aa98c2663f
1 changed files with 234 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

234
tests/test_customer_integration_engineer.py Normal file
View 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