feat: modifies bundles.create and bundle.download to accept multiple …

…input types

src/posit/connect/bundles.py

@@ -1,9 +1,9 @@
 from __future__ import annotations
+import io
 
-from typing import BinaryIO, List
-
-from requests.sessions import Session as Session
+import requests
 
+from typing import List
 
 from . import config, resources, urls
 
@@ -96,60 +96,109 @@ def delete(self) -> None:
         url = urls.append(self.config.url, path)
         self.session.delete(url)
 
-    def download(self) -> bytes:
+    def download(self, output: io.BufferedWriter | str):
         """Download a bundle.
 
-        Returns
-        -------
-        bytes
-            Archive contents in bytes representation.
+        Download a bundle to a file or memory.
+
+        Parameters
+        ----------
+        output: io.BufferedWriter | str
+            An io.BufferedWriter instance or a str representing a relative or absolute path.
+
+        Raises
+        ------
+        TypeError
+            If the output is not of type `io.BufferedWriter` or `str`.
 
         Examples
         --------
-        >>> with open('archive.tar.gz', 'wb') as file:
-        >>>     data = bundle.download()
-        >>>     file.write(data)
+        Write to a file.
+        >>> bundle.download("bundle.tar.gz")
+        None
+
+        Write to an io.BufferedWriter.
+        >>> with open('bundle.tar.gz', 'wb') as file:
+        >>>     bundle.download(file)
+        None
         """
+        if not isinstance(output, (io.BufferedWriter, str)):
+            raise TypeError(
+                f"download() expected argument type 'io.BufferedWriter` or 'str', but got '{type(input).__name__}'"
+            )
+
         path = f"v1/content/{self.content_guid}/bundles/{self.id}/download"
         url = urls.append(self.config.url, path)
         response = self.session.get(url, stream=True)
-        return response.content
+        if isinstance(output, io.BufferedWriter):
+            for chunk in response.iter_content():
+                output.write(chunk)
+            return
+
+        if isinstance(output, str):
+            with open(output, "wb") as file:
+                for chunk in response.iter_content():
+                    file.write(chunk)
+            return
 
 
 class Bundles(resources.Resources):
     def __init__(
-        self, config: config.Config, session: Session, content_guid: str
+        self,
+        config: config.Config,
+        session: requests.Session,
+        content_guid: str,
     ) -> None:
         super().__init__(config, session)
         self.content_guid = content_guid
 
-    def create(self, data: BinaryIO | bytes) -> Bundle:
+    def create(self, input: io.BufferedReader | bytes | str) -> Bundle:
         """Create a bundle.
 
-        Create a bundle by upload via archive format.
+        Create a bundle from a file or memory.
 
         Parameters
         ----------
-        data : BinaryIO | bytes
-            Archive contents in BinaryIO or bytes representation.
+        input : io.BufferedReader | bytes | str
+            Input archive for bundle creation. A 'str' type assumes a relative or absolute filepath.
 
         Returns
         -------
         Bundle
+            The created bundle.
+
+        Raises
+        ------
+        TypeError
+            If the input is not of type `io.BufferedReader`, `bytes`, or `str`.
 
         Examples
         --------
-        Create a bundle using a file object.
-
+        Create a bundle from io.BufferedReader
         >>> with open('bundle.tar.gz', 'rb') as file:
         >>>     bundle.create(file)
+        None
 
-        Create a bundle using bytes.
-
+        Create a bundle from bytes.
         >>> with open('bundle.tar.gz', 'rb') as file:
-        >>>     data = file.read()
+        >>>     data: bytes = file.read()
         >>>     bundle.create(data)
+        None
+
+        Create a bundle from pathname.
+        >>> bundle.create("bundle.tar.gz")
+        None
         """
+        if isinstance(input, (io.BufferedReader, bytes)):
+            data = input
+        elif isinstance(input, str):
+            with open(input, "rb") as file:
+                data = file.read()
+        else:
+            raise TypeError(
+                f"create() expected argument type 'io.BufferedReader', 'bytes', or 'str', but got '{type(input).__name__}'"
+            )
+
         path = f"v1/content/{self.content_guid}/bundles"
         url = urls.append(self.config.url, path)
         response = self.session.post(url, data=data)

tests/posit/connect/test_bundles.py

@@ -1,7 +1,10 @@
+import io
+
+import pytest
 import requests
 import responses
 
-from responses import matchers
+from unittest import mock
 
 from posit.connect import Client
 from posit.connect.config import Config
@@ -122,8 +125,9 @@ def test(self):
 
 
 class TestBundleDownload:
+    @mock.patch("builtins.open", new_callable=mock.mock_open)
     @responses.activate
