feat(sdk-generator): emit SSE contract tests in the contract-tests backend (TS + Python)

packages/channel-harness/package.json

@@ -41,7 +41,7 @@
     "test:watch": "vitest"
   },
   "dependencies": {
-    "@archastro/sdk-generator": "^0.5.0",
+    "@archastro/sdk-generator": "^0.6.0",
     "ajv": "^8.12.0",
     "ajv-formats": "^3.0.1",
     "ws": "^8.18.0",

packages/sdk-generator/__tests__/backends/streaming.test.ts

@@ -2,6 +2,7 @@ import { describe, it, expect } from "vitest";
 import { parseOpenApiSpec } from "../../src/frontend/index.js";
 import { emitResourceFile } from "../../src/backends/typescript/resource-emitter.js";
 import { emitPythonResourceFile } from "../../src/backends/python/resource-emitter.js";
+import { generateContractTests } from "../../src/backends/contract-tests/index.js";
 
 // A spec with a single SSE streaming operation (`x-sdk-streaming`), modeled on
 // POST /ai/chat/completions/stream: a typed request body plus a discriminated
@@ -123,3 +124,43 @@ describe("Python SSE streaming emission", () => {
     expect(pyOutput).toContain("Union[");
   });
 });
+
+describe("SSE contract-test generation", () => {
+  it("emits a TypeScript stream contract test driving the harness", () => {
+    const files = generateContractTests(ast, { lang: "typescript", outDir: "sdk" });
+    const entry = Object.entries(files).find(
+      ([p]) => p.includes("/streams/") && p.endsWith(".contract.test.ts")
+    );
+    expect(entry).toBeDefined();
+    const content = entry![1];
+    expect(content).toContain('import { HarnessServiceClient } from "@archastro/channel-harness";');
+    expect(content).toContain("registerStreamScenario");
+    // autoEmit lets the harness synthesize contract-valid event payloads.
+    expect(content).toContain('{ type: "autoEmit", event: "message_delta" }');
+    expect(content).toMatch(/for await \(const ev of .*\.stream\(/);
+    // error path asserts a non-2xx surfaces as ApiError
+    expect(content).toContain("ApiError");
+  });
+
+  it("emits a Python stream contract test driving the harness", () => {
+    const files = generateContractTests(ast, { lang: "python", outDir: "sdk" });
+    const entry = Object.entries(files).find(
+      ([p]) => p.includes("/streams/") && p.endsWith(".py")
+    );
+    expect(entry).toBeDefined();
+    const content = entry![1];
+    expect(content).toContain("from archastro.phx_channel import HarnessServiceClient");
+    expect(content).toContain("register_stream_scenario");
+    expect(content).toContain('{"type": "autoEmit", "event": "message_delta"}');
+    expect(content).toMatch(/async for ev in .*\.stream\(/);
+    expect(content).toContain("ApiError");
+  });
+
+  it("gates the stream tests behind the opt-in env + boots the harness", () => {
+    const ts = generateContractTests(ast, { lang: "typescript", outDir: "sdk" });
+    const vitestConfig = ts["sdk/__tests__/contract/vitest.contract.config.ts"];
+    expect(vitestConfig).toContain('"__tests__/contract/streams/**/*"');
+    const globalSetup = ts["sdk/__tests__/contract/global-setup.ts"];
+    expect(globalSetup).toContain("harnessProcess");
+  });
+});

packages/sdk-generator/src/backends/contract-tests/method-chain-builder.ts

@@ -39,21 +39,49 @@ export interface MethodCallInfo {
 }
 
 /**
- * Determine whether an operation should be included in contract tests.
+ * Determine whether an operation should be included in the normal
+ * call-and-assert-response contract tests. Streaming ops are excluded here —
+ * they get dedicated SSE tests via {@link buildStreamCalls}.
  */
 function isTestableOperation(op: OperationDef): boolean {
   if (op.streaming) return false;
   return true;
 }
 
 /**
- * Walk the resource tree and yield MethodCallInfo for every testable operation.
+ * Walk the resource tree and yield MethodCallInfo for every non-streaming
+ * (testable) operation.
  */
 export function buildMethodCalls(
   spec: SdkSpec,
   versionSet: VersionedResourceSet,
   lang: "typescript" | "python",
   opts: ValueOptions = {}
+): MethodCallInfo[] {
+  return collectCalls(spec, versionSet, lang, opts, isTestableOperation);
+}
+
+/**
+ * Walk the resource tree and yield MethodCallInfo for every SSE streaming
+ * operation (`op.streaming`) — the dedicated set the stream contract tests
+ * drive against the harness. Each result's `operation.streaming.events` carries
+ * the declared SSE events.
+ */
+export function buildStreamCalls(
+  spec: SdkSpec,
+  versionSet: VersionedResourceSet,
+  lang: "typescript" | "python",
+  opts: ValueOptions = {}
+): MethodCallInfo[] {
+  return collectCalls(spec, versionSet, lang, opts, (op) => Boolean(op.streaming));
+}
+
+function collectCalls(
+  spec: SdkSpec,
+  versionSet: VersionedResourceSet,
+  lang: "typescript" | "python",
+  opts: ValueOptions,
+  include: (op: OperationDef) => boolean
 ): MethodCallInfo[] {
   const results: MethodCallInfo[] = [];
   const clientPrefix = `client.${versionSet.version}`;
@@ -71,7 +99,7 @@ export function buildMethodCalls(
         : resource.name;
 
       for (const op of resource.operations) {
-        if (!isTestableOperation(op)) continue;
+        if (!include(op)) continue;
 
         const args: MethodArg[] = [];
 

packages/sdk-generator/src/backends/contract-tests/python-emitter.ts

@@ -8,6 +8,7 @@ import {
 } from "../python/response-type.js";
 import {
   buildMethodCalls,
+  buildStreamCalls,
   groupByTopLevelResource,
   type MethodCallInfo,
 } from "./method-chain-builder.js";
@@ -55,17 +56,131 @@ export function emitPythonContractTests(
     files[filePath] = emitPythonChannelContractTestFile(channel, modulePath);
   }
 
+  // Per-resource SSE stream contract tests. Like channels, these are
+  // harness-backed (the SDK's async stream() runs against the channel-harness)
+  // and gated by the same opt-in env var.
+  let streamCallCount = 0;
+  for (const versionSet of spec.versions) {
+    const streamCalls = buildStreamCalls(spec, versionSet, "python");
+    streamCallCount += streamCalls.length;
+    const groups = groupByTopLevelResource(streamCalls);
+
+    for (const [resourceName, resourceCalls] of groups) {
+      const filePath = `${testDir}/streams/test_${snakeCase(resourceName)}.py`;
+      files[filePath] = emitPythonStreamTestFile(resourceCalls);
+    }
+  }
+
   // Conftest drives lifecycle for both backends. We only spawn Prism when
   // the spec actually has REST operations (Prism refuses to mock a spec with
-  // no paths) and only spawn the harness service when the spec has channels.
+  // no paths) and only spawn the harness service when the spec has channels
+  // or SSE streams.
   files[`${testDir}/conftest.py`] = emitConftest({
     includePrism: restCallCount > 0,
-    includeHarness: spec.channels.length > 0,
+    includeHarness: spec.channels.length > 0 || streamCallCount > 0,
   });
 
   return files;
 }
 
+// ─── SSE stream contract tests ───────────────────────────────────
+
+/**
+ * Emit a pytest file driving a resource's async stream() methods through the
+ * SDK runtime's stream_sse against the channel-harness: register an autoEmit
+ * scenario over the control API, iterate the method, assert the yielded events;
+ * and a non-2xx scenario that must surface as ApiError.
+ */
+function emitPythonStreamTestFile(calls: MethodCallInfo[]): string {
+  const cb = new CodeBuilder("    ");
+
+  for (const line of generatedHeaderPython().trim().split("\n")) cb.line(line);
+  cb.line();
+  cb.line("import pytest");
+  cb.line("from archastro.platform import AsyncPlatformClient");
+  cb.line("from archastro.platform.runtime.http_client import ApiError");
+  cb.line("from archastro.phx_channel import HarnessServiceClient");
+  cb.line();
+  cb.line("# Mark every coroutine in this file as async so the tests run whether or");
+  cb.line("# not the consuming project sets asyncio_mode=auto.");
+  cb.line("pytestmark = pytest.mark.asyncio");
+  cb.line();
+
+  for (const call of calls) emitPythonStreamTest(cb, call);
+
+  return cb.toString();
+}
+
+function emitPythonStreamTest(cb: CodeBuilder, call: MethodCallInfo): void {
+  const routeKey = `${call.httpMethod} ${call.httpPath}`;
+  const events = call.operation.streaming?.events ?? [];
+  const argStr = buildPythonArgs(call);
+  const chainPy = call.accessorChain.replace("client.", "");
+  const methodCall = `client.${chainPy}.${pythonParameterName(call.methodName)}(${argStr})`;
+  // autoEmit: the harness synthesizes contract-valid payloads from each event
+  // schema (resolving $refs), so the test stays valid without SDK-side fixtures.
+  const actions = events
+    .map((e) => `{"type": "autoEmit", "event": ${JSON.stringify(e.event)}}`)
+    .join(", ");
+  const expectedNames = `[${events.map((e) => JSON.stringify(e.event)).join(", ")}]`;
+
+  const emitClient = (): void => {
+    cb.line(
+      'harness = HarnessServiceClient(ws_url=harness_service["wsUrl"], control_url=harness_service["controlUrl"])'
+    );
+    cb.line("await harness.reset()");
+  };
+  const emitSdkClient = (): void => {
+    cb.line(
+      'client = AsyncPlatformClient(base_url=harness_service["controlUrl"], default_headers={"x-archastro-api-key": "pk_test-key"}, access_token="test-token")'
+    );
+  };
+
+  // Happy path: the SDK yields the declared events, in order.
+  cb.line(`async def ${buildTestName(call, "yields_sse_events")}(harness_service):`);
+  cb.indent();
+  emitClient();
+  cb.line(
+    `await harness.register_stream_scenario({"route": ${JSON.stringify(routeKey)}, "actions": [${actions}]})`
+  );
+  emitSdkClient();
+  cb.pyBlock("try", () => {
+    cb.line("events = []");
+    cb.pyBlock(`async for ev in ${methodCall}`, () => {
+      cb.line("events.append(ev)");
+    });
+    cb.line(`assert [e["event"] for e in events] == ${expectedNames}`);
+  });
+  cb.pyBlock("finally", () => {
+    cb.line("await client.close()");
+    cb.line("await harness.close()");
+  });
+  cb.dedent();
+  cb.line();
+
+  // Error path: a non-2xx status surfaces as ApiError.
+  cb.line(`async def ${buildTestName(call, "rejects_non_2xx")}(harness_service):`);
+  cb.indent();
+  emitClient();
+  cb.line(
+    `await harness.register_stream_scenario({"route": ${JSON.stringify(routeKey)}, "actions": [{"type": "status", "code": 402, "body": {"error": {"code": "plan_not_entitled"}}}]})`
+  );
+  emitSdkClient();
+  cb.pyBlock("try", () => {
+    cb.pyBlock("with pytest.raises(ApiError)", () => {
+      cb.pyBlock(`async for _ in ${methodCall}`, () => {
+        cb.line("pass");
+      });
+    });
+  });
+  cb.pyBlock("finally", () => {
+    cb.line("await client.close()");
+    cb.line("await harness.close()");
+  });
+  cb.dedent();
+  cb.line();
+}
+
 function emitResourceTestFile(
   _resourceName: string,
   calls: MethodCallInfo[]
@@ -420,7 +535,7 @@ function emitConftest(opts: {
       "",
       "# Skip the channel test tree entirely at collection time when the env var",
       "# is not set, so CI runs REST tests without pulling in the harness service.",
-      'collect_ignore_glob = [] if _channel_tests_enabled() else ["channels/*"]'
+      'collect_ignore_glob = [] if _channel_tests_enabled() else ["channels/*", "streams/*"]'
     );
   }