feature: Represent UUID zero as None (#37)

src/kio/constants.py

@@ -0,0 +1,4 @@
+import uuid
+from typing import Final
+
+uuid_zero: Final = uuid.UUID(int=0)

src/kio/serial/decoders.py

@@ -12,6 +12,7 @@
 from typing import assert_never
 from uuid import UUID
 
+from kio.constants import uuid_zero
 from kio.schema.primitive import i8
 from kio.schema.primitive import i16
 from kio.schema.primitive import i32
@@ -194,8 +195,10 @@ def decode_compact_array_length() -> Cursor[int]:
     return decoded_value - 1
 
 
-def decode_uuid() -> Cursor[UUID]:
+def decode_uuid() -> Cursor[UUID | None]:
     byte_value: bytes = yield 16
+    if byte_value == uuid_zero.bytes:
+        return None
     return UUID(bytes=byte_value)
 
 

src/kio/serial/encoders.py

@@ -10,6 +10,7 @@
 from typing import TypeVar
 from uuid import UUID
 
+from kio.constants import uuid_zero
 from kio.schema.primitive import i8
 from kio.schema.primitive import i16
 from kio.schema.primitive import i32
@@ -160,8 +161,11 @@ def write_compact_array_length(buffer: Writable, value: int) -> None:
     write_unsigned_varint(buffer, value + 1)
 
 
-def write_uuid(buffer: Writable, value: UUID) -> None:
-    buffer.write(value.bytes)
+def write_uuid(buffer: Writable, value: UUID | None) -> None:
+    if value is None:
+        buffer.write(uuid_zero.bytes)
+    else:
+        buffer.write(value.bytes)
 
 
 def compact_array_writer(item_writer: Writer[T]) -> Writer[Sequence[T]]:

tests/serial/test_decoders.py

@@ -2,10 +2,12 @@
 import io
 import struct
 import sys
+import uuid
 from typing import IO
 
 import pytest
 
+from kio.constants import uuid_zero
 from kio.serial.decoders import Decoder
 from kio.serial.decoders import decode_compact_string
 from kio.serial.decoders import decode_compact_string_as_bytes
@@ -26,6 +28,7 @@
 from kio.serial.decoders import decode_uint32
 from kio.serial.decoders import decode_uint64
 from kio.serial.decoders import decode_unsigned_varint
+from kio.serial.decoders import decode_uuid
 from kio.serial.decoders import read_async
 from kio.serial.decoders import read_sync
 from kio.serial.errors import UnexpectedNull
@@ -631,3 +634,16 @@ async def test_can_decode_bytes_async(
         stream_writer.write(byte_value)
         await stream_writer.drain()
         assert byte_value == await read_async(stream_reader, decode_legacy_bytes)
+
+
+class TestDecodeUUID:
+    def test_decodes_zero_as_none(self, buffer: io.BytesIO) -> None:
+        buffer.write(uuid_zero.bytes)
+        buffer.seek(0)
+        assert read_sync(buffer, decode_uuid) is None
+
+    def test_can_decode_uuid4(self, buffer: io.BytesIO) -> None:
+        value = uuid.uuid4()
+        buffer.write(value.bytes)
+        buffer.seek(0)
+        assert read_sync(buffer, decode_uuid) == value

tests/serial/test_encoders.py

@@ -1,12 +1,14 @@
 import io
 import struct
 import sys
+import uuid
 from contextlib import closing
 from typing import Generic
 from typing import TypeVar
 
 import pytest
 
+from kio.constants import uuid_zero
 from kio.serial.encoders import Writer
 from kio.serial.encoders import write_compact_string
 from kio.serial.encoders import write_empty_tagged_fields
@@ -23,6 +25,7 @@
 from kio.serial.encoders import write_uint32
 from kio.serial.encoders import write_uint64
 from kio.serial.encoders import write_unsigned_varint
+from kio.serial.encoders import write_uuid
 from kio.serial.errors import OutOfBoundValue
 
 _I = TypeVar("_I", bound=int, contravariant=True)
@@ -298,3 +301,16 @@ def test_raises_out_of_bound_value_for_too_large_string(
     ) -> None:
         with pytest.raises(OutOfBoundValue):
             write_nullable_legacy_string(buffer, 2**15 * "a")
+
+
+class TestWriteUUID:
+    def test_writes_none_as_uuid_zero(self, buffer: io.BytesIO) -> None:
+        write_uuid(buffer, None)
+        buffer.seek(0)
+        assert buffer.read(16) == uuid_zero.bytes
+
+    def test_can_write_uuid4(self, buffer: io.BytesIO) -> None:
+        value = uuid.uuid4()
+        write_uuid(buffer, value)
+        buffer.seek(0)
+        assert buffer.read(16) == value.bytes

tests/serial/test_roundtrips.py

@@ -290,8 +290,8 @@ async def test_legacy_bytes_roundtrip_async(a: bytes, b: bytes) -> None:
         assert b == await read_async(stream_reader, decode_legacy_bytes)
 
 
-@given(uuids(), uuids())
-def test_uuid_roundtrip_sync(a: uuid.UUID, b: uuid.UUID) -> None:
+@given(uuids() | none(), uuids() | none())
+def test_uuid_roundtrip_sync(a: uuid.UUID | None, b: uuid.UUID | None) -> None:
     buffer = io.BytesIO()
     write_uuid(buffer, a)
     write_uuid(buffer, b)

tests/test_integration.py

@@ -1,12 +1,10 @@
 import asyncio
 import io
 import secrets
-import uuid
 from asyncio import StreamReader
 from asyncio import StreamWriter
 from contextlib import closing
 from typing import Any
-from typing import Final
 from typing import TypeVar
 from unittest import mock
 
@@ -15,6 +13,7 @@
 import kio.schema.request_header.v2.header
 import kio.schema.response_header.v0.header
 import kio.schema.response_header.v1.header
+from kio.constants import uuid_zero
 from kio.schema.api_versions.v2 import request as api_versions_v2_request
 from kio.schema.api_versions.v2 import response as api_versions_v2_response
 from kio.schema.api_versions.v3 import request as api_versions_v3_request
@@ -44,8 +43,6 @@
 from kio.serial.encoders import Writable
 from kio.serial.encoders import write_int32
 
-uuid_zero: Final = uuid.UUID(int=0)
-
 
 def write_request_header(
     buffer: Writable,