#8: Implement TimeEntry entities fetching (#52)

nifadyev · web-flow · commit c4042aeb4410 · 2024-08-30T14:55:38.000+03:00
diff --git a/tests/responses/me_get.py b/tests/responses/me_get.py
@@ -105,3 +105,28 @@
     "send_weekly_report": False,
     "timeofday_format": "H:mm",
 }
+
+ME_WEB_TIMER_RESPONSE: Dict[str, Union[None, List[Dict]]] = {
+    "clients": None,
+    "projects": [],
+    "tags": [],
+    "tasks": None,
+    "time_entries": [
+        {
+            "billable": False,
+            "deleted": None,
+            "description": "test timer",
+            "duration_in_seconds": 52,
+            "ignore_start_and_stop": True,
+            "planned_task_id": None,
+            "project_id": 202793182,
+            "tag_ids": [16501871],
+            "task_id": 3545645770,
+            "updated_at": "2024-07-30T08:14:38+00:00",
+            "user_id": 30809356,
+            "utc_start": "2024-07-30T08:13:46+00:00",
+            "utc_stop": "2024-07-30T08:14:38+00:00",
+            "workspace_id": 43644207,
+        },
+    ],
+}
diff --git a/tests/responses/time_entry_get.py b/tests/responses/time_entry_get.py
@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+from typing import Dict, Union
+
+
+ME_TIME_ENTRY_RESPONSE: Dict[str, Union[str, int, bool, list, None]] = {
+    "at": "2024-07-29T12:28:56+00:00",
+    "billable": False,
+    "description": "test timer",
+    "duration": 22,
+    "duronly": True,
+    "id": 3544298808,
+    "permissions": None,
+    "project_id": None,
+    "server_deleted_at": None,
+    "start": "2024-07-29T12:28:33+00:00",
+    "stop": "2024-07-29T12:28:55+00:00",
+    "tag_ids": [],
+    "tags": [],
+    "task_id": None,
+    "user_id": 30809356,
+    "workspace_id": 43644207,
+}
+
+ME_TIME_ENTRY_WITH_META_RESPONSE: Dict[str, Union[str, int, bool, list, None]] = {
+    "at": "2024-07-29T12:28:56+00:00",
+    "billable": False,
+    "description": "test timer",
+    "duration": 22,
+    "duronly": True,
+    "id": 3544298808,
+    "permissions": None,
+    "project_id": None,
+    "server_deleted_at": None,
+    "start": "2024-07-29T12:28:33+00:00",
+    "stop": "2024-07-29T12:28:55+00:00",
+    "tag_ids": [],
+    "tags": [],
+    "task_id": None,
+    "user_avatar_url": "",
+    "user_id": 30809356,
+    "user_name": "Test User",
+    "workspace_id": 43644207,
+}
diff --git a/tests/test_time_entry.py b/tests/test_time_entry.py
@@ -0,0 +1,214 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import TYPE_CHECKING, Dict, Union
+from unittest.mock import Mock, patch
+
+import pytest
+from httpx import Response
+from pydantic import ValidationError
+from toggl_python.exceptions import BadRequest
+from toggl_python.schemas.time_entry import (
+    MeTimeEntryResponse,
+    MeTimeEntryWithMetaResponse,
+    MeWebTimerResponse,
+)
+
+from tests.responses.me_get import ME_WEB_TIMER_RESPONSE
+from tests.responses.time_entry_get import ME_TIME_ENTRY_RESPONSE, ME_TIME_ENTRY_WITH_META_RESPONSE
+
+
+if TYPE_CHECKING:
+    from respx import MockRouter
+    from toggl_python.entities.user import CurrentUser
+
+
+def test_get_time_entry__without_query_params(
+    response_mock: MockRouter, authed_current_user: CurrentUser
+) -> None:
+    fake_time_entry_id = 123
+    mocked_route = response_mock.get(f"/me/time_entries/{fake_time_entry_id}").mock(
+        return_value=Response(status_code=200, json=ME_TIME_ENTRY_RESPONSE),
+    )
+    expected_result = MeTimeEntryResponse.model_validate(ME_TIME_ENTRY_RESPONSE)
+
+    result = authed_current_user.get_time_entry(fake_time_entry_id)
+
+    assert mocked_route.called is True
+    assert result == expected_result
+
+
+def test_get_time_entry__with_meta_query_param(
+    response_mock: MockRouter, authed_current_user: CurrentUser
+) -> None:
+    fake_time_entry_id = 123
+    mocked_route = response_mock.get(f"/me/time_entries/{fake_time_entry_id}?meta=true").mock(
+        return_value=Response(status_code=200, json=ME_TIME_ENTRY_WITH_META_RESPONSE),
+    )
+    expected_result = MeTimeEntryWithMetaResponse.model_validate(ME_TIME_ENTRY_WITH_META_RESPONSE)
+
+    result = authed_current_user.get_time_entry(fake_time_entry_id, meta=True)
+
+    assert mocked_route.called is True
+    assert result == expected_result
+
+
+def test_get_current_time_entry__ok(
+    response_mock: MockRouter, authed_current_user: CurrentUser
+) -> None:
+    mocked_route = response_mock.get("/me/time_entries/current").mock(
+        return_value=Response(status_code=200, json=ME_TIME_ENTRY_RESPONSE),
+    )
+    expected_result = MeTimeEntryResponse.model_validate(ME_TIME_ENTRY_RESPONSE)
+
+    result = authed_current_user.get_current_time_entry()
+
+    assert mocked_route.called is True
+    assert result == expected_result
+
+
+def test_get_current_time_entry__no_current_entry(
+    response_mock: MockRouter, authed_current_user: CurrentUser
+) -> None:
+    mocked_route = response_mock.get("/me/time_entries/current").mock(
+        return_value=Response(status_code=200, json={}),
+    )
+
+    result = authed_current_user.get_current_time_entry()
+
+    assert mocked_route.called is True
+    assert result is None
+
+
+def test_get_time_entries__without_query_params(
+    response_mock: MockRouter, authed_current_user: CurrentUser
+) -> None:
+    fake_response = [ME_TIME_ENTRY_RESPONSE]
+    mocked_route = response_mock.get("/me/time_entries").mock(
+        return_value=Response(status_code=200, json=fake_response),
+    )
+    expected_result = [MeTimeEntryResponse.model_validate(ME_TIME_ENTRY_RESPONSE)]
+
+    result = authed_current_user.get_time_entries()
+
+    assert mocked_route.called is True
+    assert result == expected_result
+
+
+def test_get_time_entries__with_meta_query_param(
+    response_mock: MockRouter, authed_current_user: CurrentUser
+) -> None:
+    mocked_route = response_mock.get("/me/time_entries", params={"meta": True}).mock(
+        return_value=Response(status_code=200, json=[ME_TIME_ENTRY_WITH_META_RESPONSE]),
+    )
+    expected_result = [
+        MeTimeEntryWithMetaResponse.model_validate(ME_TIME_ENTRY_WITH_META_RESPONSE)
+    ]
+
+    result = authed_current_user.get_time_entries(meta=True)
+
+    assert mocked_route.called is True
+    assert result == expected_result
+
+
+@patch("toggl_python.schemas.time_entry.datetime")
+@pytest.mark.parametrize(
+    argnames="query_params, method_kwargs",
+    argvalues=(
+        (
+            {"since": 1715299200},
+            {"since": int(datetime(2024, 5, 10, tzinfo=timezone.utc).timestamp())},
+        ),
+        ({"since": 1718755200}, {"since": 1718755200}),
+        ({"before": "2024-07-28T12:30:43+00:00"}, {"before": "2024-07-28T12:30:43+00:00"}),
+        (
+            {"before": "2023-01-01T00:00:00+00:00"},
+            {"before": datetime(2023, 1, 1, tzinfo=timezone.utc)},
+        ),
+        (
+            {"start_date": "2023-09-12T00:00:00-03:00", "end_date": "2023-10-12T00:00:00-01:00"},
+            {"start_date": "2023-09-12T00:00:00-03:00", "end_date": "2023-10-12T00:00:00-01:00"},
+        ),
+    ),
+)
+def test_get_time_entries__with_datetime_query_params(
+    mocked_datetime: Mock,
+    query_params: Dict[str, Union[int, str]],
+    method_kwargs: Dict[str, Union[datetime, str]],
+    response_mock: MockRouter,
+    authed_current_user: CurrentUser,
+) -> None:
+    query_params["meta"] = False
+    # Required to pass `since` query param validation
+    mocked_datetime.now.return_value = datetime(2024, 6, 20, tzinfo=timezone.utc)
+    mocked_route = response_mock.get("/me/time_entries", params=query_params).mock(
+        return_value=Response(status_code=200, json=[ME_TIME_ENTRY_RESPONSE]),
+    )
+    expected_result = [MeTimeEntryResponse.model_validate(ME_TIME_ENTRY_RESPONSE)]
+
+    result = authed_current_user.get_time_entries(**method_kwargs)
+
+    assert mocked_route.called is True
+    assert result == expected_result
+
+
+@pytest.mark.parametrize(
+    argnames="query_params",
+    argvalues=(
+        {"start_date": "2010-01-01T00:00:00+08:00"},
+        {"end_date": "2010-02-01T00:00:00+03:00"},
+        {"since": 17223107204, "before": "2024-07-28T00:00:00+10:00"},
+        {
+            "since": 17223107204,
+            "start_date": "2020-11-11T09:30:00-04:00",
+            "end_date": "2021-01-11T09:30:00-04:00",
+        },
+        {
+            "before": "2020-12-15T09:30:00-04:00",
+            "start_date": "2020-11-11T09:30:00-04:00",
+            "end_date": "2021-01-11T09:30:00-04:00",
+        },
+        {
+            "since": 17223107204,
+            "before": "2020-12-15T09:30:00-04:00",
+            "start_date": "2020-11-11T09:30:00-04:00",
+            "end_date": "2021-01-11T09:30:00-04:00",
+        },
+    ),
+)
+def test_get_time_entries__invalid_query_params(
+    query_params: Dict[str, Union[int, str]],
+    response_mock: MockRouter,
+    authed_current_user: CurrentUser,
+) -> None:
+    error_message = "can not be present simultaneously"
+    _ = response_mock.get("/me/time_entries", params=query_params).mock(
+        return_value=Response(status_code=400, json=error_message),
+    )
+
+    with pytest.raises(BadRequest, match=error_message):
+        _ = authed_current_user.get_time_entries(**query_params)
+
+
+@patch("toggl_python.schemas.time_entry.datetime")
+def test_get_time_entries__too_old_since_value(
+    mocked_datetime: Mock, authed_current_user: CurrentUser
+) -> None:
+    error_message = "Since cannot be older than 3 months"
+    since = datetime(2020, 1, 1, tzinfo=timezone.utc)
+    mocked_datetime.now.return_value = datetime(2020, 4, 1, tzinfo=timezone.utc)
+
+    with pytest.raises(ValidationError, match=error_message):
+        _ = authed_current_user.get_time_entries(since=since)
+
+
+def test_get_web_timer__ok(response_mock: MockRouter, authed_current_user: CurrentUser) -> None:
+    mocked_route = response_mock.get("/me/web-timer").mock(
+        return_value=Response(status_code=200, json=ME_WEB_TIMER_RESPONSE),
+    )
+    expected_result = MeWebTimerResponse.model_validate(ME_WEB_TIMER_RESPONSE)
+
+    result = authed_current_user.get_web_timer()
+
+    assert mocked_route.called is True
+    assert result == expected_result
diff --git a/toggl_python/entities/user.py b/toggl_python/entities/user.py
@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, List, Optional
+from typing import TYPE_CHECKING, List, Optional, Union
 
 from toggl_python.api import ApiWrapper
 from toggl_python.schemas.current_user import (
@@ -16,9 +16,17 @@
     UpdateMeRequest,
     UpdateMeResponse,
 )
