[PT-5013] Webhook active record (#618)

* Move webhooks related code from base.py

* Bump version

* Webhook active record implementation

* Unify workspace ids

* Bump version

* Scouting

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.4a6
+
+**May 30, 2024**
+
+- Remodeled webhook as active record.
+
 ## 1.0.4a5
 
 **May 30, 2024**

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "up42-py"
-version = "1.0.4a5"
+version = "1.0.4a6"
 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

@@ -38,6 +38,7 @@ def test_should_get_credits_balance(self, requests_mock):
 @dataclasses.dataclass(eq=True)
 class ActiveRecord:
     session = base.Session()
+    class_workspace_id = base.WorkspaceId()
     workspace_id: Union[str, base.WorkspaceId] = dataclasses.field(default=base.WorkspaceId())
 
 
@@ -56,12 +57,15 @@ def test_should_provide_session(self):
     def test_session_should_not_be_represented(self):
         assert "session" not in repr(ActiveRecord())
 
-    def test_should_allow_to_set_workspace_id(self):
+    def test_workspace_id_should_provide_instance_value_when_invoked_for_object(self):
         record = ActiveRecord(workspace_id="custom_workspace_id")
         assert record.workspace_id == "custom_workspace_id"
+        assert record.class_workspace_id == "custom_workspace_id"
+        assert ActiveRecord.class_workspace_id == constants.WORKSPACE_ID
         assert "custom_workspace_id" in repr(record)
 
-    def test_should_provide_default_workspace_id(self):
+    def test_workspace_id_should_provide_default_value(self):
         record = ActiveRecord()
         assert record.workspace_id == constants.WORKSPACE_ID
+        assert ActiveRecord.class_workspace_id == constants.WORKSPACE_ID
         assert constants.WORKSPACE_ID in repr(record)

tests/test_webhooks.py

@@ -1,9 +1,12 @@
+import dataclasses
 import random
 import string
 import uuid
 from typing import List, Optional
+from unittest import mock
 
 import pytest
+import requests
 import requests_mock as req_mock
 
 from up42 import webhooks
@@ -15,6 +18,14 @@
 HOOK_URL = f"{HOOKS_URL}/{WEBHOOK_ID}"
 
 
+@pytest.fixture(autouse=True)
+def workspace():
+    with mock.patch("up42.base.workspace") as workspace_mock:
+        workspace_mock.auth.session = requests.session()
+        workspace_mock.id = constants.WORKSPACE_ID
+        yield
+
+
 def random_alphanumeric():
     return "".join(random.choices(string.ascii_letters + string.digits, k=10))
 
@@ -35,36 +46,35 @@ def random_metadata() -> dict:
 metadata = random_metadata()
 
 
-class TestWebhook:
-    def test_should_initialize(self, auth_mock):
-        webhook = webhooks.Webhook(auth_mock, constants.WORKSPACE_ID, WEBHOOK_ID, metadata)
-        assert webhook.auth == auth_mock
-        assert webhook.workspace_id == constants.WORKSPACE_ID
-        assert webhook.webhook_id == WEBHOOK_ID
-        assert repr(webhook) == repr(metadata)
+@pytest.fixture(name="webhook")
+def _webhook():
+    return webhooks.Webhook(
+        id=metadata["id"],
+        secret=metadata.get("secret"),
+        active=metadata["active"],
+        url=metadata["url"],
+        name=metadata["name"],
+        events=metadata["events"],
+        created_at=metadata.get("createdAt"),
+        updated_at=metadata.get("updatedAt"),
+    )
 
-    def test_should_initialize_existing_hook(self, auth_mock, requests_mock: req_mock.Mocker):
+
+class TestWebhook:
+    def test_should_get_webhook(self, requests_mock: req_mock.Mocker, webhook: webhooks.Webhook):
         requests_mock.get(url=HOOK_URL, json={"data": metadata})
-        webhook = webhooks.Webhook(auth_mock, constants.WORKSPACE_ID, WEBHOOK_ID)
-        assert webhook.auth == auth_mock
-        assert webhook.workspace_id == constants.WORKSPACE_ID
+        assert webhooks.Webhook.get(WEBHOOK_ID) == webhook
+
+    def test_should_provide_info(self, webhook):
+        info = metadata.copy()
+        info["created_at"] = info.pop("createdAt")
+        info["updated_at"] = info.pop("updatedAt")
+        assert webhook.info == info
+
+    def test_should_provide_webhook_id(self, webhook):
         assert webhook.webhook_id == WEBHOOK_ID
-        assert webhook.info == metadata
-        assert repr(webhook) == repr(metadata)
 
-    def test_should_refresh_info(
-        self,
-        auth_mock,
-        requests_mock: req_mock.Mocker,
-    ):
-        webhook = webhooks.Webhook(auth_mock, constants.WORKSPACE_ID, WEBHOOK_ID, metadata)
-        new_metadata = random_metadata()
-        requests_mock.get(url=HOOK_URL, json={"data": new_metadata})
-        assert webhook.info == new_metadata
-        assert repr(webhook) == repr(new_metadata)
-
-    def test_should_trigger_test_events(self, auth_mock, requests_mock: req_mock.Mocker):
-        webhook = webhooks.Webhook(auth_mock, constants.WORKSPACE_ID, WEBHOOK_ID, metadata)
+    def test_should_trigger_test_events(self, requests_mock: req_mock.Mocker, webhook: webhooks.Webhook):
         url = f"{HOOK_URL}/tests"
         test_event = {
             "startedAt": "2022-06-20T04:33:48.770826Z",
@@ -77,39 +87,85 @@ def test_should_trigger_test_events(self, auth_mock, requests_mock: req_mock.Moc
     @pytest.mark.parametrize("name", [random_alphanumeric(), None])
     @pytest.mark.parametrize("url", [random_alphanumeric(), None])
     @pytest.mark.parametrize("events", [[random_alphanumeric()], None])
-    @pytest.mark.parametrize("active", [random.choice([True, False]), None])
+    @pytest.mark.parametrize("active", [True, False, None])
     @pytest.mark.parametrize("secret", [random_alphanumeric(), None])
     def test_should_update(
         self,
-        auth_mock,
         requests_mock: req_mock.Mocker,
+        webhook,
         name: Optional[str],
         url: Optional[str],
         events: Optional[List[str]],
         active: Optional[bool],
         secret: Optional[str],
     ):
-        webhook = webhooks.Webhook(auth_mock, constants.WORKSPACE_ID, WEBHOOK_ID, metadata)
-        latest_metadata = random_metadata()
-        new_metadata = random_metadata()
-        requests_mock.get(url=HOOK_URL, json={"data": latest_metadata})
-        requests_mock.put(url=HOOK_URL, json={"data": new_metadata})
-        updated_webhook = webhook.update(name=name, url=url, active=active, events=events, secret=secret)
-        assert repr(updated_webhook) == repr(new_metadata)
-        assert requests_mock.last_request and requests_mock.last_request.json() == {
-            "name": name or latest_metadata["name"],
-            "url": url or latest_metadata["url"],
-            "events": events or latest_metadata["events"],
-            "active": active if active is not None else latest_metadata["active"],
-            "secret": secret or latest_metadata["secret"],
+        updated_at = "new-updated-at"
+        requests_mock.put(url=HOOK_URL, json={"data": {"updatedAt": updated_at}})
+        updates = {
+            "name": name or metadata["name"],
+            "url": url or metadata["url"],
+            "events": events or metadata["events"],
+            "active": active if active is not None else metadata["active"],
+            "secret": secret or metadata["secret"],
         }
 
-    def test_should_delete(self, auth_mock, requests_mock: req_mock.Mocker):
-        webhook = webhooks.Webhook(auth_mock, constants.WORKSPACE_ID, WEBHOOK_ID, metadata)
+        hook = dataclasses.replace(webhook)
+        assert (
+            hook.update(name=name, url=url, active=active, events=events, secret=secret)
+            == hook
+            == dataclasses.replace(
+                webhook,
+                name=updates["name"],
+                url=updates["url"],
+                active=updates["active"],
+                events=updates["events"],
+                secret=updates["secret"],
+                updated_at=updated_at,
+            )
+        )
+        assert requests_mock.last_request and requests_mock.last_request.json() == updates
+
+    def test_should_delete(self, requests_mock: req_mock.Mocker, webhook: webhooks.Webhook):
         requests_mock.delete(url=HOOK_URL)
         webhook.delete()
         assert requests_mock.called
 
+    def test_should_save_new(self, requests_mock: req_mock.Mocker, webhook: webhooks.Webhook):
+        hook = dataclasses.replace(webhook, id=None, created_at=None, updated_at=None)
+        requests_mock.post(
+            HOOKS_URL,
+            json={
+                "data": {
+                    "id": metadata["id"],
+                    "createdAt": metadata["createdAt"],
+                    "updatedAt": metadata["updatedAt"],
+                }
+            },
+        )
+        hook.save()
+        assert requests_mock.last_request and requests_mock.last_request.json() == {
+            "name": metadata["name"],
+            "url": metadata["url"],
+            "events": metadata["events"],
+            "active": metadata["active"],
+            "secret": metadata["secret"],
+        }
+
+    def test_should_save(self, requests_mock: req_mock.Mocker, webhook):
+        updates = {
+            "name": "new-name",
+            "url": "new-url",
+            "events": ["new-event"],
+            "secret": "new-secret",
+            "active": False,
+        }
+        hook = dataclasses.replace(webhook, **updates)
+        updated_at = "new-updated-at"
+        requests_mock.put(HOOK_URL, json={"data": {"updatedAt": updated_at}})
+        hook.save()
+        assert hook == dataclasses.replace(webhook, updated_at=updated_at, **updates)
+        assert requests_mock.last_request and requests_mock.last_request.json() == updates
+
 
 class TestWebhooks:
     @pytest.fixture(autouse=True)
@@ -128,36 +184,30 @@ def test_should_get_webhook_events(self, requests_mock: req_mock.Mocker):
         )
         assert self.webhooks.get_webhook_events() == events
 
-    def test_should_get_webhooks(self, requests_mock: req_mock.Mocker):
+    def test_should_get_webhooks(self, requests_mock: req_mock.Mocker, webhook: webhooks.Webhook):
         requests_mock.get(HOOKS_URL, json={"data": [metadata]})
-        (webhook,) = self.webhooks.get_webhooks()
-        assert isinstance(webhook, webhooks.Webhook)
-        assert webhook.auth == self.webhooks.auth
-        assert webhook.webhook_id == WEBHOOK_ID
-        assert webhook.workspace_id == constants.WORKSPACE_ID
-        assert repr(webhook) == repr(metadata)
+        assert self.webhooks.get_webhooks() == [webhook]
 
     def test_should_get_webhooks_as_dict(self, requests_mock: req_mock.Mocker):
-        hook = {**metadata, "id": WEBHOOK_ID}
-        requests_mock.get(HOOKS_URL, json={"data": [hook]})
-        assert self.webhooks.get_webhooks(return_json=True) == [hook]
+        requests_mock.get(HOOKS_URL, json={"data": [metadata]})
+        assert self.webhooks.get_webhooks(return_json=True) == [metadata]
 
     @pytest.mark.parametrize("secret", [random_alphanumeric(), None])
     def test_should_create_webhook(
         self,
         requests_mock: req_mock.Mocker,
+        webhook: webhooks.Webhook,
         secret: Optional[str],
     ):
         name = random_alphanumeric()
         url = random_alphanumeric()
         events = [random_alphanumeric()]
         active = random.choice([True, False])
         requests_mock.post(HOOKS_URL, json={"data": metadata})
-        webhook = self.webhooks.create_webhook(name=name, url=url, events=events, active=active, secret=secret)
-        assert webhook.auth == self.webhooks.auth
-        assert webhook.webhook_id == WEBHOOK_ID
-        assert webhook.workspace_id == constants.WORKSPACE_ID
-        assert repr(webhook) == repr(metadata)
+        assert self.webhooks.create_webhook(
+            name=name, url=url, events=events, active=active, secret=secret
+        ) == dataclasses.replace(webhook, name=name, url=url, active=active, events=events, secret=secret)
+
         assert requests_mock.last_request and requests_mock.last_request.json() == {
             "name": name,
             "url": url,

up42/base.py

@@ -84,7 +84,9 @@ def __get__(self, obj, obj_type=None) -> requests.Session:
 
 class WorkspaceId:
     def __get__(self, obj, obj_type=None) -> str:
-        return obj.__dict__.get("workspace_id", workspace.id)
+        if obj:
+            return obj.__dict__.get("workspace_id", workspace.id)
+        return workspace.id
 
     def __set__(self, obj, value: str) -> None:
         if value == self:

up42/initialization.py

@@ -5,7 +5,6 @@
 
 logger = utils.get_logger(__name__, level=logging.INFO)
 
-DEPRECATION_MESSAGE = "after May 15th, 2024, and will be replaced by new processing functionalities."
 INITIALIZED_MSG = "Initialized %s"
 
 
@@ -58,7 +57,7 @@ def initialize_webhook(webhook_id: str) -> webhooks.Webhook:
     Args:
         webhook_id: The UP42 webhook_id
     """
-    webhook = webhooks.Webhook(auth=base.workspace.auth, workspace_id=base.workspace.id, webhook_id=webhook_id)
+    webhook = webhooks.Webhook.get(webhook_id)
     logger.info(INITIALIZED_MSG, webhook)
     return webhook