From 72d772289519347bdcefa12faa2807e4e54e84fd Mon Sep 17 00:00:00 2001 From: axshani Date: Sat, 27 Nov 2021 21:35:18 +0200 Subject: [PATCH] implementation of API Token with tests --- e2e_tests/api_token_test.py | 34 ++++++++++++++++++++++++ unit/__init__.py | 2 ++ unit/api/api_token_resource.py | 34 ++++++++++++++++++++++++ unit/models/api_token.py | 48 ++++++++++++++++++++++++++++++++++ unit/models/codecs.py | 3 +++ 5 files changed, 121 insertions(+) create mode 100644 e2e_tests/api_token_test.py create mode 100644 unit/api/api_token_resource.py create mode 100644 unit/models/api_token.py diff --git a/e2e_tests/api_token_test.py b/e2e_tests/api_token_test.py new file mode 100644 index 00000000..ed923ef5 --- /dev/null +++ b/e2e_tests/api_token_test.py @@ -0,0 +1,34 @@ +import os +import unittest +from unit import Unit +from unit.models.api_token import CreateAPITokenRequest + + +class APITokenE2eTests(unittest.TestCase): + token = os.environ.get("token") + client = Unit("https://api.s.unit.sh", token) + user_id = "252" + + def create_api_token(self): + request = CreateAPITokenRequest(self.user_id, "Test token", "customers applications", "2022-07-01T13:47:17.000Z") + return self.client.api_tokens.create(request).data + + def test_list_api_tokens(self): + api_tokens_ids = [] + response = self.client.api_tokens.list(self.user_id) + + for t in response.data: + self.assertTrue(t.type == "apiToken") + + def test_create_api_token(self): + token = self.create_api_token() + self.assertTrue(token.type == "apiToken") + + def test_delete_api_token(self): + token = self.create_api_token() + response = self.client.api_tokens.revoke(self.user_id, token.id) + self.assertTrue(response.data == []) + + +if __name__ == '__main__': + unittest.main() diff --git a/unit/__init__.py b/unit/__init__.py index ec46ad76..fa025c8f 100644 --- a/unit/__init__.py +++ b/unit/__init__.py @@ -13,6 +13,7 @@ from unit.api.event_resource import EventResource from unit.api.webhook_resource import WebhookResource from unit.api.institution_resource import InstitutionResource +from unit.api.api_token_resource import APITokenResource __all__ = ["api", "models", "utils"] @@ -33,3 +34,4 @@ def __init__(self, api_url, token): self.events = EventResource(api_url, token) self.webhooks = WebhookResource(api_url, token) self.institutions = InstitutionResource(api_url, token) + self.api_tokens = APITokenResource(api_url, token) diff --git a/unit/api/api_token_resource.py b/unit/api/api_token_resource.py new file mode 100644 index 00000000..b8fff379 --- /dev/null +++ b/unit/api/api_token_resource.py @@ -0,0 +1,34 @@ +from unit.api.base_resource import BaseResource +from unit.models.api_token import * +from unit.models.codecs import DtoDecoder + + +class APITokenResource(BaseResource): + def __init__(self, api_url, token): + super().__init__(api_url, token) + self.resource = "users" + + def create(self, request: CreateAPITokenRequest) -> Union[UnitResponse[APITokenDTO], UnitError]: + payload = request.to_json_api() + response = super().post(f"{self.resource}/{request.user_id}/api-tokens", payload) + if super().is_20x(response.status_code): + data = response.json().get("data") + return UnitResponse[APITokenDTO](DtoDecoder.decode(data), None) + else: + return UnitError.from_json_api(response.json()) + + def list(self, user_id: str) -> Union[UnitResponse[list[APITokenDTO]], UnitError]: + response = super().get(f"{self.resource}/{user_id}/api-tokens") + if super().is_20x(response.status_code): + data = response.json().get("data") + return UnitResponse[APITokenDTO](DtoDecoder.decode(data), None) + else: + return UnitError.from_json_api(response.json()) + + def revoke(self, user_id: str, token_id: str) -> Union[UnitResponse, UnitError]: + response = super().delete(f"{self.resource}/{user_id}/api-tokens/{token_id}") + if super().is_20x(response.status_code): + return UnitResponse([], None) + else: + return UnitError.from_json_api(response.json()) + diff --git a/unit/models/api_token.py b/unit/models/api_token.py new file mode 100644 index 00000000..80858ec4 --- /dev/null +++ b/unit/models/api_token.py @@ -0,0 +1,48 @@ +from unit.models import * +from unit.utils import date_utils + + +class APITokenDTO(object): + def __init__(self, id: str, created_at: datetime, description: str, expiration: datetime, token: Optional[str], + source_ip: Optional[str]): + self.id = id + self.type = "apiToken" + self.attributes = {"createdAt": created_at, "description": description, "expiration": expiration, + "token": token, "sourceIp": source_ip} + + @staticmethod + def from_json_api(_id, _type, attributes, relationships): + return APITokenDTO(_id, date_utils.to_datetime(attributes["createdAt"]), attributes["description"], + date_utils.to_datetime(attributes["expiration"]), attributes.get("token"), + attributes.get("sourceIp")) + + +class CreateAPITokenRequest(object): + def __init__(self, user_id: str, description: str, scope: str, expiration: datetime, + source_ip: Optional[str] = None): + self.user_id = user_id + self.description = description + self.scope = scope + self.expiration = expiration + self.source_ip = source_ip + + def to_json_api(self) -> dict: + payload = { + "data": { + "type": "apiToken", + "attributes": { + "description": self.description, + "scope": self.scope, + "expiration": self.expiration + } + } + } + + if self.source_ip: + payload["data"]["attributes"]["sourceIp"] = self.source_ip + + return payload + + def __repr__(self): + json.dumps(self.to_json_api()) + diff --git a/unit/models/codecs.py b/unit/models/codecs.py index c26319b1..7ee2ef8d 100644 --- a/unit/models/codecs.py +++ b/unit/models/codecs.py @@ -15,6 +15,7 @@ from unit.models.counterparty import CounterpartyDTO from unit.models.webhook import WebhookDTO from unit.models.institution import InstitutionDTO +from unit.models.api_token import APITokenDTO mappings = { "individualApplication": lambda _id, _type, attributes, relationships: @@ -199,6 +200,8 @@ "institution": lambda _id, _type, attributes, relationships: InstitutionDTO.from_json_api(_id, _type, attributes, relationships), + "apiToken": lambda _id, _type, attributes, relationships: + APITokenDTO.from_json_api(_id, _type, attributes, relationships), }