feat: 🚧 add query param time-offset to summary endpoint

Author: Angeluz-07

commons/data_access_layer/cosmos_db.py

@@ -1,8 +1,6 @@
 import dataclasses
 import logging
-import uuid
-from datetime import datetime, timezone
-from typing import Callable, List, Dict
+from typing import Callable, List
 
 import azure.cosmos.cosmos_client as cosmos_client
 import azure.cosmos.exceptions as exceptions
@@ -380,62 +378,7 @@ def init_app(app: Flask) -> None:
     cosmos_helper = CosmosDBFacade.from_flask_config(app)
 
 
-def current_datetime() -> datetime:
-    return datetime.now(timezone.utc)
-
-
-def datetime_str(value: datetime) -> str:
-    if value is not None:
-        return value.isoformat()
-    else:
-        return None
-
-
-def current_datetime_str() -> str:
-    return datetime_str(current_datetime())
-
-
 def generate_uuid4() -> str:
-    return str(uuid.uuid4())
-
-
-def get_last_day_of_month(year: int, month: int) -> int:
-    from calendar import monthrange
-
-    return monthrange(year=year, month=month)[1]
-
-
-def get_current_year() -> int:
-    return datetime.now().year
-
-
-def get_current_month() -> int:
-    return datetime.now().month
-
-
-def get_current_day() -> int:
-    return datetime.now().day
-
-
-def get_date_range_of_month(year: int, month: int) -> Dict[str, str]:
-    first_day_of_month = 1
-    start_date = datetime(
-        year=year, month=month, day=first_day_of_month, tzinfo=timezone.utc
-    )
-
-    last_day_of_month = get_last_day_of_month(year=year, month=month)
-    end_date = datetime(
-        year=year,
-        month=month,
-        day=last_day_of_month,
-        hour=23,
-        minute=59,
-        second=59,
-        microsecond=999999,
-        tzinfo=timezone.utc,
-    )
+    from uuid import uuid4
 
-    return {
-        'start_date': datetime_str(start_date),
-        'end_date': datetime_str(end_date),
-    }
+    return str(uuid4())

tests/commons/data_access_layer/cosmos_db_test.py

@@ -1,24 +1,23 @@
 from dataclasses import dataclass
 from datetime import timedelta
 from typing import Callable
+from faker import Faker
 
 import pytest
+from pytest import fail
+
 from azure.cosmos.exceptions import (
     CosmosResourceExistsError,
     CosmosResourceNotFoundError,
 )
-from faker import Faker
-from flask_restplus._http import HTTPStatus
-from pytest import fail
 
+from commons.data_access_layer.database import EventContext
 from commons.data_access_layer.cosmos_db import (
     CosmosDBRepository,
     CosmosDBModel,
-    CustomError,
-    current_datetime,
-    datetime_str,
 )
-from commons.data_access_layer.database import EventContext
+
+from utils.time import datetime_str, current_datetime
 
 fake = Faker()
 Faker.seed()

tests/time_tracker_api/time_entries/time_entries_model_test.py

@@ -3,10 +3,13 @@
 import pytest
 from faker import Faker
 
-from commons.data_access_layer.cosmos_db import current_datetime, datetime_str
+from utils.time import datetime_str, current_datetime
 from commons.data_access_layer.database import EventContext
-from time_tracker_api.time_entries.time_entries_model import TimeEntryCosmosDBRepository, TimeEntryCosmosDBModel, \
-    container_definition
+from time_tracker_api.time_entries.time_entries_model import (
+    TimeEntryCosmosDBRepository,
+    TimeEntryCosmosDBModel,
+    container_definition,
+)
 
 fake = Faker()
 
@@ -15,42 +18,60 @@
 two_days_ago = current_datetime() - timedelta(days=2)
 
 
