Move zero-column exchange test to conformance suite and bump version to 0.1.11

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: rustyconover

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "vgi-rpc"
-version = "0.1.10"
+version = "0.1.11"
 description = "Vector Gateway Interface - RPC framework based on Apache Arrow"
 readme = "README.md"
 requires-python = ">=3.13"

tests/test_conformance.py

@@ -811,6 +811,17 @@ def test_empty_exchange_session(self, conformance_conn: ConnFactory) -> None:
             # Verify transport is still usable
             assert proxy.echo_int(value=42) == 42
 
+    def test_zero_column_exchange(self, conformance_conn: ConnFactory) -> None:
+        """Exchange stream with zero-column batches works over 100 iterations."""
+        empty_schema = pa.schema([])
+        empty_input = AnnotatedBatch(batch=pa.record_batch([], schema=empty_schema))
+        with conformance_conn() as proxy, proxy.exchange_zero_columns() as session:
+            for _ in range(100):
+                output = session.exchange(empty_input)
+                assert output.batch.schema == empty_schema
+                assert output.batch.num_rows == 0
+                assert output.batch.num_columns == 0
+
     def test_zero_row_input(self, conformance_conn: ConnFactory) -> None:
         """Send zero-row batch to exchange."""
         with conformance_conn() as proxy, proxy.exchange_scale(factor=2.0) as session:
@@ -919,7 +930,7 @@ def test_run_describe_conformance(self, service_description: ServiceDescription)
 
     def test_describe_via_rpc(self, service_description: ServiceDescription) -> None:
         """Smoke test: basic transport-level describe call works."""
-        assert len(service_description.methods) == 46
+        assert len(service_description.methods) == 47
         assert service_description.protocol_name == "ConformanceService"
         echo_str = service_description.methods["echo_string"]
         assert echo_str.method_type == MethodType.UNARY

tests/test_rpc.py

@@ -222,18 +222,6 @@ def process(self, input: AnnotatedBatch, out: OutputCollector, ctx: CallContext)
         out.finish()
 
 
-@dataclass
-class ZeroColumnExchangeState(ExchangeState):
-    """State for an exchange stream with zero-column batches."""
-
-    call_count: int = 0
-
-    def exchange(self, input: AnnotatedBatch, out: OutputCollector, ctx: CallContext) -> None:
-        """Accept a zero-column batch and emit a zero-column batch."""
-        self.call_count += 1
-        out.emit(pa.record_batch([], schema=pa.schema([])))
-
-
 # ---------------------------------------------------------------------------
 # Test fixtures: Protocol + Implementation
 # ---------------------------------------------------------------------------
@@ -349,10 +337,6 @@ def transform_with_header(self, factor: float) -> Stream[ExchangeState, StreamHe
         """Exchange stream with a stream header."""
         ...
 
-    def zero_column_exchange(self) -> Stream[ExchangeState]:
-        """Exchange stream with zero-column input and output."""
-        ...
-
     def fail_stream_init_with_header(self) -> Stream[ProducerState, StreamHeader]:
         """Stream that raises during init (with header declared)."""
         ...
@@ -466,11 +450,6 @@ def transform_with_header(self, factor: float) -> Stream[TransformWithHeaderStat
             output_schema=schema, state=TransformWithHeaderState(factor=factor), input_schema=schema, header=header
         )
 
-    def zero_column_exchange(self) -> Stream[ZeroColumnExchangeState]:
-        """Exchange stream with zero-column input and output."""
-        empty = pa.schema([])
-        return Stream(output_schema=empty, state=ZeroColumnExchangeState(), input_schema=empty)
-
     def fail_stream_init_with_header(self) -> Stream[GenerateWithHeaderState, StreamHeader]:
         """Stream that raises during init (with header declared)."""
         raise ValueError("init boom with header")
@@ -742,17 +721,6 @@ def test_bidi_context_manager(self, make_conn: ConnFactory) -> None:
             output = session.exchange(AnnotatedBatch(batch=pa.RecordBatch.from_pydict({"value": [5.0]})))
             assert output.batch.column("value").to_pylist() == [10.0]
 
-    def test_zero_column_exchange_100_batches(self, make_conn: ConnFactory) -> None:
-        """Exchange stream with zero-column batches works over 100 iterations."""
-        empty_schema = pa.schema([])
-        empty_input = AnnotatedBatch(batch=pa.record_batch([], schema=empty_schema))
-        with make_conn() as proxy, proxy.zero_column_exchange() as session:
-            for _ in range(100):
-                output = session.exchange(empty_input)
-                assert output.batch.schema == empty_schema
-                assert output.batch.num_rows == 0
-                assert output.batch.num_columns == 0
-
 
 # ---------------------------------------------------------------------------
 # Tests: RpcConnection (context manager + proxy)

uv.lock

@@ -39,6 +39,7 @@
     ScaleExchangeState,
     SingleProducerState,
     Status,
+    ZeroColumnExchangeState,
     build_dynamic_schema,
     build_rich_header,
 )
