“

Series: S6 Platform Engineering · Article 17 | Level: Advanced | Reading Time: 18 min

SDK Testing Strategy: Comprehensive Coverage from Unit Tests to Contract Tests

#TL;DR

The coomia-dip SDK testing strategy spans five layers: unit tests (logic correctness), integration tests (gRPC communication), contract tests (API compatibility), end-to-end tests (real scenarios), and performance tests (benchmark regression). Testing infrastructure includes Mock gRPC servers, schema fixture generators, snapshot testing, and CI pipeline integration. This article covers the test pyramid, implementation for each layer, testing infrastructure, and CI/CD integration.

#1. Test Pyramid

Code

           /\
          /  \         E2E Tests (5%)
         /----\        Real platform, full flows
        /      \
       /--------\      Performance Tests (5%)
      /          \     Benchmark regression, SLA verification
     /------------\
    /              \   Contract Tests (15%)
   /----------------\  API compatibility, Protobuf contracts
  /                  \
 /--------------------\ Integration Tests (25%)
/                      \ gRPC communication, serialization, interceptors
/------------------------\
/                          \ Unit Tests (50%)
/----------------------------\ Models, filters, query builders

#2. Unit Tests

#2.1 Model Tests

Python

class TestOntologyModels:
    def test_object_creation(self):
        employee = Employee(
            id="emp-001", name="John", department="Engineering",
            salary=50000, hire_date=date(2024, 1, 15),
        )
        assert employee.name == "John"
        assert employee.salary == 50000

    def test_object_validation(self):
        with pytest.raises(ValidationError) as exc_info:
            Employee(id="emp-001", name="John", department="Engineering",
                     salary="not_a_number", hire_date=date(2024, 1, 15))
        assert "salary" in str(exc_info.value)

    def test_object_serialization(self):
        employee = Employee(id="emp-001", name="John", department="Eng",
                           salary=50000, hire_date=date(2024, 1, 15))
        data = employee.to_dict()
        restored = Employee.from_dict(data)
        assert restored == employee

#2.2 Query Builder Tests

Python

class TestQueryBuilder:
    def test_simple_filter(self):
        query = QueryBuilder("Employee").where(
            Filter("department", "eq", "Engineering")
        ).build()
        assert query.filters[0].field == "department"

    def test_compound_filter(self):
        query = (QueryBuilder("Employee")
            .where(Filter("department", "eq", "Engineering"))
            .where(Filter("salary", "gt", 50000))
            .build())
        assert len(query.filters) == 2

    def test_select_fields(self):
        query = QueryBuilder("Employee").select("name", "salary").build()
        assert query.selected_fields == ["name", "salary"]

    def test_ordering(self):
        query = QueryBuilder("Employee").order_by("salary", "desc").build()
        assert query.order_by[0].field == "salary"

#2.3 Error Mapping Tests

Python

class TestErrorMapping:
    def test_not_found_mapping(self):
        grpc_error = make_grpc_error(grpc.StatusCode.NOT_FOUND, "Not found")
        sdk_error = GrpcErrorMapper().map(grpc_error)
        assert isinstance(sdk_error, ObjectNotFoundError)

    def test_permission_denied_mapping(self):
        grpc_error = make_grpc_error(grpc.StatusCode.PERMISSION_DENIED, "Denied")
        sdk_error = GrpcErrorMapper().map(grpc_error)
        assert isinstance(sdk_error, AuthorizationError)

#3. Integration Tests

#3.1 Mock gRPC Server

Python

class MockOntologyServer:
    def __init__(self):
        self._objects: dict[str, dict[str, dict]] = {}
        self._server: grpc.aio.Server | None = None

    async def start(self, port: int = 0) -> int:
        self._server = grpc.aio.server()
        add_OntologyServiceServicer_to_server(
            MockOntologyServicer(self._objects), self._server,
        )
        actual_port = self._server.add_insecure_port(f"[::]:{port}")
        await self._server.start()
        return actual_port

    async def stop(self):
        await self._server.stop(grace=5)

    def seed_data(self, object_type: str, objects: list[dict]):
        self._objects[object_type] = {obj["id"]: obj for obj in objects}

#3.2 Integration Test Cases

Python

class TestGrpcIntegration:
    @pytest.fixture
    async def mock_server(self):
        server = MockOntologyServer()
        server.seed_data("Employee", [
            {"id": "emp-001", "name": "John", "department": "Engineering", "salary": 50000},
        ])
        port = await server.start()
        yield f"localhost:{port}"
        await server.stop()

    @pytest.mark.asyncio
    async def test_get_object(self, mock_server):
        async with AsyncOntoPlatform.connect(mock_server) as client:
            employee = await client.objects.get("Employee", "emp-001")
            assert employee.name == "John"

    @pytest.mark.asyncio
    async def test_not_found(self, mock_server):
        async with AsyncOntoPlatform.connect(mock_server) as client:
            with pytest.raises(ObjectNotFoundError):
                await client.objects.get("Employee", "nonexistent")

