feat: TT-353 Create V2 Activities DAO

V2/source/activities_data.json

@@ -0,0 +1,66 @@
+[
+    {
+        "name": "Development",
+        "description": "Development",
+        "tenant_id": "cc925a5d-9644-4a4f-8d99-0bee49aadd05",
+        "id": "c61a4a49-3364-49a3-a7f7-0c5f2d15072b",
+        "_rid": "QUwFAPuumiRhAAAAAAAAAA==",
+        "_self": "dbs/QUwFAA==/colls/QUwFAPuumiQ=/docs/QUwFAPuumiRhAAAAAAAAAA==/",
+        "_etag": "\"4e006cc9-0000-0500-0000-607dcc0d0000\"",
+        "_attachments": "attachments/",
+        "_last_event_ctx": {
+            "user_id": "dd76e5d6-3949-46fd-b418-f15bf7c354fa",
+            "tenant_id": "cc925a5d-9644-4a4f-8d99-0bee49aadd05",
+            "action": "delete",
+            "description": null,
+            "container_id": "activity",
+            "session_id": null
+        },
+        "deleted": "b4327ba6-9f96-49ee-a9ac-3c1edf525172",
+        "status": null,
+        "_ts": 1618856973
+    },
+    {
+        "name": "Management",
+        "description": null,
+        "tenant_id": "cc925a5d-9644-4a4f-8d99-0bee49aadd05",
+        "id": "94ec92e2-a500-4700-a9f6-e41eb7b5507c",
+        "_last_event_ctx": {
+            "user_id": "dd76e5d6-3949-46fd-b418-f15bf7c354fa",
+            "tenant_id": "cc925a5d-9644-4a4f-8d99-0bee49aadd05",
+            "action": "delete",
+            "description": null,
+            "container_id": "activity",
+            "session_id": null
+        },
+        "_rid": "QUwFAPuumiRfAAAAAAAAAA==",
+        "_self": "dbs/QUwFAA==/colls/QUwFAPuumiQ=/docs/QUwFAPuumiRfAAAAAAAAAA==/",
+        "_etag": "\"4e0069c9-0000-0500-0000-607dcc0d0000\"",
+        "_attachments": "attachments/",
+        "deleted": "7cf6efe5-a221-4fe4-b94f-8945127a489a",
+        "status": null,
+        "_ts": 1618856973
+    },
+    {
+        "name": "Operations",
+        "description": "Operation activities performed.",
+        "tenant_id": "cc925a5d-9644-4a4f-8d99-0bee49aadd05",
+        "id": "d45c770a-b1a0-4bd8-a713-22c01a23e41b",
+        "_rid": "QUwFAPuumiRjAAAAAAAAAA==",
+        "_self": "dbs/QUwFAA==/colls/QUwFAPuumiQ=/docs/QUwFAPuumiRjAAAAAAAAAA==/",
+        "_etag": "\"09009a4d-0000-0500-0000-614b66fb0000\"",
+        "_attachments": "attachments/",
+        "_last_event_ctx": {
+            "user_id": "82ed0f65-051c-4898-890f-870805900e21",
+            "tenant_id": "cc925a5d-9644-4a4f-8d99-0bee49aadd05",
+            "action": "update",
+            "description": null,
+            "container_id": "activity",
+            "session_id": null
+        },
+        "deleted": "7cf6efe5-a221-4fe4-b94f-8945127a489a",
+        "status": "active",
+        "_ts": 1632331515
+    }
+]
+

V2/source/daos/activities_dao_interface.py

@@ -0,0 +1,12 @@
+from V2.source.dtos.activity import ActivityDto
+import abc
+
+
+class ActivitiesDaoInterface(abc.ABC):
+    @abc.abstractmethod
+    def get_by_id(self, id: str) -> ActivityDto:
+        pass
+
+    @abc.abstractmethod
+    def get_all(self) -> list:
+        pass

V2/source/daos/activities_json_dao.py

