feat: TT-403 deleted time entries

Author: mandres2015

V2/pyvenv.cfg

@@ -0,0 +1,3 @@
+home = /usr/bin
+include-system-site-packages = false
+version = 3.8.10

V2/tests/api/azure/time_entry_azure_endpoints_test.py

@@ -1,15 +1,86 @@
-from time_tracker.time_entries._application import _time_entries as time_entries
+import pytest
+import json
+from faker import Faker
 
 import azure.functions as func
-import json
+
+from time_tracker.time_entries._application import _time_entries as time_entries
+import time_tracker.time_entries._application._time_entries as azure_time_entries
+import time_tracker.time_entries._infrastructure as infrastructure
+from time_tracker._infrastructure import DB
+from time_tracker.time_entries import _domain
+
 
 
 TIME_ENTRY_URL = "/api/time-entries/"
 
 
+@pytest.fixture(name='insert_time_entry')
+def _insert_time_entry() -> dict:
+    def _new_time_entry(time_entry: _domain.TimeEntry, database: DB):
+        dao = infrastructure.TimeEntriesSQLDao(database)
+        new_time_entry = dao.create(time_entry)
+        return new_time_entry.__dict__
+    return _new_time_entry
+
+
 def test__delete_time_entries_azure_endpoint__returns_an_time_entry_with_true_deleted__when_its_id_is_found(
-    create_temp_time_entries,
+    create_fake_database, time_entry_factory, insert_time_entry, create_temp_time_entries,
 ):
+    # fake_database = create_fake_database
+    # existent_activity = activity_factory()
+    # inserted_activity = insert_activity(existent_activity, fake_database)
+    # existent_time_entry = time_entry_factory()
+    # inserted_time_entry = insert_time_entry(existent_time_entry, fake_database)
+    azure_time_entries._.DATABASE = create_fake_database
+    time_entry_body = {
+        'id': 1,
+        'start_date': Faker().date(),
+        'owner_id': Faker().pyint(),
+        'description': Faker().sentence(),
+        'activity_id': 1,
+        'uri': Faker().uri(),
+        'technologies': None,
+        'end_date': Faker().date(),
+        'deleted': False,
+        'timezone_offset': None,
+        'project_id': 1,
+    }
+    body = json.dumps(time_entry_body).encode("utf-8")
+    req = func.HttpRequest(
+         method='DELETE',
+         body=body,
+         url=TIME_ENTRY_URL,
+        #  route_params={"id": 1},
+    )
+
+    response = azure_time_entries._delete_time_entry.delete_time_entry(req)
+    assert response.get_body() == ""
+    time_entry_json_data = json.loads(response.get_body().decode("utf-8"))
+    # activity_body['id'] = activitiy_json_data['id']
+
+    # azure_time_entries._delete_time_entry.DATABASE = fake_database
+    # req = func.HttpRequest(
+    #     method='DELETE',
+    #     body=None,
+    #     url=TIME_ENTRY_URL,
+    #     route_params={"id": inserted_time_entry["id"]},
+    # )
+
+    # assert req.get_body() == ""
+
+    # response = azure_activities._delete_activity.delete_activity(req)
+    # activity_json_data = json.loads(response.get_body().decode("utf-8"))
+
+    assert response.status_code == 200
+    # assert activity_json_data['status'] == 0
+    # assert activity_json_data['deleted'] is True
+
+
+
+
+
+
     time_entries_json, tmp_directory = create_temp_time_entries
     time_entries.delete_time_entry.JSON_PATH = tmp_directory
     req = func.HttpRequest(
@@ -19,6 +90,7 @@ def test__delete_time_entries_azure_endpoint__returns_an_time_entry_with_true_de
         route_params={"id": time_entries_json[0]["id"]},
     )
 
+
     response = time_entries.delete_time_entry(req)
     time_entry_json_data = json.loads(response.get_body().decode("utf-8"))
 

V2/tests/conftest.py

@@ -1,8 +1,5 @@
 # flake8: noqa
-<<<<<<< HEAD
 from fixtures import _activity_factory, _create_fake_dao, _create_fake_database
-=======
 from fixtures import create_temp_activities
 from fixtures import create_temp_time_entries
 from fixtures import _time_entry_factory
->>>>>>> 42cfc46... feat: TT-401 validated request create time entry