#4. Contract Tests

Python

class TestProtobufContract:
    def test_request_message_fields(self):
        request = GetObjectRequest(object_type="Employee", object_id="emp-001")
        assert request.object_type
        assert request.object_id

    def test_response_backward_compatible(self):
        old_response_data = {"object": {"id": "emp-001", "properties": {"name": "John"}}}
        response = GetObjectResponse()
        Parse(json.dumps(old_response_data), response)
        assert response.object.id == "emp-001"

    def test_enum_values_stable(self):
        assert FilterOperator.Value("EQUALS") == 1
        assert FilterOperator.Value("GREATER_THAN") == 2

class TestAPICompatibility:
    def test_sdk_supports_server_v1(self):
        negotiator = VersionNegotiator(sdk_version="1.2.0")
        result = negotiator.negotiate(server_version="1.0.0")
        assert result.compatible

#5. End-to-End Tests

Python

class TestEndToEnd:
    @pytest.fixture
    def platform_client(self):
        url = os.environ.get("ONTO_PLATFORM_URL", "http://localhost:8080")
        token = os.environ.get("ONTO_PLATFORM_TOKEN")
        if not token:
            pytest.skip("No platform token configured")
        return SyncOntoPlatform.connect(url, token=token)

    def test_full_crud_lifecycle(self, platform_client):
        # Create
        employee = platform_client.objects.create("Employee", {
            "name": "E2E Test User", "department": "Testing",
            "salary": 50000, "hire_date": "2024-01-01",
        })
        assert employee.id is not None

        # Read
        fetched = platform_client.objects.get("Employee", employee.id)
        assert fetched.name == "E2E Test User"

        # Update
        updated = platform_client.objects.update("Employee", employee.id, {"salary": 55000})
        assert updated.salary == 55000

        # Delete
        platform_client.objects.delete("Employee", employee.id)
        with pytest.raises(ObjectNotFoundError):
            platform_client.objects.get("Employee", employee.id)

#6. Performance Tests

Python

class TestPerformance:
    @pytest.mark.benchmark
    def test_serialization_performance(self, benchmark):
        employee = Employee(id="emp-001", name="John", department="Eng",
                           salary=50000, hire_date=date(2024, 1, 15))
        result = benchmark(employee.to_protobuf)
        assert result is not None

    @pytest.mark.asyncio
    async def test_latency_sla(self, mock_server):
        async with AsyncOntoPlatform.connect(mock_server) as client:
            latencies = []
            for _ in range(100):
                start = time.monotonic()
                await client.objects.get("Employee", "emp-001")
                latencies.append(time.monotonic() - start)
            p99 = sorted(latencies)[98]
            assert p99 < 0.1  # P99 < 100ms

#7. Testing Infrastructure

#7.1 Fixture Generator

Python

class SchemaFixtureGenerator:
    @staticmethod
    def generate_employee_schema() -> ObjectTypeSchema:
        return ObjectTypeSchema(
            api_name="Employee",
            properties=[
                PropertySchema(name="name", data_type="string", required=True),
                PropertySchema(name="department", data_type="string", required=True),
                PropertySchema(name="salary", data_type="integer", required=True),
                PropertySchema(name="hire_date", data_type="date", required=True),
            ],
        )

    @staticmethod
    def generate_employee_data(count: int = 10) -> list[dict]:
        return [
            {"id": f"emp-{i:03d}", "name": f"Employee {i}",
             "department": random.choice(["Engineering", "Marketing"]),
             "salary": random.randint(40000, 120000),
             "hire_date": date(2020, 1, 1) + timedelta(days=random.randint(0, 1500))}
            for i in range(count)
        ]

#7.2 CI Integration

YAML

sdk-test:
  stage: test
  script:
    - pip install -e ".[dev]"
    - pytest tests/unit/ -v --cov=ontology_sdk
    - pytest tests/integration/ -v --cov-append
    - pytest tests/contract/ -v --cov-append

sdk-e2e:
  stage: e2e
  when: manual
  script:
    - pytest tests/e2e/ -v

sdk-performance:
  stage: performance
  script:
    - pytest tests/performance/ --benchmark-json=benchmark.json

#8. Summary

The coomia-dip SDK testing strategy ensures quality and reliability through five-layer coverage:

Unit tests (50%): Model validation, query builders, error mapping
Integration tests (25%): Mock gRPC server verifies communication correctness
Contract tests (15%): Protobuf contracts and API compatibility
Performance tests (5%): Serialization benchmarks, throughput and latency SLA
E2E tests (5%): Full CRUD lifecycle against real platform

The next article will explore the coomia-dip Docker Compose deployment strategy.