@@ -0,0 +1,54 @@
+from V2.source.daos.activities_dao_interface import ActivitiesDaoInterface
+from V2.source.dtos.activity import ActivityDto
+from http import HTTPStatus
+import json
+
+
+class ActivitiesJsonDao(ActivitiesDaoInterface):
+    def __init__(self, json_data_file_path: str):
+        self.json_data_file_path = json_data_file_path
+        self.activity_keys = ActivityDto.__dataclass_fields__.keys()
+
+    def get_by_id(self, id: str) -> ActivityDto:
+        activities = self.__get_activities_from_file()
+
+        try:
+            activity = next(
+                (
+                    activity
+                    for activity in activities
+                    if activity.get('id') == id
+                )
+            )
+        except Exception:
+            return HTTPStatus.NOT_FOUND
+
+        activity_dto = self.__create_activity_dto(activity)
+
+        return activity_dto
+
+    def get_all(self) -> list:
+        activities = self.__get_activities_from_file()
+        list_activities = []
+
+        for activity in activities:
+            activity = self.__create_activity_dto(activity)
+            list_activities.append(activity)
+
+        return list_activities
+
+    def __get_activities_from_file(self) -> list:
+        try:
+            file = open(self.json_data_file_path)
+            activities = json.load(file)
+            file.close()
+
+            return activities
+
+        except FileNotFoundError:
+            return []
+
+    def __create_activity_dto(self, activity: dict) -> ActivityDto:
+        activity = {key: activity.get(key) for key in self.activity_keys}
+        activity_dto = ActivityDto(**activity)
+        return activity_dto

V2/source/dtos/activity.py

@@ -0,0 +1,11 @@
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class ActivityDto:
+    id: str
+    name: str
+    description: str
+    deleted: str
+    status: str
+    tenant_id: str

V2/source/services/activity_service.py

@@ -0,0 +1,15 @@
+from V2.source.daos.activities_dao_interface import ActivitiesDaoInterface
+from V2.source.dtos.activity import ActivityDto
+
+
+class ActivityService:
+    def __init__(self, activities_dao: ActivitiesDaoInterface):
+        self.activities_dao = activities_dao
+
+    def get_by_id(self, id: str) -> ActivityDto:
+        activity_dto = self.activities_dao.get_by_id(id)
+        return activity_dto
+
+    def get_all(self) -> list:
+        list_activities = self.activities_dao.get_all()
+        return list_activities

V2/tests/daos/activities_json_dao_test.py

@@ -0,0 +1,78 @@
+from V2.source.daos.activities_json_dao import ActivitiesJsonDao
+from V2.source.dtos.activity import ActivityDto
+from http import HTTPStatus
+import json
+
+
+OPEN_FILE = 'builtins.open'
+
+
+def test_get_by_id__return_activity_dto__when_find_activity_that_matches_its_id(
+    mocker,
+):
+    activities_json_dao = ActivitiesJsonDao('non-important-path')
+    activities = [
+        {
+            "name": "test_name",
+            "description": "test_description",
+            "tenant_id": "test_tenant_id",
+            "id": "test_id",
+            "deleted": "test_deleted",
+            "status": "test_status",
+        }
+    ]
+    read_data = json.dumps(activities)
+    activity_dto = ActivityDto(**activities.pop())
+
+    mocker.patch(OPEN_FILE, mocker.mock_open(read_data=read_data))
+    result = activities_json_dao.get_by_id(activity_dto.id)
+
+    assert result == activity_dto
+
+
+def test__get_by_id__return_httpstatus_not_found__when_no_activity_matches_its_id(
+    mocker,
+):
+    activities_json_dao = ActivitiesJsonDao('non-important-path')
+    activities = []
+    read_data = json.dumps(activities)
+
+    mocker.patch(OPEN_FILE, mocker.mock_open(read_data=read_data))
+    result = activities_json_dao.get_by_id('non-important-id')
+
+    assert result == HTTPStatus.NOT_FOUND
+
+
+def test_get_all__return_list_of_activity_dto__when_find_one_or_more_activities(
+    mocker,
+):
+    activities_json_dao = ActivitiesJsonDao('non-important-path')
+    activity = {
+        "name": "test_name",
+        "description": "test_description",
+        "tenant_id": "test_tenant_id",
+        "id": "test_id",
+        "deleted": "test_deleted",
+        "status": "test_status",
+    }
+    number_of_activities = 3
+    activity_dto = ActivityDto(**activity)
+    list_activities_dto = [activity_dto] * number_of_activities
+    activities = [activity] * number_of_activities
+    read_data = json.dumps(activities)
+
+    mocker.patch(OPEN_FILE, mocker.mock_open(read_data=read_data))
+    result = activities_json_dao.get_all()
+
+    assert result == list_activities_dto
+
+
+def test_get_all__return_empty_list__when_doesnt_found_any_activities(mocker):
+    activities_json_dao = ActivitiesJsonDao('non-important-path')
+    activities = []
+    read_data = json.dumps(activities)
+
+    mocker.patch(OPEN_FILE, mocker.mock_open(read_data=read_data))
+    result = activities_json_dao.get_all()
+
+    assert result == activities