V2/tests/integration/daos/time_entries_dao.test.py

@@ -0,0 +1,47 @@
+import json
+import pytest
+import typing
+
+from faker import Faker
+
+from time_tracker.time_entries._domain import TimeEntry, TimeEntriesDao
+from time_tracker.time_entries._infrastructure import TimeEntriesSQLDao
+from time_tracker.activities._domain import Activity, ActivitiesDao
+
+
+@pytest.fixture(name='create_fake_time_entries')
+def _create_fake_time_entries() -> TimeEntry:
+    def _new_time_entry(time_entry: TimeEntry, dao: TimeEntriesDao):
+        new_time_entry = dao.create(time_entry)
+        return new_time_entry
+    return _new_time_entry
+
+
+def test_delete__returns_an_time_entry_with_true_deleted__when_an_time_entry_matching_its_id_is_found(
+    create_fake_dao, create_fake_time_entries, time_entry_factory, activity_factory, insert_activity
+):
+    dao = create_fake_dao
+
+    existent_time_entry = time_entry_factory()
+    inserted_time_entry = create_fake_time_entries(existent_time_entry, dao)
+
+
+    expected_description = Faker().sentence()
+    time_entries_json_dao = TimeEntriesSQLDao(Faker().file_path())
+    time_entries = create_fake_time_entries([time_entry_factory().__dict__])
+
+    time_entry_dto = time_entries[0]
+    result = time_entries_json_dao.delete(time_entry_dto.id)
+
+    assert result.deleted is True
+
+
+def test_delete__returns_none__when_no_time_entry_matching_its_id_is_found(
+    create_fake_time_entries,
+):
+    time_entries_json_dao = TimeEntriesSQLDao(Faker().file_path())
+    create_fake_time_entries([])
+
+    result = time_entries_json_dao.delete(Faker().pyint())
+
+    assert result is None

V2/tests/integration/daos/time_entries_dao_test.py

@@ -14,7 +14,7 @@ def __init__(self, conn_string: str = conn_string):
         self.engine = sqlalchemy.create_engine(conn_string)
 
     def get_session(self):
+        self.metadata.create_all(self.engine)
         if self.connection is None:
-            self.metadata.create_all(self.engine)
             self.connection = self.engine.connect()
         return self.connection

V2/time_tracker/_infrastructure/_db.py

@@ -4,14 +4,13 @@
 
 from ... import _domain
 from ... import _infrastructure
+from time_tracker._infrastructure import DB
 
-_JSON_PATH = (
-    'time_tracker/time_entries/_infrastructure/_data_persistence/time_entries_data.json'
-)
+DATABASE = DB()
 
 
 def delete_time_entry(req: func.HttpRequest) -> func.HttpResponse:
-    time_entry_dao = _infrastructure.TimeEntriesJsonDao(_JSON_PATH)
+    time_entry_dao = _infrastructure.TimeEntriesSQLDao(DATABASE)
     time_entry_service = _domain.TimeEntryService(time_entry_dao)
     use_case = _domain._use_cases.DeleteTimeEntryUseCase(time_entry_service)
 
@@ -21,20 +20,20 @@ def delete_time_entry(req: func.HttpRequest) -> func.HttpResponse:
         deleted_time_entry = use_case.delete_time_entry(time_entry_id)
         if not deleted_time_entry:
             return func.HttpResponse(
-              body='Not found',
-              status_code=404,
-              mimetype="application/json"
+                body="Not found", 
+                status_code=404, 
+                mimetype="application/json"
             )
 
         return func.HttpResponse(
-          body=json.dumps(deleted_time_entry.__dict__),
-          status_code=200,
-          mimetype="application/json"
+            body=json.dumps(deleted_time_entry.__dict__),
+            status_code=200,
+            mimetype="application/json",
         )
 
     except ValueError:
         return func.HttpResponse(
-            body=b'Invalid Format ID',
-            status_code=400,
+            body=b"Invalid Format ID", 
+            status_code=400, 
             mimetype="application/json"
         )

V2/time_tracker/time_entries/_application/_time_entries/_delete_time_entry.py

@@ -1,2 +1,2 @@
 # flake8: noqa
-from ._data_persistence import TimeEntriesJsonDao
+from ._data_persistence import TimeEntriesSQLDao

