feat: add method to retrieve active bundle (#404)

Adds `active()` method to `Bundles` class to retrieve the currently
active bundle.

Resolves #380

Author: tdstein

integration/tests/posit/connect/test_bundles.py

@@ -0,0 +1,99 @@
+import time
+from pathlib import Path
+
+import pytest
+from packaging import version
+
+from posit import connect
+
+from . import CONNECT_VERSION
+
+
+class TestBundles:
+    @classmethod
+    def setup_class(cls):
+        cls.client = connect.Client()
+        cls.content = cls.client.content.create(
+            name=f"test-bundles-{int(time.time())}",
+            title="Test Bundles",
+            access_type="all",
+        )
+        # Path to the test bundle
+        bundle_path = Path(
+            "../../../resources/connect/bundles/example-flask-minimal/bundle.tar.gz"
+        )
+        cls.bundle_path = (Path(__file__).parent / bundle_path).resolve()
+
+    @classmethod
+    def teardown_class(cls):
+        cls.content.delete()
+
+    def test_create_bundle(self):
+        """Test creating a bundle."""
+        bundle = self.content.bundles.create(str(self.bundle_path))
+        assert bundle["id"]
+        assert bundle["content_guid"] == self.content["guid"]
+
+    def test_find_bundles(self):
+        """Test finding all bundles."""
+        # Create a bundle first
+        self.content.bundles.create(str(self.bundle_path))
+
+        # Find all bundles
+        bundles = self.content.bundles.find()
+        assert len(bundles) >= 1
+
+    def test_find_one_bundle(self):
+        """Test finding a single bundle."""
+        # Create a bundle first
+        self.content.bundles.create(str(self.bundle_path))
+
+        # Find one bundle
+        bundle = self.content.bundles.find_one()
+        assert bundle is not None
+        assert bundle["content_guid"] == self.content["guid"]
+
+    def test_get_bundle(self):
+        """Test getting a specific bundle."""
+        # Create a bundle first
+        created_bundle = self.content.bundles.create(str(self.bundle_path))
+
+        # Get the bundle by ID
+        bundle = self.content.bundles.get(created_bundle["id"])
+        assert bundle["id"] == created_bundle["id"]
+
+    @pytest.mark.skipif(
+        CONNECT_VERSION < version.parse("2025.02.0"), reason="Requires Connect 2025.02.0 or later"
+    )
+    def test_active_bundle(self):
+        """Test retrieving the active bundle."""
+        # Initially, no bundle should be active
+        assert self.content.bundles.active() is None
+
+        # Create and deploy a bundle
+        bundle = self.content.bundles.create(str(self.bundle_path))
+        task = bundle.deploy()
+        task.wait_for()
+
+        # Wait for the bundle to become active
+        max_retries = 10
+        active_bundle = None
+        for _ in range(max_retries):
+            active_bundle = self.content.bundles.active()
+            if active_bundle is not None:
+                break
+            time.sleep(1)
+
+        # Verify the bundle is now active
+        assert active_bundle is not None
+        assert active_bundle["id"] == bundle["id"]
+        assert active_bundle.get("active") is True
+
+        # Create another bundle but don't deploy it
+        bundle2 = self.content.bundles.create(str(self.bundle_path))
+
+        # Verify the active bundle is still the first one
+        active_bundle = self.content.bundles.active()
+        assert active_bundle is not None
+        assert active_bundle["id"] == bundle["id"]
+        assert active_bundle["id"] != bundle2["id"]

src/posit/connect/bundles.py

@@ -6,6 +6,8 @@
 
 from typing_extensions import TYPE_CHECKING, List
 
+from posit.connect.context import requires
+
 from . import resources, tasks
 
 if TYPE_CHECKING:
@@ -193,6 +195,24 @@ def find_one(self) -> Bundle | None:
         bundles = self.find()
         return next(iter(bundles), None)
 
+    @requires("2025.02.0")
+    def active(self) -> Bundle | None:
+        """Retrieve the active bundle.
+
+        Returns
+        -------
+        Bundle | None
+            The currently active bundle, or None if no active bundle exists.
+
+        Examples
+        --------
+        >>> client = connect.Client()
+        >>> content = client.content.get(guid)
+        >>> bundle = content.bundles.active()
+        """
+        bundles = self.find()
+        return next((bundle for bundle in bundles if bundle.get("active", False)), None)
+
     def get(self, uid: str) -> Bundle:
         """Get a bundle.
 

tests/posit/connect/test_bundles.py

@@ -373,3 +373,66 @@ def test(self):
         assert mock_content_get.call_count == 1
         assert mock_bundle_get.call_count == 1
         assert bundle["id"] == bundle_id
+
+
+class TestBundlesActive:
+    @responses.activate
+    def test_active_bundle_exists(self):
+        content_guid = "f2f37341-e21d-3d80-c698-a935ad614066"
+
+        # Create a modified copy of bundles.json with one bundle as active
+        bundles_data = load_mock(f"v1/content/{content_guid}/bundles.json")
+        active_bundles_data = bundles_data.copy()
+        active_bundles_data[0]["active"] = True
+
+        # 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_bundles_get = responses.get(
+            f"https://connect.example/__api__/v1/content/{content_guid}/bundles",
+            json=active_bundles_data,
+        )
+
+        # setup
+        c = Client("https://connect.example", "12345")
+        c._ctx.version = None
+
+        # invoke
+        bundle = c.content.get(content_guid).bundles.active()
+
+        # assert
+        assert mock_content_get.call_count == 1
+        assert mock_bundles_get.call_count == 1
+        assert bundle is not None
+        assert bundle["id"] == "101"
+        assert bundle["active"] is True
+
+    @responses.activate
+    def test_no_active_bundle(self):
+        content_guid = "f2f37341-e21d-3d80-c698-a935ad614066"
+
+        # 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_bundles_get = responses.get(
+            f"https://connect.example/__api__/v1/content/{content_guid}/bundles",
+            json=load_mock(f"v1/content/{content_guid}/bundles.json"),
+        )
+
+        # setup
+        c = Client("https://connect.example", "12345")
+        c._ctx.version = None
+
+        # invoke
+        bundle = c.content.get(content_guid).bundles.active()
+
+        # assert
+        assert mock_content_get.call_count == 1
+        assert mock_bundles_get.call_count == 1
+        assert bundle is None