Integration Flow Testing

Integration & Flow Testing

Overview

Generate the two test layers that close the "did the pieces actually connect" gap:

1. Integration tests — one file per integration_points[] entry in the contract map, asserting that the component correctly handles the declared protocol + endpoint + failure mode.
2. Flow tests — one file per flows[] entry, asserting end-to-end behaviour. Flows are consumer-driven only — the skill reads the declared flows: block and does NOT auto-traverse the call graph. Any auto-traversal would violate M5 (flow explosion) and makes the test matrix unpredictable.

The skill GENERATES test source files. Bob's trusted runner EXECUTES them (spec section 11.2, CB3 fix). The skill NEVER writes to the ledger directly and NEVER runs tests itself.

When Bob Invokes This Skill (with a claim_uuid bob already issued)

Bob's Step 4.5 (spec section 14.2) calls this skill with:

• component_id — the component to generate integration + flow tests for
• claim_uuid — opaque claim token bob issued via claims.issue_claim(wp_id, "integration-flow-testing")
• contract_map_path — progress/contract-map.yaml (read-only)
• project_root — project root
• language_target — pytest or jest (v1 supports ONLY these two)

Bob has already:

1. Run gates.py G1/G2/G3 — all passed
2. Confirmed the component is at stage UNIT_TESTED in the ledger
3. Issued the claim
4. Generated fixtures via sample-data-scaffolding

The skill's job is narrow: generate test files, heartbeat, emit a transition request.

Pre-Flight: Verify Bob Issued a Claim Token

import yaml
from pathlib import Path

claim = yaml.safe_load(Path(f".ledger/claims/{claim_uuid}.claim.yaml").read_text())
if claim.get("issued_by") != "bob" or claim.get("skill") != "integration-flow-testing":
    raise SystemExit("CLAIM_OWNERSHIP_VIOLATION")

Step 1: Receive Inputs from Bob

Reference CLI:

python3 -m integration_flow_testing generate \
    --component auth-service \
    --claim <UUID> \
    --contract-map progress/contract-map.yaml \
    --language pytest \
    --project-root .

Step 2: Start Heartbeat Loop (every 60s)

Same pattern as sample-data-scaffolding. Background thread calls claims.heartbeat_claim(claim_uuid) every 60s. Any non-ok state → STOP work immediately, do not emit a transition request.

Step 3: Load Contract Map, Locate Component + Flows

map_yaml = yaml.safe_load(Path("progress/contract-map.yaml").read_text())
component = next(c for c in map_yaml["components"] if c["id"] == component_id)
component_flows = [f for f in (map_yaml.get("flows") or []) if component_id in f["path"]]

Step 4: Generate Per-Component Integration Tests from integration_points[]

For each integration_points[] entry in the component:

integration_points:
  - with: user-service
    direction: outbound
    protocol: http
    endpoint: "GET /users/{id}"
    failure_mode: "404 when user unknown"

Generate ONE test file per integration point. For pytest target:

