From d03e977de72f0494d9a9f0cf96d334cd9a6b042b Mon Sep 17 00:00:00 2001 From: a2a-cloud Date: Mon, 13 Jul 2026 12:05:34 +0000 Subject: [PATCH] a2a-source-edit: write tests/test_customer_integration_engineer.py --- tests/test_customer_integration_engineer.py | 234 ++++++++++++++++++++ 1 file changed, 234 insertions(+) create mode 100644 tests/test_customer_integration_engineer.py diff --git a/tests/test_customer_integration_engineer.py b/tests/test_customer_integration_engineer.py new file mode 100644 index 0000000..2f1c092 --- /dev/null +++ b/tests/test_customer_integration_engineer.py @@ -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