-def create_time_entry(start_date: datetime, end_date: datetime,
-                      owner_id: str, tenant_id: str, event_context: EventContext,
-                      time_entry_repository: TimeEntryCosmosDBRepository) -> TimeEntryCosmosDBModel:
+def create_time_entry(
+    start_date: datetime,
+    end_date: datetime,
+    owner_id: str,
+    tenant_id: str,
+    event_context: EventContext,
+    time_entry_repository: TimeEntryCosmosDBRepository,
+) -> TimeEntryCosmosDBModel:
     data = {
         "project_id": fake.uuid4(),
         "activity_id": fake.uuid4(),
         "description": fake.paragraph(nb_sentences=2),
         "start_date": datetime_str(start_date),
         "end_date": datetime_str(end_date),
         "owner_id": owner_id,
-        "tenant_id": tenant_id
+        "tenant_id": tenant_id,
     }
 
-    created_item = time_entry_repository.create(data, event_context,
-                                                mapper=TimeEntryCosmosDBModel)
+    created_item = time_entry_repository.create(
+        data, event_context, mapper=TimeEntryCosmosDBModel
+    )
     return created_item
 
 
 @pytest.mark.parametrize(
     'start_date,end_date', [(two_days_ago, yesterday), (now, None)]
 )
-def test_find_interception_with_date_range_should_find(start_date: datetime,
-                                                       end_date: datetime,
-                                                       owner_id: str,
-                                                       tenant_id: str,
-                                                       time_entry_repository: TimeEntryCosmosDBRepository):
-    event_ctx = EventContext(container_definition["id"], "create",
-                             tenant_id=tenant_id, user_id=owner_id)
-
-    existing_item = create_time_entry(start_date, end_date, owner_id, tenant_id,
-                                      event_ctx, time_entry_repository)
+def test_find_interception_with_date_range_should_find(
+    start_date: datetime,
+    end_date: datetime,
+    owner_id: str,
+    tenant_id: str,
+    time_entry_repository: TimeEntryCosmosDBRepository,
+):
+    event_ctx = EventContext(
+        container_definition["id"],
+        "create",
+        tenant_id=tenant_id,
+        user_id=owner_id,
+    )
+
+    existing_item = create_time_entry(
+        start_date,
+        end_date,
+        owner_id,
+        tenant_id,
+        event_ctx,
+        time_entry_repository,
+    )
 
     try:
-        result = time_entry_repository.find_interception_with_date_range(datetime_str(yesterday),
-                                                                         datetime_str(now),
-                                                                         owner_id, tenant_id)
+        result = time_entry_repository.find_interception_with_date_range(
+            datetime_str(yesterday), datetime_str(now), owner_id, tenant_id
+        )
 
         assert result is not None
         assert len(result) > 0