V2/tests/services/activity_service_test.py

@@ -0,0 +1,82 @@
+from V2.source.services.activity_service import ActivityService
+from V2.source.daos.activities_dao_interface import ActivitiesDaoInterface
+from V2.source.dtos.activity import ActivityDto
+from http import HTTPStatus
+
+
+def test__get_by_id__return_activity_dto__when_find_activity_that_matches_its_id(
+    mocker,
+):
+    activity_service = ActivityService(ActivitiesDaoInterface)
+    activity = {
+        "name": "test_name",
+        "description": "test_description",
+        "tenant_id": "test_tenant_id",
+        "id": "test_id",
+        "deleted": "test_deleted",
+        "status": "test_status",
+    }
+    activity_dto = ActivityDto(**activity)
+
+    activities_dao_mock = mocker.patch.object(
+        ActivitiesDaoInterface, 'get_by_id'
+    )
+    activities_dao_mock.return_value = activity_dto
+    result = activity_service.get_by_id(activity.get('id'))
+
+    assert result == activity_dto
+
+
+def test__get_by_id__return_httpstatus_not_found__when_no_activity_matches_its_id(
+    mocker,
+):
+    activity_service = ActivityService(ActivitiesDaoInterface)
+
+    activities_dao_mock = mocker.patch.object(
+        ActivitiesDaoInterface, 'get_by_id'
+    )
+    activities_dao_mock.return_value = HTTPStatus.NOT_FOUND
+
+    result = activity_service.get_by_id('non-important-id')
+
+    assert result == HTTPStatus.NOT_FOUND
+
+
+def test__get_all__return_list_of_activity_dto__when_find_one_or_more_activities(
+    mocker,
+):
+    activity_service = ActivityService(ActivitiesDaoInterface)
+    activity = {
+        "name": "test_name",
+        "description": "test_description",
+        "tenant_id": "test_tenant_id",
+        "id": "test_id",
+        "deleted": "test_deleted",
+        "status": "test_status",
+    }
+    number_of_activities = 3
+    activity_dto = ActivityDto(**activity)
+    list_activities_dto = [activity_dto] * number_of_activities
+
+    activities_dao_mock = mocker.patch.object(
+        ActivitiesDaoInterface, 'get_all'
+    )
+    activities_dao_mock.return_value = list_activities_dto
+    result = activity_service.get_all()
+
+    assert result == list_activities_dto
+
+
+def test__get_all__return_empty_list__when_doesnt_found_any_activities(
+    mocker,
+):
+    activity_service = ActivityService(ActivitiesDaoInterface)
+    activities = []
+
+    activities_dao_mock = mocker.patch.object(
+        ActivitiesDaoInterface, 'get_all'
+    )
+    activities_dao_mock.return_value = activities
+    result = activity_service.get_all()
+
+    assert result == activities