a2a-source-edit: remove tests/test_customer_integration_engineer.py
This commit is contained in:
a2a-cloud
@@ -1,234 +0,0 @@
|
|||||||
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