V2/time_tracker/time_entries/_infrastructure/__init__.py

@@ -1,2 +1,2 @@
 # flake8: noqa
-from ._time_entries_dao import TimeEntriesJsonDao
+from ._time_entries_dao import TimeEntriesSQLDao

V2/time_tracker/time_entries/_infrastructure/_data_persistence/__init__.py

@@ -1,58 +1,52 @@
 import json
 import dataclasses
 import typing
+import sqlalchemy
+import sqlalchemy.sql as sql
 
 from time_tracker.time_entries._domain import TimeEntriesDao, TimeEntry
-
-
-class TimeEntriesJsonDao(TimeEntriesDao):
-
-    def __init__(self, json_data_file_path: str):
-        self.json_data_file_path = json_data_file_path
-        self.time_entry_key = [field.name for field in dataclasses.fields(TimeEntry)]
-
-    def __get_time_entries_from_file(self) -> typing.List[dict]:
-        try:
-            file = open(self.json_data_file_path)
-            time_entries = json.load(file)
-            file.close()
-
-            return time_entries
-
-        except FileNotFoundError:
-            return []
+import time_tracker.time_entries._domain as domain
+from time_tracker._infrastructure import _db
+
+
+class TimeEntriesSQLDao(TimeEntriesDao):
+    def __init__(self, database: _db.DB):
+        self.time_entry_keys = [
+            field.name for field in dataclasses.fields(domain.TimeEntry)
+        ]
+        self.db = database
+        self.time_entry = sqlalchemy.Table(
+            "time_entry",
+            self.db.metadata,
+            sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True, autoincrement=True),
+            sqlalchemy.Column("start_date", sqlalchemy.DateTime),
+            sqlalchemy.Column("owner_id", sqlalchemy.Integer),
+            sqlalchemy.Column("description", sqlalchemy.String),
+            sqlalchemy.Column("activity_id", sqlalchemy.Integer, sqlalchemy.ForeignKey("activity.id")),
+            sqlalchemy.Column("uri", sqlalchemy.String),
+            sqlalchemy.Column("technologies", sqlalchemy.ARRAY(sqlalchemy.String).with_variant(sqlalchemy.String,"sqlite")),
+            sqlalchemy.Column("end_date", sqlalchemy.DateTime),
+            sqlalchemy.Column("deleted", sqlalchemy.Boolean),
+            sqlalchemy.Column("timezone_offset", sqlalchemy.String),
+            sqlalchemy.Column("project_id", sqlalchemy.Integer),
+            extend_existing=True,
+        )
 
     def __create_time_entry_dto(self, time_entry: dict) -> TimeEntry:
-        time_entry = {key: time_entry.get(key) for key in self.time_entry_key}
+        time_entry = {key: time_entry.get(key) for key in self.time_entry_keys}
         return TimeEntry(**time_entry)
 
-    def delete(self, time_entry_id: int) -> TimeEntry:
-        time_entry = {
-            time_entry.get('id'): time_entry
-            for time_entry in self.__get_time_entries_from_file()
-        }.get(int(time_entry_id))
-
-        if time_entry:
-            time_entry_deleted = {**time_entry, 'deleted': True}
-
-            time_entries_updated = list(
-                map(
-                    lambda time_entry: time_entry
-                    if time_entry.get('id') != time_entry_id
-                    else time_entry_deleted,
-                    self.__get_time_entries_from_file(),
-                )
-            )
+    def get_by_id(self, time_entry_id: int) -> domain.TimeEntry:
+        query = sql.select(self.time_entry).where(self.time_entry.c.id == time_entry_id)
+        time_entry = self.db.get_session().execute(query).one_or_none()
+        return self.__create_time_entry_dto(dict(time_entry)) if time_entry else None
 
-            try:
-                file = open(self.json_data_file_path, 'w')
-                json.dump(time_entries_updated, file)
-                file.close()
-
-                return self.__create_time_entry_dto(time_entry_deleted)
-
-            except FileNotFoundError:
-                return None
+    def delete(self, time_entry_id: int) -> TimeEntry:
 
-        else:
-            return None
+        query = (
+            self.time_entry.update()
+            .where(self.time_entry.c.id == time_entry_id)
+            .values({"deleted": True})
+        )
+        self.db.get_session().execute(query)
+        return self.get_by_id(time_entry_id)