@@ -59,42 +80,63 @@ def test_find_interception_with_date_range_should_find(start_date: datetime,
         time_entry_repository.delete_permanently(existing_item.id, event_ctx)
 
 
-def test_find_interception_should_ignore_id_of_existing_item(owner_id: str, tenant_id: str,
-                                                             time_entry_repository: TimeEntryCosmosDBRepository):
-    event_ctx = EventContext(container_definition["id"], "create", tenant_id=tenant_id, user_id=owner_id)
+def test_find_interception_should_ignore_id_of_existing_item(
+    owner_id: str,
+    tenant_id: str,
+    time_entry_repository: TimeEntryCosmosDBRepository,
+):
+    event_ctx = EventContext(
+        container_definition["id"],
+        "create",
+        tenant_id=tenant_id,
+        user_id=owner_id,
+    )
     start_date = datetime_str(yesterday)
     end_date = datetime_str(now)
-    existing_item = create_time_entry(yesterday, now, owner_id, tenant_id, event_ctx, time_entry_repository)
+    existing_item = create_time_entry(
+        yesterday, now, owner_id, tenant_id, event_ctx, time_entry_repository
+    )
     try:
-        colliding_result = time_entry_repository.find_interception_with_date_range(start_date, end_date,
-                                                                                   owner_id, tenant_id)
-
-        non_colliding_result = time_entry_repository.find_interception_with_date_range(start_date, end_date,
-                                                                                       owner_id, tenant_id,
-                                                                                       ignore_id=existing_item.id)
+        colliding_result = time_entry_repository.find_interception_with_date_range(
+            start_date, end_date, owner_id, tenant_id
+        )
+
+        non_colliding_result = time_entry_repository.find_interception_with_date_range(
+            start_date,
+            end_date,
+            owner_id,
+            tenant_id,
+            ignore_id=existing_item.id,
+        )
 
         colliding_result is not None
         assert any([existing_item.id == item.id for item in colliding_result])
 
         non_colliding_result is not None
-        assert not any([existing_item.id == item.id for item in non_colliding_result])
+        assert not any(
+            [existing_item.id == item.id for item in non_colliding_result]
+        )
     finally:
         time_entry_repository.delete_permanently(existing_item.id, event_ctx)
 
 
-def test_find_running_should_return_running_time_entry(running_time_entry,
-                                                       time_entry_repository: TimeEntryCosmosDBRepository):
-    found_time_entry = time_entry_repository.find_running(running_time_entry.tenant_id,
-                                                          running_time_entry.owner_id)
+def test_find_running_should_return_running_time_entry(
+    running_time_entry, time_entry_repository: TimeEntryCosmosDBRepository
+):
+    found_time_entry = time_entry_repository.find_running(
+        running_time_entry.tenant_id, running_time_entry.owner_id
+    )
 
     assert found_time_entry is not None
     assert found_time_entry.id == running_time_entry.id
     assert found_time_entry.owner_id == running_time_entry.owner_id
 
 
-def test_find_running_should_not_find_any_item(tenant_id: str,
-                                               owner_id: str,
-                                               time_entry_repository: TimeEntryCosmosDBRepository):
+def test_find_running_should_not_find_any_item(
+    tenant_id: str,
+    owner_id: str,
+    time_entry_repository: TimeEntryCosmosDBRepository,
+):
     try:
         time_entry_repository.find_running(tenant_id, owner_id)
     except Exception as e:

tests/time_tracker_api/time_entries/time_entries_namespace_test.py

@@ -7,13 +7,12 @@
 from flask_restplus._http import HTTPStatus
 from pytest_mock import MockFixture, pytest
 
-from commons.data_access_layer.cosmos_db import (
+from utils.time import (
+    get_current_year,
+    get_current_month,
     current_datetime,
     current_datetime_str,
-    get_current_month,
-    get_current_year,
 )
-
 from utils import worked_time
 
 from werkzeug.exceptions import NotFound, UnprocessableEntity, HTTPException

time_tracker_api/time_entries/time_entries_model.py

@@ -11,13 +11,8 @@
     CosmosDBRepository,
     CustomError,
     CosmosDBModel,
-    current_datetime_str,
-    datetime_str,
-    get_date_range_of_month,
-    get_current_year,
-    get_current_month,
-    get_current_day,
 )
+
 from commons.data_access_layer.database import EventContext
 from time_tracker_api.activities import activities_model
 
@@ -28,10 +23,18 @@
     create_custom_query_from_str,
     add_user_email_to_time_entries,
 )
+from utils.time import (
+    datetime_str,
+    get_current_year,
+    get_current_month,
+    get_current_day,
+    get_date_range_of_month,
+    current_datetime_str,
+)
 from utils import worked_time
 from utils.worked_time import str_to_datetime
-
 from utils.azure_users import AzureConnection
+
 from time_tracker_api.projects.projects_model import ProjectCosmosDBModel
 from time_tracker_api.projects import projects_model
 from time_tracker_api.database import CRUDDao, APICosmosDBDao
@@ -440,18 +443,20 @@ def find_running(self):
         # self.stop_time_entry_if_was_left_running(time_entry)
         return time_entry
 
-    def get_worked_time(self, conditions: dict = {}):
+    def get_worked_time(self, args: dict = {}):
         event_ctx = self.create_event_context(
             "read", "Summary of worked time in the current month"
         )
-        conditions.update({"owner_id": event_ctx.user_id})
 
+        conditions = {"owner_id": event_ctx.user_id}
         time_entries = self.repository.find_all(
             event_ctx,
             conditions=conditions,
             date_range=worked_time.date_range(),
         )
-        return worked_time.summary(time_entries)
+        return worked_time.summary(
+            time_entries, time_offset=args.get('time_offset')
+        )
 
     @staticmethod
     def handle_date_filter_args(args: dict) -> dict: