Producer Testing Guide
This guide covers how to use CVT for producer-side contract testing. Producer testing ensures your API implementation matches your OpenAPI specification before deployment.
The examples in this guide use the Petstore schema from sdks/shared/openapi.json. Copy it to your project as openapi.json or adjust the paths to reference it directly.
All registerSchema methods accept both file paths and URLs:
// From file
await validator.registerSchema("petstore", "./openapi.json");
// From URL
await validator.registerSchema(
"petstore",
"https://petstore3.swagger.io/api/v3/openapi.json",
);
Overview
Producer testing answers the question: "Does my API match my spec?"
As the producer, you own the OpenAPI spec — it is the source of truth for your API's contract. Your CI/CD pipeline should register the schema with the shared CVT server so that consumers can validate against it and register their dependencies. This enables deployment safety via can-i-deploy. See Schema Registration in CI/CD below.
| Approach | Who Uses It | What It Tests |
|---|---|---|
| Consumer Testing | API consumers | "Can I call this API correctly?" |
| Producer Testing | API producers | "Does my API match my spec?" |
┌─────────────────────┐ HTTP ┌─────────────────────┐
│ Client Requests │ ────────────► │ Your API Server │
└─────────────────────┘ │ + CVT Middleware │
└─────────────────────┘
│
│ Validate
▼
┌─────────────────────┐
│ CVT Server │
│ + Your Schema │
└─────────────────────┘
Capabilities
| Capability | Server Required? | What It Answers |
|---|---|---|
| Schema compliance tests | Yes | "Does my handler return spec-compliant responses?" |
| Breaking change detection | No (CLI) | "What changed between v1 and v2 of my spec?" |
| Consumer registry | Yes | "Which services depend on my API?" |
| can-i-deploy | Yes | "Will this change break real consumers?" |
Validation Approaches
| Test Type | Services Required | Speed | Purpose |
|---|---|---|---|
| Schema Compliance | CVT only | Fast | Unit test handlers against schema |
| Middleware Modes | CVT only | Fast | Test Strict/Warn/Shadow behavior |
| Consumer Registry | CVT only | Fast | can-i-deploy checks |
| HTTP Integration | Producer + CVT | Medium | Full end-to-end validation |
- Schema Compliance: Test handlers directly without running a server. Fast feedback during development.
- Middleware Modes: CVT middleware validates requests/responses in real-time. Three modes for different stages.
- Consumer Registry: Track which endpoints consumers use. Enables
can-i-deploysafety checks. - HTTP Integration: Real HTTP calls to running API. Tests complete stack including routing/serialization.
Schema Compliance Testing
Schema compliance testing validates that your API handlers return responses matching your OpenAPI specification.
How It Works
- Register your OpenAPI schema with CVT server
- Call your handler with test data
- Validate the response against the schema
- Get detailed error messages for any mismatches
Basic Example
- Node.js
- Python
- Go
- Java
import { ProducerTestKit } from "@sahina/cvt-sdk/producer";
describe("Petstore API", () => {
let testKit: ProducerTestKit;
beforeAll(async () => {
testKit = new ProducerTestKit({
schemaId: "petstore",
serverAddress: "localhost:9550",
});
});
afterAll(async () => {
testKit.close();
});
it("GET /pet/{petId} returns valid response", async () => {
// Call your actual handler
const response = await petHandler.getPet("123");
// Validate against schema
const result = await testKit.validateResponse({
method: "GET",
path: "/pet/123",
response: {
statusCode: 200,
body: response,
},
});
expect(result.valid).toBe(true);
expect(result.errors).toHaveLength(0);
});
it("detects missing required fields", async () => {
const result = await testKit.validateResponse({
method: "GET",
path: "/pet/123",
response: {
statusCode: 200,
body: { id: 123 }, // missing 'name' and 'photoUrls' fields
},
});
expect(result.valid).toBe(false);
expect(result.errors[0]).toContain("name");
});
});
import pytest
from cvt_sdk.producer import ProducerTestKit, ProducerTestConfig
@pytest.fixture
def test_kit():
kit = ProducerTestKit(ProducerTestConfig(
schema_id="petstore",
server_address="localhost:9550",
))
yield kit
kit.close()
def test_get_pet_returns_valid_response(test_kit, pet_handler):
# Call your handler
response = pet_handler.get_pet("123")
# Validate
result = test_kit.validate_response(
method="GET",
path="/pet/123",
status_code=200,
body=response,
)
assert result.valid
assert len(result.errors) == 0
def test_detects_missing_required_fields(test_kit):
result = test_kit.validate_response(
method="GET",
path="/pet/123",
status_code=200,
body={"id": 123}, # missing 'name' and 'photoUrls' fields
)
assert not result.valid
assert "name" in result.errors[0]
func TestPetHandler(t *testing.T) {
testKit, err := producer.NewProducerTestKit(producer.TestConfig{
SchemaID: "petstore",
ServerAddress: "localhost:9550",
})
require.NoError(t, err)
defer func() { _ = testKit.Close() }()
t.Run("GET /pet/{petId} returns valid response", func(t *testing.T) {
// Call your handler
resp := petHandler.GetPet(ctx, "123")
// Validate
result, err := testKit.ValidateResponse(ctx, producer.ValidateResponseParams{
Method: "GET",
Path: "/pet/123",
Response: producer.TestResponseData{
StatusCode: 200,
Body: resp,
},
})
require.NoError(t, err)
assert.True(t, result.Valid)
})
t.Run("detects missing required fields", func(t *testing.T) {
result, err := testKit.ValidateResponse(ctx, producer.ValidateResponseParams{
Method: "GET",
Path: "/pet/123",
Response: producer.TestResponseData{
StatusCode: 200,
Body: map[string]any{"id": 123}, // missing 'name' and 'photoUrls' fields
},
})
require.NoError(t, err)
assert.False(t, result.Valid)
})
}
public class PetHandlerTest {
private ProducerTestKit testKit;
@BeforeEach
void setup() {
testKit = ProducerTestKit.builder()
.schemaId("petstore")
.serverAddress("localhost:9550")
.build();
}
@AfterEach
void teardown() {
testKit.close();
}
@Test
void getPetReturnsValidResponse() {
// Call your handler
var response = petHandler.getPet("123");
// Validate
var result = testKit.validateResponse(
"GET",
"/pet/123",
TestResponseData.builder()
.statusCode(200)
.body(response)
.build()
);
assertTrue(result.isValid());
assertTrue(result.getErrors().isEmpty());
}
@Test
void detectsMissingRequiredFields() {
var result = testKit.validateResponse(
"GET",
"/pet/123",
TestResponseData.builder()
.statusCode(200)
.body(Map.of("id", 123)) // missing 'name' and 'photoUrls' fields
.build()
);
assertFalse(result.isValid());
assertTrue(result.getErrors().get(0).contains("name"));
}
}
Producer Middleware
For runtime validation, add CVT middleware to your HTTP server.
How Middleware Works
Framework Examples
- Express
- Fastify
- Go net/http
- Gin
- FastAPI
- Flask
- Spring
import { createExpressMiddleware } from "@sahina/cvt-sdk/producer";
app.use(
createExpressMiddleware({
schemaId: "petstore",
validator,
mode: "strict", // or 'warn' or 'shadow'
}),
);
import { fastifyProducerPlugin } from "@sahina/cvt-sdk/producer";
fastify.register(fastifyProducerPlugin, { schemaId: "petstore", validator });
import "github.com/sahina/cvt/sdks/go/cvt/producer/adapters"
config := producer.Config{
SchemaID: "petstore",
Validator: validator,
Mode: producer.ModeStrict,
}
http.Handle("/", adapters.NetHTTPMiddleware(config)(myHandler))
router := gin.Default()
router.Use(adapters.GinMiddleware(config))
from cvt_sdk.producer import ProducerConfig, ValidationMode
from cvt_sdk.producer.adapters import ASGIMiddleware
config = ProducerConfig(
schema_id="petstore",
validator=validator,
mode=ValidationMode.STRICT,
)
app.add_middleware(ASGIMiddleware, config=config)
from cvt_sdk.producer.adapters import WSGIMiddleware
app.wsgi_app = WSGIMiddleware(app.wsgi_app, config=config)
registry.addInterceptor(new SpringInterceptor(config))
.addPathPatterns("/api/**");
Path Filtering
Exclude health checks, metrics, or other paths from validation:
- Node.js
- Python
- Go
- Java
createExpressMiddleware({
schemaId: "petstore",
validator,
mode: "strict",
excludePaths: ["/health", "/metrics", "/ready"],
includePaths: ["/api/**"],
});
config = ProducerConfig(
schema_id="petstore",
validator=validator,
mode=ValidationMode.STRICT,
exclude_paths=["/health", "/metrics", "/ready"],
include_paths=["/api/**"],
)
config := producer.Config{
SchemaID: "petstore",
Validator: validator,
Mode: producer.ModeStrict,
ExcludePaths: []string{"/health", "/metrics", "/ready"},
IncludePaths: []string{"/api/**"},
}
ProducerConfig config = ProducerConfig.builder()
.schemaId("petstore")
.validator(validator)
.mode(ValidationMode.STRICT)
.excludePaths(List.of("/health", "/metrics", "/ready"))
.includePaths(List.of("/api/**"))
.build();
Validation Modes
See Validation Modes for detailed information.
| Mode | Request Violation | Response Violation | Use Case |
|---|---|---|---|
| strict | Reject with 400 | Log error | Production enforcement |
| warn | Log, continue | Log, continue | Gradual rollout |
| shadow | Silent | Silent | Initial deployment |
Recommended Rollout
Deploy with SHADOW → Analyze metrics → Switch to WARN → Fix issues → Switch to STRICT
Consumer Registry
Track which services depend on your API.
Listing Your Consumers
- Node.js
- Python
- Go
- Java
const consumers = await validator.listConsumers("petstore", "prod");
console.log(`${consumers.length} services depend on petstore in prod`);
for (const consumer of consumers) {
console.log(`- ${consumer.consumerId} v${consumer.consumerVersion}`);
}
consumers = validator.list_consumers("petstore", "prod")
print(f"{len(consumers)} services depend on petstore in prod")
for consumer in consumers:
print(f"- {consumer['consumer_id']} v{consumer['consumer_version']}")
consumers, err := validator.ListConsumers(ctx, "petstore", "prod")
if err != nil {
log.Fatal(err)
}
fmt.Printf("%d services depend on petstore in prod\n", len(consumers))
for _, consumer := range consumers {
fmt.Printf("- %s v%s\n", consumer.ConsumerID, consumer.ConsumerVersion)
}
List<ConsumerInfo> consumers = validator.listConsumers("petstore", "prod");
System.out.printf("%d services depend on petstore in prod%n", consumers.size());
for (ConsumerInfo consumer : consumers) {
System.out.printf("- %s v%s%n", consumer.getConsumerId(), consumer.getConsumerVersion());
}
Understanding Consumer Registrations
Consumers register after their contract tests pass:
// A consumer (not you) registers like this:
await validator.registerConsumer({
consumerId: "order-service",
consumerVersion: "2.1.0",
schemaId: "petstore", // Your API
schemaVersion: "1.0.0",
environment: "prod",
usedEndpoints: [
{
method: "GET",
path: "/pet/{petId}",
usedFields: ["id", "name", "status"],
},
],
});
This tells you:
order-servicedepends on your API- They use
GET /pet/{petId} - They specifically need the
id,name, andstatusfields
Deployment Safety (can-i-deploy)
Before deploying a new schema version, use can-i-deploy to check if it will break any registered consumers. For complete CLI usage, SDK examples, breaking change types, and output formats, see the Breaking Changes Guide.
Test Fixture Generation
Generate test data from your OpenAPI schema for documentation and testing.
- Node.js
- Python
- Go
- Java
// Generate complete request/response fixture
const fixture = await validator.generateFixture("GET", "/pet/{petId}");
console.log(fixture.request); // { method, path, headers, body }
console.log(fixture.response); // { statusCode, headers, body }
// Generate response only with schema examples
const response = await validator.generateResponse("GET", "/pet/{petId}", {
statusCode: 200,
useExamples: true,
});
from cvt_sdk import GenerateOptions
# Generate complete request/response fixture
fixture = validator.generate_fixture("GET", "/pet/{petId}")
print(fixture["request"]) # { method, path, headers, body }
print(fixture["response"]) # { status_code, headers, body }
# Generate response only with schema examples
response = validator.generate_response("GET", "/pet/{petId}",
GenerateOptions(status_code=200, use_examples=True),
)
// Generate complete request/response fixture
fixture, err := validator.GenerateFixture(ctx, "GET", "/pet/{petId}", nil)
if err != nil {
log.Fatal(err)
}
fmt.Println(fixture.Request) // { Method, Path, Headers, Body }
fmt.Println(fixture.Response) // { StatusCode, Headers, Body }
// Generate response only with schema examples
response, err := validator.GenerateResponse(ctx, "GET", "/pet/{petId}",
&cvt.GenerateOptions{
StatusCode: 200,
UseExamples: true,
},
)
// Generate complete request/response fixture
Fixture fixture = validator.generateFixture("GET", "/pet/{petId}");
System.out.println(fixture.getRequest()); // { method, path, headers, body }
System.out.println(fixture.getResponse()); // { statusCode, headers, body }
// Generate response only with schema examples
ResponseFixture response = validator.generateResponse("GET", "/pet/{petId}",
GenerateResponseOptions.builder()
.statusCode(200)
.useExamples(true)
.build()
);
CLI Usage
# List all endpoints in a schema
cvt generate --schema ./openapi.json --list
# Generate fixture for an endpoint
cvt generate --schema ./openapi.json --method GET --path /pet/{petId}
# Generate request only
cvt generate --schema ./openapi.json --method POST --path /pet --output-type request
# Use examples from schema
cvt generate --schema ./openapi.json --method GET --path /pet/{petId} --use-examples
Schema Registration in CI/CD
As the API producer, your CI/CD pipeline is responsible for registering your OpenAPI schema with the shared CVT server. This is how consumers discover your contract and how can-i-deploy knows what to check against.
Producer CI/CD Pipeline:
Build & Test → Register Schema → Deploy
│
▼
CVT Server (shared)
▲
│
Consumer CI/CD: Validate interactions → Register as consumer
Register on merge to main
- CLI
- Node.js
- Go
# In your producer CI/CD pipeline
cvt register-schema my-api ./openapi.json \
--version "$API_VERSION" \
--server "$CVT_SERVER_URL" \
--check-compatibility \
--fail-on-breaking
// In a CI/CD script or test setup
const validator = new ContractValidator(process.env.CVT_SERVER_URL);
await validator.registerSchemaWithVersion(
"my-api",
"./openapi.json",
process.env.API_VERSION,
);
validator, _ := cvt.NewValidator(os.Getenv("CVT_SERVER_URL"))
err := validator.RegisterSchemaWithVersion(ctx, "my-api", "./openapi.json", os.Getenv("API_VERSION"))
Why the producer must register
| If the producer registers... | If the consumer registers... |
|---|---|
| Schema always matches the real API | Schema may be stale or incorrect |
| Single source of truth | Multiple consumers may register different versions |
can-i-deploy checks are meaningful | Breaking change detection is unreliable |
| Clear ownership and accountability | Unclear who maintains the contract |
CI/CD Integration
For complete CI/CD pipeline examples (GitHub Actions, GitLab CI, Jenkins) including producer deployment safety checks and contract testing, see the CI/CD Integration Guide.
Best Practices
1. Test All Response Scenarios
Don't just test the happy path:
- Node.js
- Python
- Go
- Java
it("validates 404 response", async () => {
const result = await testKit.validateResponse({
method: "GET",
path: "/pet/nonexistent",
response: {
statusCode: 404,
body: {},
},
});
expect(result.valid).toBe(true);
});
it("validates 400 response for bad request", async () => {
const result = await testKit.validateResponse({
method: "POST",
path: "/pet",
response: {
statusCode: 400,
body: {},
},
});
expect(result.valid).toBe(true);
});
def test_validates_404_response(test_kit):
result = test_kit.validate_response(
method="GET",
path="/pet/nonexistent",
status_code=404,
body={},
)
assert result.valid
def test_validates_400_response_for_bad_request(test_kit):
result = test_kit.validate_response(
method="POST",
path="/pet",
status_code=400,
body={},
)
assert result.valid
func TestValidates404Response(t *testing.T) {
result, err := testKit.ValidateResponse(ctx, producer.ValidateResponseParams{
Method: "GET",
Path: "/pet/nonexistent",
Response: producer.TestResponseData{
StatusCode: 404,
Body: map[string]any{},
},
})
require.NoError(t, err)
assert.True(t, result.Valid)
}
func TestValidates400ResponseForBadRequest(t *testing.T) {
result, err := testKit.ValidateResponse(ctx, producer.ValidateResponseParams{
Method: "POST",
Path: "/pet",
Response: producer.TestResponseData{
StatusCode: 400,
Body: map[string]any{},
},
})
require.NoError(t, err)
assert.True(t, result.Valid)
}
@Test
void validates404Response() {
var result = testKit.validateResponse(
"GET",
"/pet/nonexistent",
TestResponseData.builder()
.statusCode(404)
.body(Map.of())
.build()
);
assertTrue(result.isValid());
}
@Test
void validates400ResponseForBadRequest() {
var result = testKit.validateResponse(
"POST",
"/pet",
TestResponseData.builder()
.statusCode(400)
.body(Map.of())
.build()
);
assertTrue(result.isValid());
}
2. Run can-i-deploy in CI
Make deployment safety checks a required gate:
cvt can-i-deploy --schema petstore --version $NEW_VERSION --env prod || exit 1
3. Use Environment-Specific Checks
Check each environment before promoting:
# Check staging first
cvt can-i-deploy --schema petstore --version 2.0.0 --env staging
# Then production
cvt can-i-deploy --schema petstore --version 2.0.0 --env prod
4. Gradual Middleware Rollout
Start with shadow mode, progress to strict:
// Week 1: Shadow mode - metrics only
mode: "shadow";
// Week 2: Warn mode - log violations
mode: "warn";
// Week 3: Strict mode - full enforcement
mode: "strict";
Troubleshooting
"Schema not found" Error
Ensure the schema is registered before running producer tests:
- Node.js
- Python
- Go
- Java
await validator.registerSchema("petstore", "./openapi.json");
validator.register_schema("petstore", "./openapi.json")
err := validator.RegisterSchema(ctx, "petstore", "./openapi.json")
validator.registerSchema("petstore", "./openapi.json");
"Path not found" Error
Check that the path in your test matches the OpenAPI spec:
- Node.js
- Python
- Go
- Java
// If spec has: /pet/{petId}
// Use actual path values:
path: '/pet/123', // NOT '/pet/{petId}'
# If spec has: /pet/{petId}
# Use actual path values:
path="/pet/123" # NOT "/pet/{petId}"
// If spec has: /pet/{petId}
// Use actual path values:
Path: "/pet/123", // NOT "/pet/{petId}"
// If spec has: /pet/{petId}
// Use actual path values:
"/pet/123" // NOT "/pet/{petId}"
"No consumers registered" Warning
This is normal if you're the first to deploy or if no consumers have registered:
SAFE TO DEPLOY
No consumers registered for this schema in prod.
Next Steps
- Consumer Testing Guide - Test your API integrations
- CI/CD Integration Guide - Pipeline examples for deployment safety
- Validation Modes - Configure validation behavior
- Breaking Changes Guide - Understand schema compatibility
- API Reference - Full API documentation