@@ -293,6 +294,11 @@ def exchange_with_logs(self) -> Stream[LoggingExchangeState]:
             input_schema=_LOGGING_EXCHANGE_INPUT,
         )
 
+    def exchange_zero_columns(self) -> Stream[ZeroColumnExchangeState]:
+        """Exchange stream with zero-column input and output."""
+        empty = pa.schema([])
+        return Stream(output_schema=empty, state=ZeroColumnExchangeState(), input_schema=empty)
+
     def exchange_error_on_nth(self, fail_on: int) -> Stream[FailOnExchangeNState]:
         """Raise on Nth exchange."""
         return Stream(

vgi_rpc/conformance/_impl.py

@@ -3,7 +3,7 @@
 
 """ConformanceService Protocol definition.
 
-Defines ~43 RPC methods covering every framework capability:
+Defines ~44 RPC methods covering every framework capability:
 scalar echo, void, complex types, optionals, dataclass round-trip,
 annotated types, multi-param, errors, logging, producer streams,
 exchange streams, headers, and introspection.
@@ -249,6 +249,10 @@ def exchange_error_on_nth(self, fail_on: int) -> Stream[StreamState]:
         """Raise on the Nth exchange (1-indexed)."""
         ...
 
+    def exchange_zero_columns(self) -> Stream[StreamState]:
+        """Exchange stream with zero-column input and output."""
+        ...
+
     def exchange_error_on_init(self) -> Stream[StreamState]:
         """Raise during exchange stream initialization."""
         ...

vgi_rpc/conformance/_protocol.py

@@ -1080,6 +1080,7 @@ def decorator(
         "exchange_with_header",
         "exchange_with_logs",
         "exchange_with_rich_header",
+        "exchange_zero_columns",
         "inspect_point",
         "produce_dynamic_schema",
         "produce_empty",
@@ -1143,6 +1144,7 @@ def decorator(
         "exchange_with_header",
         "exchange_with_logs",
         "exchange_with_rich_header",
+        "exchange_zero_columns",
         "produce_dynamic_schema",
         "produce_empty",
         "produce_error_mid_stream",
@@ -1201,7 +1203,7 @@ def _test_desc_describe_version(desc: ServiceDescription) -> None:
 
 @_describe_test(category="describe_service", name="method_count")
 def _test_desc_method_count(desc: ServiceDescription) -> None:
-    assert len(desc.methods) == 46
+    assert len(desc.methods) == 47
 
 
 # ---------------------------------------------------------------------------

vgi_rpc/conformance/_runner.py

@@ -264,6 +264,18 @@ def exchange(self, input: AnnotatedBatch, out: OutputCollector, ctx: CallContext
         out.emit(input.batch)
 
 
+@dataclass
+class ZeroColumnExchangeState(ExchangeState):
+    """Accept and emit zero-column batches."""
+
+    call_count: int = 0
+
+    def exchange(self, input: AnnotatedBatch, out: OutputCollector, ctx: CallContext) -> None:
+        """Accept a zero-column batch and emit a zero-column batch."""
+        self.call_count += 1
+        out.emit(pa.record_batch([], schema=pa.schema([])))
+
+
 @dataclass
 class FailOnExchangeNState(ExchangeState):
     """Raise on the Nth exchange (1-indexed)."""