Support different types of bytes input (#82)

Previously, we required bytes input as a BufferedReader.  With this
patch, we also accept FileIO and bytes instances, as supported by the
generated API client.

Fixes: #63

nethsm/__init__.py

@@ -17,7 +17,7 @@
 from base64 import b64encode
 from dataclasses import dataclass
 from datetime import datetime
-from io import BufferedReader
+from io import BufferedReader, FileIO
 from typing import TYPE_CHECKING, Any, Iterator, Literal, Mapping, Optional, Union, cast
 from urllib.parse import urlencode
 
@@ -31,6 +31,9 @@
     from .client.apis.tags.default_api import DefaultApi
 
 
+Bytes = Union[BufferedReader, FileIO, bytes]
+
+
 class Role(enum.Enum):
     ADMINISTRATOR = "Administrator"
     OPERATOR = "Operator"
@@ -363,7 +366,7 @@ def request(
         method: str,
         endpoint: str,
         params: Optional[dict[str, str]] = None,
-        data: Optional[BufferedReader] = None,
+        data: Optional[Bytes] = None,
         mime_type: Optional[str] = "application/json",
         json_obj: Optional[Any] = None,
     ) -> HTTPResponse:
@@ -387,7 +390,10 @@ def request(
 
         body: Union[str, bytes, None] = None
         if data:
-            body = data.read(-1)
+            if isinstance(data, bytes):
+                body = data
+            else:
+                body = data.read()
         elif json_obj:
             encoder = json.JSONEncoder()
             body = encoder.encode(json_obj)
@@ -1016,7 +1022,7 @@ def get_key_certificate(self, key_id: str) -> bytes:
             )
         return response.response.data
 
-    def set_certificate(self, cert: BufferedReader) -> None:
+    def set_certificate(self, cert: Bytes) -> None:
         try:
             self.request(
                 "PUT",
@@ -1034,7 +1040,7 @@ def set_certificate(self, cert: BufferedReader) -> None:
                 },
             )
 
-    def set_key_certificate(self, key_id: str, cert: BufferedReader) -> None:
+    def set_key_certificate(self, key_id: str, cert: Bytes) -> None:
         try:
             from .client.paths.keys_key_id_cert.put.path_parameters import (
                 PathParametersDict,
@@ -1304,7 +1310,7 @@ def backup(self) -> bytes:
             )
         return response.response.data
 
-    def restore(self, backup: BufferedReader, passphrase: str, time: datetime) -> None:
+    def restore(self, backup: Bytes, passphrase: str, time: datetime) -> None:
         try:
             from .client.paths.system_restore.post.request_body.content.multipart_form_data.schema import (
                 ArgumentsDict,
@@ -1328,7 +1334,7 @@ def restore(self, backup: BufferedReader, passphrase: str, time: datetime) -> No
                 },
             )
 
-    def update(self, image: BufferedReader) -> str:
+    def update(self, image: Bytes) -> str:
         try:
             response = self.get_api().system_update_post(body=image)
         except Exception as e:

tests/test_nethsm_config.py

@@ -74,7 +74,7 @@ def test_set_certificate(nethsm: NetHSM) -> None:
         C.EMAIL_ADDRESS,
     )
     cert = self_sign_csr(csr)
-    nethsm.set_certificate(BytesIO(cert))  # type: ignore
+    nethsm.set_certificate(cert)
 
     remote_cert = nethsm.get_certificate()
     assert cert.decode("utf-8") == remote_cert