Add encode_collection to serializer

src/pyipp/__init__.py

@@ -1,4 +1,5 @@
 """Asynchronous Python client for IPP."""
+
 from .exceptions import (
     IPPConnectionError,
     IPPConnectionUpgradeRequired,

src/pyipp/enums.py

@@ -1,4 +1,5 @@
 """Enumerators for IPP."""
+
 from enum import IntEnum
 
 
@@ -316,19 +317,19 @@ class IppOrientationRequested(IntEnum):
 
 
 ATTRIBUTE_ENUM_MAP = {
-    "document-state": IppDocumentState, # PWG5100.5
-    "finishings": IppFinishing, # RFC8011
-    "finishings-default": IppFinishing, # RFC8011
-    "finishings-supported": IppFinishing, # RFC8011
-    "job-state": IppJobState, # RFC8011
-    "media-source-feed-orientation": IppOrientationRequested, # PWG5100.7
-    "operations-supported": IppOperation, # RFC8011
-    "orientation-requested": IppOrientationRequested, # RFC8011
-    "orientation-requested-default": IppOrientationRequested, # RFC8011
-    "orientation-requested-supported": IppOrientationRequested, # RFC8011
-    "printer-state": IppPrinterState, # RFC8011
-    "print-quality": IppPrintQuality, # RFC8011
-    "print-quality-default": IppPrintQuality, # RFC8011
-    "print-quality-supported": IppPrintQuality, # RFC8011
-    "status-code": IppStatus, # RFC8011
+    "document-state": IppDocumentState,  # PWG5100.5
+    "finishings": IppFinishing,  # RFC8011
+    "finishings-default": IppFinishing,  # RFC8011
+    "finishings-supported": IppFinishing,  # RFC8011
+    "job-state": IppJobState,  # RFC8011
+    "media-source-feed-orientation": IppOrientationRequested,  # PWG5100.7
+    "operations-supported": IppOperation,  # RFC8011
+    "orientation-requested": IppOrientationRequested,  # RFC8011
+    "orientation-requested-default": IppOrientationRequested,  # RFC8011
+    "orientation-requested-supported": IppOrientationRequested,  # RFC8011
+    "printer-state": IppPrinterState,  # RFC8011
+    "print-quality": IppPrintQuality,  # RFC8011
+    "print-quality-default": IppPrintQuality,  # RFC8011
+    "print-quality-supported": IppPrintQuality,  # RFC8011
+    "status-code": IppStatus,  # RFC8011
 }

src/pyipp/ipp.py

@@ -1,4 +1,5 @@
 """Asynchronous Python client for IPP."""
+
 from __future__ import annotations
 
 import asyncio
@@ -42,6 +43,7 @@
 
 VERSION = metadata.version(__package__)
 
+
 @dataclass
 class IPP:
     """Main class for handling connections with IPP servers."""
@@ -242,7 +244,7 @@ async def printer(self) -> Printer:
 
         return self._printer
 
-    async def __aenter__(self) -> IPP:   # noqa: PYI034
+    async def __aenter__(self) -> IPP:  # noqa: PYI034
         """Async enter."""
         return self
 

src/pyipp/models.py

@@ -1,4 +1,5 @@
 """Models for IPP."""
+
 # pylint: disable=R0912,R0915
 from __future__ import annotations
 
@@ -172,7 +173,6 @@ def update_from_dict(self, data: dict[str, Any]) -> Printer:
 
         return self
 
-
     @staticmethod
     def from_dict(data: dict[str, Any]) -> Printer:
         """Return Printer object from IPP response data."""
@@ -317,6 +317,7 @@ def _utcnow() -> datetime:
     """Return the current date and time in UTC."""
     return datetime.now(tz=timezone.utc)
 
+
 def _str_or_none(value: str) -> str | None:
     """Return string while handling string representations of None."""
     if value == "none":

src/pyipp/parser.py

@@ -1,4 +1,5 @@
 """Response Parser for IPP."""
+
 from __future__ import annotations
 
 import logging
@@ -91,7 +92,9 @@ def parse_attribute(  # noqa: PLR0912, PLR0915
 
     if attribute["name"]:
         _LOGGER.debug(
-            "Attribute Name: %s (%s)", attribute["name"], hex(attribute["tag"]),
+            "Attribute Name: %s (%s)",
+            attribute["name"],
+            hex(attribute["tag"]),
         )
     else:
         _LOGGER.debug("Attribute Tag: %s", hex(attribute["tag"]))

src/pyipp/serializer.py

@@ -1,4 +1,5 @@
 """Data Serializer for IPP."""
+
 from __future__ import annotations
 
 import logging
@@ -61,6 +62,52 @@ def construct_attribute(name: str, value: Any, tag: IppTag | None = None) -> byt
     return byte_str
 
 
+def encode_collection(name: str, collection: dict[str, Any]) -> bytes:
+    """Encode a dict representing an IPP collection as a byte string.
+
+    Args:
+    ----
+        name (str): The name of the collection
+        collection (dict[str, Any]): The collection contents
+
+    Returns:
+    -------
+        bytes: A binary string representing the collection
+
+    """
+    byte_str = b""
+
+    byte_str += struct.pack(">b", IppTag.BEGIN_COLLECTION.value)
+    byte_str += struct.pack(">h", len(name))
+    byte_str += name.encode("utf-8")
+    byte_str += struct.pack(">h", 0)
+
+    for member_name, value in collection.items():
+        if isinstance(value, dict):
+            byte_str += encode_collection(name=member_name, collection=value)
+        else:
+            byte_str += struct.pack(">b", IppTag.MEMBER_NAME.value)
+            byte_str += struct.pack(">h", 0)
+            byte_str += struct.pack(">h", len(member_name))
+            byte_str += member_name.encode("utf-8")
+            if isinstance(value, int):
+                byte_str += struct.pack(">b", IppTag.INTEGER.value)
+                byte_str += struct.pack(">h", 0)
+                byte_str += struct.pack(">h", 4)
+                byte_str += struct.pack(">h", value)
+            else:
+                byte_str += struct.pack(">b", IppTag.KEYWORD.value)
+                byte_str += struct.pack(">h", 0)
+                byte_str += struct.pack(">h", len(value))
+                byte_str += value.encode("utf-8")
+
+    byte_str += struct.pack(">b", IppTag.END_COLLECTION.value)
+    byte_str += struct.pack(">h", 0)
+    byte_str += struct.pack(">h", 0)
+
+    return byte_str
+
+
 def encode_dict(data: dict[str, Any]) -> bytes:
     """Serialize a dictionary of data into IPP format."""
     version = data["version"] or DEFAULT_PROTO_VERSION
@@ -83,7 +130,11 @@ def encode_dict(data: dict[str, Any]) -> bytes:
         encoded += struct.pack(">b", IppTag.JOB.value)
 
         for attr, value in data["job-attributes-tag"].items():
-            encoded += construct_attribute(attr, value)
+            encoded += (
+                encode_collection(attr, value)
+                if isinstance(value, dict)
+                else construct_attribute(attr, value)
+            )
 
     if isinstance(data.get("printer-attributes-tag"), dict):
         encoded += struct.pack(">b", IppTag.PRINTER.value)

tests/__init__.py

@@ -1,4 +1,5 @@
 """Tests for IPP."""
+
 from __future__ import annotations
 
 import os

tests/test_client.py

@@ -1,4 +1,5 @@
 """Tests for IPP Client."""
+
 import asyncio
 
 import pytest

tests/test_interface.py

@@ -1,4 +1,5 @@
 """Tests for IPP public interface."""
+
 import pytest
 from aiohttp import ClientSession
 from aresponses import ResponsesMockServer

tests/test_models.py

@@ -1,4 +1,5 @@
 """Tests for IPP Models."""
+
 # pylint: disable=R0912,R0915
 from __future__ import annotations
 
@@ -193,6 +194,7 @@ async def test_printer() -> None:  # noqa: PLR0915
     assert printer.uris[1].authentication is None
     assert printer.uris[1].security is None
 
+
 def test_printer_as_dict() -> None:
     """Test the dictionary version of Printer."""
     parsed = parser.parse(load_fixture_binary("get-printer-attributes-epsonxp6000.bin"))
@@ -210,6 +212,7 @@ def test_printer_as_dict() -> None:
     assert isinstance(printer_dict["uris"], List)
     assert len(printer_dict["uris"]) == 2
 
+
 def test_printer_update_from_dict() -> None:
     """Test updating data of Printer."""
     parsed = parser.parse(load_fixture_binary("get-printer-attributes-epsonxp6000.bin"))
@@ -226,6 +229,7 @@ def test_printer_update_from_dict() -> None:
     assert printer.info
     assert printer.info.uptime == 2
 
+
 @pytest.mark.asyncio
 async def test_printer_with_single_marker() -> None:
     """Test Printer model with single marker."""

tests/test_parser.py

@@ -1,4 +1,5 @@
 """Tests for Parser."""
+
 import pytest
 from syrupy.assertion import SnapshotAssertion
 

tests/test_serializer.py

@@ -1,4 +1,5 @@
 """Tests for Serializer."""
+
 from pyipp import serializer
 from pyipp.const import DEFAULT_CHARSET, DEFAULT_CHARSET_LANGUAGE, DEFAULT_PROTO_VERSION
 from pyipp.enums import IppOperation, IppTag