Merge remote-tracking branch 'origin/develop' into feature/dependency_upgrades

Author: angelomelonas

bunq/sdk/model/core/bunq_model.py

@@ -3,6 +3,7 @@
 import typing
 from typing import Dict, List
 
+from bunq import T
 from bunq.sdk.http.bunq_response import BunqResponse
 from bunq.sdk.http.bunq_response_raw import BunqResponseRaw
 from bunq.sdk.json import converter
@@ -90,7 +91,7 @@ def _process_for_uuid(cls, response_raw: BunqResponseRaw) -> BunqResponse[str]:
     @classmethod
     def _from_json_list(cls,
                         response_raw: BunqResponseRaw,
-                        wrapper: str = None) -> BunqResponse[List[BunqModel]]:
+                        wrapper: str = None) -> BunqResponse[List[T]]:
         from bunq import Pagination
 
         json = response_raw.body_bytes.decode()
@@ -103,7 +104,10 @@ def _from_json_list(cls,
             item_deserialized = converter.deserialize(cls, item_unwrapped)
             array_deserialized.append(item_deserialized)
 
-        pagination = converter.deserialize(Pagination, obj[cls._FIELD_PAGINATION])
+        pagination = None
+
+        if cls._FIELD_PAGINATION in obj:
+            pagination = converter.deserialize(Pagination, obj[cls._FIELD_PAGINATION])
 
         return BunqResponse(array_deserialized, response_raw.headers, pagination)
 

bunq/sdk/model/core/notification_filter_push_user_internal.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+from typing import List, Dict
+
+from bunq.sdk.http.api_client import ApiClient
+from bunq.sdk.http.bunq_response import BunqResponse
+from bunq.sdk.json import converter
+from bunq.sdk.model.generated.endpoint import NotificationFilterPushUser
+from bunq.sdk.model.generated.object_ import NotificationFilterPush, NotificationFilterUrl
+
+
+class NotificationFilterPushUserInternal(NotificationFilterPushUser):
+    @classmethod
+    def create_with_list_response(cls,
+                                  all_notification_filter: List[NotificationFilterPush] = None,
+                                  custom_headers: Dict[str, str] = None
+                                  ) -> BunqResponse[List[NotificationFilterPush]]:
+        if all_notification_filter is None:
+            all_notification_filter = []
+
+        if custom_headers is None:
+            custom_headers = {}
+
+        request_map = {
+            cls.FIELD_NOTIFICATION_FILTERS: all_notification_filter
+        }
+        request_map_string = converter.class_to_json(request_map)
+        request_map_string = cls._remove_field_for_request(request_map_string)
+
+        api_client = ApiClient(cls._get_api_context())
+        request_bytes = request_map_string.encode()
+        endpoint_url = cls._ENDPOINT_URL_CREATE.format(cls._determine_user_id())
+        response_raw = api_client.post(endpoint_url, request_bytes, custom_headers)
+
+        return NotificationFilterUrl._from_json_list(response_raw, cls._OBJECT_TYPE_GET)

bunq/sdk/model/core/notification_filter_url_monetary_account_internal.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+from typing import List, Dict
+
+from bunq.sdk.http.api_client import ApiClient
+from bunq.sdk.http.bunq_response import BunqResponse
+from bunq.sdk.json import converter
+from bunq.sdk.model.generated.endpoint import NotificationFilterUrlMonetaryAccount
+from bunq.sdk.model.generated.object_ import NotificationFilterUrl
+
+
+class NotificationFilterUrlMonetaryAccountInternal(NotificationFilterUrlMonetaryAccount):
+    @classmethod
+    def create_with_list_response(cls,
+                                  monetary_account_id: int = None,
+                                  all_notification_filter: List[NotificationFilterUrl] = None,
+                                  custom_headers: Dict[str, str] = None
+                                  ) -> BunqResponse[List[NotificationFilterUrl]]:
+        if all_notification_filter is None:
+            all_notification_filter = []
+
+        if custom_headers is None:
+            custom_headers = {}
+
+        request_map = {
+            cls.FIELD_NOTIFICATION_FILTERS: all_notification_filter
+        }
+        request_map_string = converter.class_to_json(request_map)
+        request_map_string = cls._remove_field_for_request(request_map_string)
+
+        api_client = ApiClient(cls._get_api_context())
+        request_bytes = request_map_string.encode()
+        endpoint_url = cls._ENDPOINT_URL_CREATE.format(cls._determine_user_id(),
+                                                       cls._determine_monetary_account_id(monetary_account_id))
+        response_raw = api_client.post(endpoint_url, request_bytes, custom_headers)
+
+        return NotificationFilterUrl._from_json_list(response_raw, cls._OBJECT_TYPE_GET)

bunq/sdk/model/core/notification_filter_url_user_internal.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+from typing import List, Dict
+
+from bunq.sdk.http.api_client import ApiClient
+from bunq.sdk.http.bunq_response import BunqResponse
+from bunq.sdk.json import converter
+from bunq.sdk.model.generated.endpoint import NotificationFilterUrlUser
+from bunq.sdk.model.generated.object_ import NotificationFilterUrl
+
+
+class NotificationFilterUrlUserInternal(NotificationFilterUrlUser):
+    @classmethod
+    def create_with_list_response(cls,
+                                  all_notification_filter: List[NotificationFilterUrl] = None,
+                                  custom_headers: Dict[str, str] = None
+                                  ) -> BunqResponse[List[NotificationFilterUrl]]:
+        if all_notification_filter is None:
+            all_notification_filter = []
+
+        if custom_headers is None:
+            custom_headers = {}
+
+        request_map = {
+            cls.FIELD_NOTIFICATION_FILTERS: all_notification_filter
+        }
+        request_map_string = converter.class_to_json(request_map)
+        request_map_string = cls._remove_field_for_request(request_map_string)
+
+        api_client = ApiClient(cls._get_api_context())
+        request_bytes = request_map_string.encode()
+        endpoint_url = cls._ENDPOINT_URL_CREATE.format(cls._determine_user_id())
+        response_raw = api_client.post(endpoint_url, request_bytes, custom_headers)
+
+        return NotificationFilterUrl._from_json_list(response_raw, cls._OBJECT_TYPE_GET)

bunq/sdk/security/security.py

@@ -187,15 +187,14 @@ def validate_response(public_key_server: RsaKey,
 
 
 def is_valid_response_header_with_body(public_key_server: RsaKey,
-                      status_code: int,
-                      body_bytes: bytes,
-                      headers: Dict[str, str]) -> bool:
+                                       status_code: int,
+                                       body_bytes: bytes,
+                                       headers: Dict[str, str]) -> bool:
     head_bytes = _generate_response_head_bytes(status_code, headers)
     bytes_signed = head_bytes + body_bytes
     signer = PKCS1_v1_5.pkcs1_15.new(public_key_server)
     digest = SHA256.new()
     digest.update(bytes_signed)
-    signer.verify(digest, base64.b64decode(headers[_HEADER_SERVER_SIGNATURE]))
 
     try:
         signer.verify(digest, base64.b64decode(headers[_HEADER_SERVER_SIGNATURE]))

tests/model/core/test_notification_filter.py

@@ -0,0 +1,69 @@
+from bunq.sdk.context.bunq_context import BunqContext
+from bunq.sdk.model.core.notification_filter_push_user_internal import NotificationFilterPushUserInternal
+from bunq.sdk.model.core.notification_filter_url_monetary_account_internal import \
+    NotificationFilterUrlMonetaryAccountInternal
+from bunq.sdk.model.core.notification_filter_url_user_internal import NotificationFilterUrlUserInternal
+from bunq.sdk.model.generated.object_ import NotificationFilterUrl, NotificationFilterPush
+from tests.bunq_test import BunqSdkTestCase
+
+
+class TestNotificationFilter(BunqSdkTestCase):
+    _FILTER_CATEGORY_MUTATION = 'MUTATION'
+    _FILTER_CALLBACK_URL = 'https://test.com/callback'
+
+    def test_notification_filter_url_monetary_account(self):
+        notification_filter = self.get_notification_filter_url()
+        all_notification_filter = [notification_filter]
+
+        all_created_notification_filter = NotificationFilterUrlMonetaryAccountInternal.create_with_list_response(
+            self.get_primary_monetary_account().id_,
+            all_notification_filter
+        ).value
+
+        self.assertEqual(1, len(all_created_notification_filter))
+
+    def test_notification_filter_url_user(self):
+        notification_filter = self.get_notification_filter_url()
+        all_notification_filter = [notification_filter]
+
+        all_created_notification_filter = NotificationFilterUrlUserInternal.create_with_list_response(
+            all_notification_filter
+        ).value
+
+        self.assertEqual(1, len(all_created_notification_filter))
+
+    def test_notification_filter_push_user(self):
+        notification_filter = self.get_notification_filter_push()
+        all_notification_filter = [notification_filter]
+
+        all_created_notification_filter = NotificationFilterPushUserInternal.create_with_list_response(
+            all_notification_filter
+        ).value
+
+        self.assertEqual(1, len(all_created_notification_filter))
+
+    def test_notification_filter_clear(self):
+        all_created_notification_filter_push_user = \
+            NotificationFilterPushUserInternal.create_with_list_response().value
+        all_created_notification_filter_url_user = \
+            NotificationFilterUrlUserInternal.create_with_list_response().value
+        all_created_notification_filter_url_monetary_account = \
+            NotificationFilterUrlMonetaryAccountInternal.create_with_list_response().value
+
+        self.assertFalse(all_created_notification_filter_push_user)
+        self.assertFalse(all_created_notification_filter_url_user)
+        self.assertFalse(all_created_notification_filter_url_monetary_account)
+
+        self.assertEqual(0, len(NotificationFilterPushUserInternal.list().value))
+        self.assertEqual(0, len(NotificationFilterUrlUserInternal.list().value))
+        self.assertEqual(0, len(NotificationFilterUrlMonetaryAccountInternal.list().value))
+
+    def get_notification_filter_url(self):
+        return NotificationFilterUrl(self._FILTER_CATEGORY_MUTATION, self._FILTER_CALLBACK_URL)
+
+    def get_notification_filter_push(self):
+        return NotificationFilterPush(self._FILTER_CATEGORY_MUTATION)
+
+    @staticmethod
+    def get_primary_monetary_account():
+        return BunqContext.user_context().primary_monetary_account