[PT-5013] Move webhooks out of base (#617)

* Move webhooks related code from base.py

* Bump version

CHANGELOG.md

@@ -29,6 +29,12 @@ You can check your current version with the following command:
     ```
 
 For more information, see [UP42 Python package description](https://pypi.org/project/up42-py/).
+## 1.0.4a5
+
+**May 30, 2024**
+
+- Move webhooks related code from `base.py`
+
 ## 1.0.4a4
 
 **May 29, 2024**

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "up42-py"
-version = "1.0.4a3"
+version = "1.0.4a5"
 description = "Python SDK for UP42, the geospatial marketplace and developer platform."
 authors = ["UP42 GmbH <[email protected]>"]
 license = "https://github.com/up42/up42-py/blob/master/LICENSE"

tests/test_base.py

@@ -25,53 +25,14 @@ def test_should_authenticate(self, requests_mock):
         base.workspace.authenticate(username=constants.USER_EMAIL, password=constants.PASSWORD)
         assert base.workspace.id == constants.WORKSPACE_ID
 
-
-# TODO: these tests to be moved to test_initialization
-class TestNonWorkspace:
-    @pytest.fixture(autouse=True)
-    def workspace(self, auth_mock):
-        with mock.patch("up42.base.workspace") as workspace_mock:
-            workspace_mock.auth = auth_mock
-            workspace_mock.id = constants.WORKSPACE_ID
-            yield
-
-    @mock.patch("up42.webhooks.Webhooks")
-    def test_should_get_webhook_events(self, webhooks: mock.MagicMock, auth_mock):
-        events = mock.MagicMock()
-        webhooks().get_webhook_events.return_value = events
-        assert base.get_webhook_events() == events
-        webhooks.assert_called_with(auth=auth_mock, workspace_id=constants.WORKSPACE_ID)
-
-    @mock.patch("up42.webhooks.Webhooks")
-    @pytest.mark.parametrize("return_json", [False, True])
-    def test_should_get_webhooks(self, webhooks: mock.MagicMock, auth_mock, return_json):
-        hooks = mock.MagicMock()
-        webhooks().get_webhooks.return_value = hooks
-        assert base.get_webhooks(return_json=return_json) == hooks
-        webhooks.assert_called_with(auth=auth_mock, workspace_id=constants.WORKSPACE_ID)
-        webhooks.return_value.get_webhooks.assert_called_with(return_json=return_json)
-
-    @mock.patch("up42.webhooks.Webhooks")
-    def test_should_create_webhook(self, webhooks: mock.MagicMock, auth_mock):
-        name = "name"
-        url = "url"
-        events = ["event"]
-        active = True
-        secret = "secret"
-        webhook = mock.MagicMock()
-        webhooks().create_webhook.return_value = webhook
-        assert webhook == base.create_webhook(name, url, events, active, secret)
-        webhooks.assert_called_with(auth=auth_mock, workspace_id=constants.WORKSPACE_ID)
-        webhooks().create_webhook.assert_called_with(name=name, url=url, events=events, active=active, secret=secret)
-
     def test_should_get_credits_balance(self, requests_mock):
         balance_url = f"{constants.API_HOST}/accounts/me/credits/balance"
         balance = {"balance": 10693}
         requests_mock.get(
             url=balance_url,
             json={"data": balance},
         )
-        assert base.get_credits_balance() == balance
+        assert base.workspace.get_credits_balance() == balance
 
 
 @dataclasses.dataclass(eq=True)

tests/test_initialization.py

@@ -1,30 +1,68 @@
 from unittest import mock
 
+import pytest
+
 import up42
 from up42 import catalog, storage, tasking
 
 from .fixtures import fixtures_globals as constants
 
 
+@pytest.fixture(autouse=True)
+def workspace(auth_mock):
+    with mock.patch("up42.base.workspace") as workspace_mock:
+        workspace_mock.auth = auth_mock
+        workspace_mock.id = constants.WORKSPACE_ID
+        yield
+
+
+@pytest.fixture(name="webhooks")
+def _webhooks(auth_mock):
+    with mock.patch("up42.webhooks.Webhooks") as webhooks_class_mock:
+        yield webhooks_class_mock.return_value
+    webhooks_class_mock.assert_called_with(auth=auth_mock, workspace_id=constants.WORKSPACE_ID)
+
+
 def test_should_initialize_objects(
-    auth_mock,
     order_mock,
     asset_mock,
 ):
-    with mock.patch("up42.base.workspace") as workspace_mock:
-        workspace_mock.id = constants.WORKSPACE_ID
-        workspace_mock.auth = auth_mock
+    catalog_obj = up42.initialize_catalog()
+    assert isinstance(catalog_obj, catalog.Catalog)
+
+    storage_obj = up42.initialize_storage()
+    assert isinstance(storage_obj, storage.Storage)
+    assert storage_obj.workspace_id == constants.WORKSPACE_ID
+
+    order_obj = up42.initialize_order(order_id=constants.ORDER_ID)
+    assert order_obj.info == order_mock.info
+    asset_obj = up42.initialize_asset(asset_id=constants.ASSET_ID)
+    assert asset_obj.info == asset_mock.info
+    result = up42.initialize_tasking()
+    assert isinstance(result, tasking.Tasking)
+
+
+def test_should_get_webhook_events(webhooks: mock.MagicMock):
+    events = mock.MagicMock()
+    webhooks.get_webhook_events.return_value = events
+    assert up42.get_webhook_events() == events
+
 
-        catalog_obj = up42.initialize_catalog()
-        assert isinstance(catalog_obj, catalog.Catalog)
+@pytest.mark.parametrize("return_json", [False, True])
+def test_should_get_webhooks(webhooks: mock.MagicMock, return_json):
+    hooks = mock.MagicMock()
+    webhooks.get_webhooks.return_value = hooks
+    assert up42.get_webhooks(return_json=return_json) == hooks
+    webhooks.get_webhooks.assert_called_with(return_json=return_json)
 
-        storage_obj = up42.initialize_storage()
-        assert isinstance(storage_obj, storage.Storage)
-        assert storage_obj.workspace_id == constants.WORKSPACE_ID
 
-        order_obj = up42.initialize_order(order_id=constants.ORDER_ID)
-        assert order_obj.info == order_mock.info
-        asset_obj = up42.initialize_asset(asset_id=constants.ASSET_ID)
-        assert asset_obj.info == asset_mock.info
-        result = up42.initialize_tasking()
-        assert isinstance(result, tasking.Tasking)
+def test_should_create_webhook(webhooks: mock.MagicMock):
+    name = "name"
+    url = "url"
+    events = ["event"]
+    active = True
+    secret = "secret"
+    webhook = mock.MagicMock()
+    webhooks.create_webhook.return_value = webhook
+    assert webhook == up42.create_webhook(name, url, events, active, secret)
+    webhooks.create_webhook.assert_called_with(name=name, url=url, events=events, active=active, secret=secret)

up42/__init__.py

@@ -22,9 +22,12 @@
 # pylint: disable=only-importing-modules-is-allowed
 from up42.asset import Asset
 from up42.auth import Auth
-from up42.base import authenticate, create_webhook, get_credits_balance, get_webhook_events, get_webhooks
+from up42.base import authenticate, get_credits_balance
 from up42.catalog import Catalog
 from up42.initialization import (
+    create_webhook,
+    get_webhook_events,
+    get_webhooks,
     initialize_asset,
     initialize_catalog,
     initialize_order,

up42/base.py

@@ -1,12 +1,12 @@
 import logging
 import pathlib
 import warnings
-from typing import Any, List, Optional, Union
+from typing import Any, Optional, Union
 
 import requests
 
 from up42 import auth as up42_auth
-from up42 import host, utils, webhooks
+from up42 import host, utils
 
 logger = utils.get_logger(__name__, level=logging.INFO)
 
@@ -59,10 +59,22 @@ def authenticate(
         resp = self.auth.request("GET", url)
         self._id = resp["data"]["id"]
 
+    def get_credits_balance(self) -> dict:
+        """
+        Display the overall credits available in your account.
+
+        Returns:
+            A dict with the balance of credits available in your account.
+        """
+        endpoint_url = host.endpoint("/accounts/me/credits/balance")
+        response_json = self.auth.request(request_type="GET", url=endpoint_url)
+        return response_json["data"]
+
 
 workspace = _Workspace()
 
 authenticate = workspace.authenticate
+get_credits_balance = workspace.get_credits_balance
 
 
 class Session:
@@ -78,61 +90,3 @@ def __set__(self, obj, value: str) -> None:
         if value == self:
             value = workspace.id
         obj.__dict__["workspace_id"] = value
-
-
-def get_webhooks(return_json: bool = False) -> List[webhooks.Webhook]:
-    """
-    Gets all registered webhooks for this workspace.
-
-    Args:
-        return_json: If true returns the webhooks information as JSON instead of webhook class objects.
-    Returns:
-        A list of the registered webhooks for this workspace.
-    """
-    return webhooks.Webhooks(auth=workspace.auth, workspace_id=workspace.id).get_webhooks(return_json=return_json)
-
-
-def create_webhook(
-    name: str,
-    url: str,
-    events: List[str],
-    active: bool = False,
-    secret: Optional[str] = None,
-):
-    """
-    Registers a new webhook in the system.
-
-    Args:
-        name: Webhook name
-        url: Unique URL where the webhook will send the message (HTTPS required)
-        events: List of event types (order status / job task status)
-        active: Webhook status.
-        secret: String that acts as signature to the https request sent to the url.
-    Returns:
-        A dict with details of the registered webhook.
-    """
-    return webhooks.Webhooks(auth=workspace.auth, workspace_id=workspace.id).create_webhook(
-        name=name, url=url, events=events, active=active, secret=secret
-    )
-
-
-def get_webhook_events() -> dict:
-    """
-    Gets all available webhook events.
-
-    Returns:
-        A dict of the available webhook events.
-    """
-    return webhooks.Webhooks(auth=workspace.auth, workspace_id=workspace.id).get_webhook_events()
-
-
-def get_credits_balance() -> dict:
-    """
-    Display the overall credits available in your account.
-
-    Returns:
-        A dict with the balance of credits available in your account.
-    """
-    endpoint_url = host.endpoint("/accounts/me/credits/balance")
-    response_json = workspace.auth.request(request_type="GET", url=endpoint_url)
-    return response_json["data"]

up42/initialization.py

@@ -1,4 +1,5 @@
 import logging
+from typing import List, Optional
 
 from up42 import asset, base, catalog, order, storage, tasking, utils, webhooks
 
@@ -15,7 +16,7 @@ def initialize_catalog() -> catalog.Catalog:
     return catalog.Catalog(auth=base.workspace.auth, workspace_id=base.workspace.id)
 
 
-def initialize_tasking() -> "tasking.Tasking":
+def initialize_tasking() -> tasking.Tasking:
     """
     Returns a Tasking object for creating satellite tasking orders.
     """
@@ -60,3 +61,51 @@ def initialize_webhook(webhook_id: str) -> webhooks.Webhook:
     webhook = webhooks.Webhook(auth=base.workspace.auth, workspace_id=base.workspace.id, webhook_id=webhook_id)
     logger.info(INITIALIZED_MSG, webhook)
     return webhook
+
+
+def get_webhooks(return_json: bool = False) -> List[webhooks.Webhook]:
+    """
+    Gets all registered webhooks for this workspace.
+
+    Args:
+        return_json: If true returns the webhooks information as JSON instead of webhook class objects.
+    Returns:
+        A list of the registered webhooks for this workspace.
+    """
+    return webhooks.Webhooks(auth=base.workspace.auth, workspace_id=base.workspace.id).get_webhooks(
+        return_json=return_json
+    )
+
+
+def create_webhook(
+    name: str,
+    url: str,
+    events: List[str],
+    active: bool = False,
+    secret: Optional[str] = None,
+) -> webhooks.Webhook:
+    """
+    Registers a new webhook in the system.
+
+    Args:
+        name: Webhook name
+        url: Unique URL where the webhook will send the message (HTTPS required)
+        events: List of event types (order status / job task status)
+        active: Webhook status.
+        secret: String that acts as signature to the https request sent to the url.
+    Returns:
+        A dict with details of the registered webhook.
+    """
+    return webhooks.Webhooks(auth=base.workspace.auth, workspace_id=base.workspace.id).create_webhook(
+        name=name, url=url, events=events, active=active, secret=secret
+    )
+
+
+def get_webhook_events() -> dict:
+    """
+    Gets all available webhook events.
+
+    Returns:
+        A dict of the available webhook events.
+    """
+    return webhooks.Webhooks(auth=base.workspace.auth, workspace_id=base.workspace.id).get_webhook_events()