Filename convention (S030-quickwins #33): test files are named test_int_<component>__<target>.py to avoid sibling collisions when sibling component dirs contain hyphens (and therefore are not Python packages, so pytest's collector flattens them). Pre-#33 the path was test_int_<target>.py and bob had to manually rename. The <component>__ prefix makes it native.

# tests/integration/auth-service/test_int_auth_service__user_service.py
import json
import pytest
from pathlib import Path

# Auto-generated by integration-flow-testing@1.0.0
# Contract-map revision: {revision}
# Component: auth-service  Integration point: auth-service -> user-service

FIXTURE_DIR = Path(__file__).parent.parent.parent / "fixtures" / "auth-service"

def load_fixture(path):
    return json.loads((FIXTURE_DIR / path).read_text())

def test_auth_service_user_service_happy_path():
    """Happy: GET /users/{id} returns 200 with user record."""
    user_id = load_fixture("user_id/happy/0.json")
    # Arrange: fixtures loaded
    # Act: invoke auth-service entry point that calls user-service
    # Assert: user record returned, auth-service accepted it
    # (Implementation placeholder — real test body must be written per component.)
    assert user_id is not None

def test_auth_service_user_service_unhappy_404():
    """Failure mode: 404 when user unknown."""
    bad_id = load_fixture("user_id/adversarial/0.json")
    # Assert auth-service handles the 404 per its success_criteria
    assert bad_id is not None

Key properties:

• Test file name: test_int_<component>__<target>.py (S030-quickwins #33; was test_int_<target>.py pre-#33). Component prefix prevents collisions in the pytest collector across sibling component dirs.
• Happy path + unhappy path per failure_mode
• Fixture loading paths match the scaffolding skill's output
• Header comment records contract-map revision for traceability

For jest target: equivalent structure in tests/integration/<component>/test_int_<target>.test.js using describe/it and expect.

Step 5: Generate Flow Tests from flows[] (Declared Only — NEVER Auto-Traverse)

For each flow that contains this component in path[], check if this component is the flow's entry_input.component. If YES, the skill owns generating that flow's test file (each flow has one owner). If NO, skip — another invocation will handle that flow.

# tests/flow/test_flow_login_success.py
import json
import pytest
from pathlib import Path

# Auto-generated by integration-flow-testing@1.0.0
# Contract-map revision: {revision}
# Flow: FLOW-001 — login-success
# Path: auth-service -> user-service -> audit-log
# Priority: critical

def test_flow_login_success():
    """FLOW-001: session token valid -> user looked up -> audit logged -> return identity."""
    entry_fixture = Path(__file__).parent.parent / "fixtures/auth-service/session_token/happy/0.json"
    session_token = json.loads(entry_fixture.read_text())
    # 1. Invoke entry component (auth-service) with entry_input
    # 2. Assert terminal_output matches expected_outcome
    assert session_token is not None

NEVER auto-traverse. The only flows that get tests are the ones declared in flows:. If the user wants a new flow, they declare it explicitly in the contract map — which triggers a freeze-the-world revision per spec section 12.

Step 6: DO NOT Execute Tests — Bob's Trusted Runner Handles Execution

The skill's output is test source files + the transition request. Period. Bob will:

1. Apply the INTEGRATED transition (based on the skill's request)
2. Invoke run_trusted_test_suite(component_id, test_paths) itself
3. Produce a sanitized audit bundle with produced_by: bob-trusted-runner
4. Invoke audit_spawn.py with the bundle
5. On audit pass: apply INTEGRATED → VERIFIED

The skill plays NO role in execution or audit. Attempting to run tests from the skill violates CB3 and the audit will refuse the bundle.

Step 7: Emit Transition Request to Bob

The script natively emits the transition request when called with --emit-request --claim-uuid <UUID> --wp-id <WP-NN> (S030-quickwins #48). Without those flags the script preserves v1.0/v1.1 byte-identity and the caller emits the request manually. The schema below shows the file the script writes when --emit-request is set.

.ledger/requests/<request_id>.request.yaml:

request_id: <UUID>
claim_uuid: <UUID>
wp: WP-NNN
component_id: auth-service
requester: integration-flow-testing
target_stage: INTEGRATED
evidence:
  - type: integration_test_files
    produced_by: skill:integration-flow-testing
    paths:
      - tests/integration/auth-service/test_int_user_service.py
      - tests/integration/auth-service/test_int_audit_log.py
    hash: sha256:<hash-of-sorted-content>
  - type: flow_test_files
    produced_by: skill:integration-flow-testing
    paths:
      - tests/flow/test_flow_login_success.py
    hash: sha256:<hash>
language_target: pytest
at: <iso8601>

Bob applies the transition, then runs its trusted runner against the new test files, then invokes audit_spawn for the INTEGRATED → VERIFIED decision.

Test Codegen Templates (pytest, jest — v1 only)

v1 ships exactly two languages. No Go, Ruby, PHP, Rust, Java, C#. Per spec section 19 item 5, additional languages are deferred to future skill revisions with explicit WPs.

Templates live next to this SKILL.md (future expansion) or inline in the generator module. Bob's --language flag selects the template. Unknown language = HALT with UNSUPPORTED_LANGUAGE_TARGET.

Heartbeat Discipline

Same as sample-data-scaffolding: every 60s, stop immediately on non-ok state.

Timeout Handling

Per-WP budget (S/M/L). Bob revokes the claim on timeout; next heartbeat sees expired; skill exits.

CRITICAL REMINDERS — RE-READ BEFORE EVERY ACTION

• NEVER auto-traverse the call graph. Use ONLY the declared flows: block (M5 fix).
• NEVER execute tests. Generate files and emit a request. Bob's trusted runner runs them.
• NEVER mark a component VERIFIED. Only bob does that after the audit passes.
• NEVER accept an audit verdict of "looks good" — the audit is run by bob via audit_spawn.py, not by this skill.
• NEVER write to progress/integration-ledger.md directly.
• NEVER write to .ledger/claims/ — claims are bob-only.
• NEVER generate tests for components without test_scenarios — the map failed V12 if this happens.
• NEVER generate tests for languages outside the v1 set (pytest, jest).
• NEVER mock integration points without a corresponding real test — mocks are allowed IN ADDITION to a real test, not as a replacement.
• NEVER produce an audit bundle yourself. The audit bundle comes ONLY from bob's trusted runner (CB3).
• FORBIDDEN: generating tests before a claim is issued.
• FORBIDDEN: generating tests for a component not in the frozen map.
• FORBIDDEN: assuming a test layer that contradicts the contract map (e.g., inventing integration points).

Post-VERIFIED failure handling

If a flow test is later discovered to fail (via CI or subsequent runs after VERIFIED), the failure triggers a backward transition request VERIFIED → INTEGRATED with trigger: post-hoc-failure. This request comes from bob's CI integration, not from this skill. The skill never moves a component backward.

v1 scope: sync request/response, consumer-driven flows, pytest + jest only. Report scope violations to bob; do not extend the skill to solve them.