-    def test(self):
+    def test_output_as_str(self, mock_file: mock.MagicMock):
         content_guid = "f2f37341-e21d-3d80-c698-a935ad614066"
         bundle_id = "101"
         path = get_path(
@@ -153,21 +157,101 @@ def test(self):
         bundle = c.content.get(content_guid).bundles.get(bundle_id)
 
         # invoke
-        data = bundle.download()
+        bundle.download("pathname")
 
         # assert
         assert mock_content_get.call_count == 1
         assert mock_bundle_get.call_count == 1
         assert mock_bundle_download.call_count == 1
-        assert data == path.read_bytes()
+        mock_file.assert_called_once_with("pathname", "wb")
+
+    @responses.activate
+    def test_output_as_io(self):
+        content_guid = "f2f37341-e21d-3d80-c698-a935ad614066"
+        bundle_id = "101"
+        path = get_path(
+            f"v1/content/{content_guid}/bundles/{bundle_id}/download/bundle.tar.gz"
+        )
+
+        # behavior
+        mock_content_get = responses.get(
+            f"https://connect.example/__api__/v1/content/{content_guid}",
+            json=load_mock(f"v1/content/{content_guid}.json"),
+        )
+
+        mock_bundle_get = responses.get(
+            f"https://connect.example/__api__/v1/content/{content_guid}/bundles/{bundle_id}",
+            json=load_mock(
+                f"v1/content/{content_guid}/bundles/{bundle_id}.json"
+            ),
+        )
+
+        mock_bundle_download = responses.get(
+            f"https://connect.example/__api__/v1/content/{content_guid}/bundles/{bundle_id}/download",
+            body=path.read_bytes(),
+        )
+
+        # setup
+        c = Client("12345", "https://connect.example")
+        bundle = c.content.get(content_guid).bundles.get(bundle_id)
+
+        # invoke
+        file = io.BytesIO()
+        buffer = io.BufferedWriter(file)
+        bundle.download(buffer)
+        buffer.seek(0)
+
+        # assert
+        assert mock_content_get.call_count == 1
+        assert mock_bundle_get.call_count == 1
+        assert mock_bundle_download.call_count == 1
+        assert file.read() == path.read_bytes()
+
+    @responses.activate
+    def test_invalid_arguments(self):
+        content_guid = "f2f37341-e21d-3d80-c698-a935ad614066"
+        bundle_id = "101"
+        path = get_path(
+            f"v1/content/{content_guid}/bundles/{bundle_id}/download/bundle.tar.gz"
+        )
+
+        # behavior
+        mock_content_get = responses.get(
+            f"https://connect.example/__api__/v1/content/{content_guid}",
+            json=load_mock(f"v1/content/{content_guid}.json"),
+        )
+
+        mock_bundle_get = responses.get(
+            f"https://connect.example/__api__/v1/content/{content_guid}/bundles/{bundle_id}",
+            json=load_mock(
+                f"v1/content/{content_guid}/bundles/{bundle_id}.json"
+            ),
+        )
+
+        mock_bundle_download = responses.get(
+            f"https://connect.example/__api__/v1/content/{content_guid}/bundles/{bundle_id}/download",
+            body=path.read_bytes(),
+        )
+
+        # setup
+        c = Client("12345", "https://connect.example")
+        bundle = c.content.get(content_guid).bundles.get(bundle_id)
+
+        # invoke
+        with pytest.raises(TypeError):
+            bundle.download(None)
+
+        # assert
+        assert mock_content_get.call_count == 1
+        assert mock_bundle_get.call_count == 1
 
 
 class TestBundlesCreate:
     @responses.activate
     def test(self):
         content_guid = "f2f37341-e21d-3d80-c698-a935ad614066"
         bundle_id = "101"
-        path = get_path(
+        pathname = get_path(
             f"v1/content/{content_guid}/bundles/{bundle_id}/download/bundle.tar.gz"
         )
 
@@ -189,14 +273,66 @@ def test(self):
         content = c.content.get(content_guid)
 
         # invoke
-        data = path.read_bytes()
+        data = pathname.read_bytes()
         bundle = content.bundles.create(data)
 
         # # assert
         assert bundle.id == "101"
         assert mock_content_get.call_count == 1
         assert mock_bundle_post.call_count == 1
 
+    @responses.activate
+    def test_kwargs_pathname(self):
+        content_guid = "f2f37341-e21d-3d80-c698-a935ad614066"
+        bundle_id = "101"
+        pathname = get_path(
+            f"v1/content/{content_guid}/bundles/{bundle_id}/download/bundle.tar.gz"
+        )
+
+        # behavior
+        mock_content_get = responses.get(
+            f"https://connect.example/__api__/v1/content/{content_guid}",
+            json=load_mock(f"v1/content/{content_guid}.json"),
+        )
+
+        mock_bundle_post = responses.post(
+            f"https://connect.example/__api__/v1/content/{content_guid}/bundles",
+            json=load_mock(
+                f"v1/content/{content_guid}/bundles/{bundle_id}.json"
+            ),
+        )
+
+        # setup
+        c = Client("12345", "https://connect.example")
+        content = c.content.get(content_guid)
+
+        # invoke
+        pathname = str(pathname.absolute())
+        bundle = content.bundles.create(pathname)
+
+        # # assert
+        assert bundle.id == "101"
+        assert mock_content_get.call_count == 1
+        assert mock_bundle_post.call_count == 1
+
+    @responses.activate
+    def test_invalid_arguments(self):
+        content_guid = "f2f37341-e21d-3d80-c698-a935ad614066"
+
+        # behavior
+        responses.get(
+            f"https://connect.example/__api__/v1/content/{content_guid}",
+            json=load_mock(f"v1/content/{content_guid}.json"),
+        )
+
+        # setup
+        c = Client("12345", "https://connect.example")
+        content = c.content.get(content_guid)
+
+        # invoke
+        with pytest.raises(TypeError):
+            content.bundles.create(None)
+
 
 class TestBundlesFind:
     @responses.activate