+from toggl_python.schemas.time_entry import (
+    MeTimeEntryQueryParams,
+    MeTimeEntryResponse,
+    MeTimeEntryWithMetaResponse,
+    MeWebTimerResponse,
+)
 
 
 if TYPE_CHECKING:
+    from datetime import datetime
+
     from pydantic import EmailStr
 
 
@@ -128,3 +136,67 @@ def update_preferences(
 
         response_body = response.json()
         return MePreferencesResponse.model_validate(response_body)
+
+    def get_time_entry(
+        self, time_entry_id: int, meta: bool = False
+    ) -> Union[MeTimeEntryResponse, MeTimeEntryWithMetaResponse]:
+        """Intentionally use the same schema for requests with `include_sharing=true`.
+
+        Tested responses do not differ from requests with `include_sharing=false`
+        that is why there is no `include_sharing` method argument.
+        """
+        response = self.client.get(
+            url=f"{self.prefix}/time_entries/{time_entry_id}",
+            params={"meta": meta},
+        )
+        self.raise_for_status(response)
+
+        response_schema = MeTimeEntryWithMetaResponse if meta else MeTimeEntryResponse
+
+        response_body = response.json()
+        return response_schema.model_validate(response_body)
+
+    def get_current_time_entry(self) -> Optional[MeTimeEntryResponse]:
+        """Return empty response if there is no running TimeEntry."""
+        response = self.client.get(url=f"{self.prefix}/time_entries/current")
+        self.raise_for_status(response)
+
+        response_body = response.json()
+        return MeTimeEntryResponse.model_validate(response_body) if response_body else None
+
+    def get_time_entries(
+        self,
+        meta: bool = False,
+        since: Union[int, datetime, None] = None,
+        before: Union[str, datetime, None] = None,
+        start_date: Union[str, datetime, None] = None,
+        end_date: Union[str, datetime, None] = None,
+    ) -> List[Union[MeTimeEntryResponse, MeTimeEntryWithMetaResponse]]:
+        """Intentionally use the same schema for requests with `include_sharing=true`.
+
+        Tested responses do not differ from requests with `include_sharing=false`
+        that is why there is no `include_sharing` method argument.
+        """
+        payload_schema = MeTimeEntryQueryParams(
+            meta=meta,
+            since=since,
+            before=before,
+            start_date=start_date,
+            end_date=end_date,
+        )
+        payload = payload_schema.model_dump(mode="json", exclude_none=True)
+
+        response = self.client.get(url=f"{self.prefix}/time_entries", params=payload)
+        self.raise_for_status(response)
+
+        response_schema = MeTimeEntryWithMetaResponse if meta else MeTimeEntryResponse
+
+        response_body = response.json()
+        return [response_schema.model_validate(time_entry) for time_entry in response_body]
+
+    def get_web_timer(self) -> MeWebTimerResponse:
+        response = self.client.get(url=f"{self.prefix}/web-timer")
+        self.raise_for_status(response)
+
+        response_body = response.json()
+        return MeWebTimerResponse.model_validate(response_body)
diff --git a/toggl_python/schemas/time_entry.py b/toggl_python/schemas/time_entry.py
