refactor(toggl): split API implementation from service

Author: thekaveman

compiler_admin/api/toggl.py

@@ -0,0 +1,110 @@
+from base64 import b64encode
+from datetime import datetime
+
+import requests
+
+from compiler_admin import __version__
+
+
+class Toggl:
+    """Toggl API Client.
+
+    See https://engineering.toggl.com/docs/.
+    """
+
+    API_BASE_URL = "https://api.track.toggl.com"
+    API_REPORTS_BASE_URL = "reports/api/v3"
+    API_WORKSPACE = "workspace/{}"
+    API_HEADERS = {"Content-Type": "application/json", "User-Agent": "compilerla/compiler-admin:{}".format(__version__)}
+
+    def __init__(self, api_token: str, workspace_id: int, **kwargs):
+        self._token = api_token
+        self.workspace_id = workspace_id
+
+        self.headers = dict(Toggl.API_HEADERS)
+        self.headers.update(self._authorization_header())
+
+        self.timeout = int(kwargs.get("timeout", 5))
+
+    @property
+    def workspace_url_fragment(self):
+        """The workspace portion of an API URL."""
+        return Toggl.API_WORKSPACE.format(self.workspace_id)
+
+    def _authorization_header(self):
+        """Gets an `Authorization: Basic xyz` header using the Toggl API token.
+
+        See https://engineering.toggl.com/docs/authentication.
+        """
+        creds = f"{self._token}:api_token"
+        creds64 = b64encode(bytes(creds, "utf-8")).decode("utf-8")
+        return {"Authorization": "Basic {}".format(creds64)}
+
+    def _make_report_url(self, endpoint: str):
+        """Get a fully formed URL for the Toggl Reports API v3 endpoint.
+
+        See https://engineering.toggl.com/docs/reports_start.
+        """
+        return "/".join((Toggl.API_BASE_URL, Toggl.API_REPORTS_BASE_URL, self.workspace_url_fragment, endpoint))
+
+    def detailed_time_entries(self, start_date: datetime, end_date: datetime, **kwargs):
+        """Request a CSV report from Toggl of detailed time entries for the given date range.
+
+        Args:
+            start_date (datetime): The beginning of the reporting period.
+
+            end_date (str): The end of the reporting period.
+
+        Extra `kwargs` are passed through as a POST json body.
+
+        By default, requests a report with the following configuration:
+            * `billable=True`
+            * `rounding=1` (True, but this is an int param)
+            * `rounding_minutes=15`
+
+        See https://engineering.toggl.com/docs/reports/detailed_reports#post-export-detailed-report.
+
+        Returns:
+            response (requests.Response): The HTTP response.
+        """
+        # ensure start_date precedes end_date
+        start_date, end_date = min(start_date, end_date), max(start_date, end_date)
+        start = start_date.strftime("%Y-%m-%d")
+        end = end_date.strftime("%Y-%m-%d")
+
+        # calculate a timeout based on the size of the reporting period in days
+        # approximately 5 seconds per month of query size, with a minimum of 5 seconds
+        range_days = (end_date - start_date).days
+        current_timeout = self.timeout
+        dynamic_timeout = int((max(30, range_days) / 30.0) * 5)
+        self.timeout = max(current_timeout, dynamic_timeout)
+
+        params = dict(
+            billable=True,
+            start_date=start,
+            end_date=end,
+            rounding=1,
+            rounding_minutes=15,
+        )
+        params.update(kwargs)
+
+        response = self.post_reports("search/time_entries.csv", **params)
+        self.timeout = current_timeout
+
+        return response
+
+    def post_reports(self, endpoint: str, **kwargs) -> requests.Response:
+        """Send a POST request to the Reports v3 `endpoint`.
+
+        Extra `kwargs` are passed through as a POST json body.
+
+        Will raise for non-200 status codes.
+
+        See https://engineering.toggl.com/docs/reports_start.
+        """
+        url = self._make_report_url(endpoint)
+
+        response = requests.post(url, json=kwargs, timeout=self.timeout)
+        response.raise_for_status()
+
+        return response

compiler_admin/services/toggl.py

@@ -1,14 +1,12 @@
-from base64 import b64encode
 from datetime import datetime
 import io
 import os
 import sys
 from typing import TextIO
 
 import pandas as pd
-import requests
 
-from compiler_admin import __version__
+from compiler_admin.api.toggl import Toggl
 from compiler_admin.services.google import user_info as google_user_info
 import compiler_admin.services.files as files
 
@@ -23,110 +21,6 @@
 OUTPUT_COLUMNS = ["Date", "Client", "Project", "Task", "Notes", "Hours", "First name", "Last name"]
 
 
-class Toggl:
-    """Toggl API Client.
-
-    See https://engineering.toggl.com/docs/.
-    """
-
-    API_BASE_URL = "https://api.track.toggl.com"
-    API_REPORTS_BASE_URL = "reports/api/v3"
-    API_WORKSPACE = "workspace/{}"
-    API_HEADERS = {"Content-Type": "application/json", "User-Agent": "compilerla/compiler-admin:{}".format(__version__)}
-
-    def __init__(self, api_token: str, workspace_id: int, **kwargs):
-        self._token = api_token
-        self.workspace_id = workspace_id
-
-        self.headers = dict(Toggl.API_HEADERS)
-        self.headers.update(self._authorization_header())
-
-        self.timeout = int(kwargs.get("timeout", 5))
-
-    @property
-    def workspace_url_fragment(self):
-        """The workspace portion of an API URL."""
-        return Toggl.API_WORKSPACE.format(self.workspace_id)
-
-    def _authorization_header(self):
-        """Gets an `Authorization: Basic xyz` header using the Toggl API token.
-
-        See https://engineering.toggl.com/docs/authentication.
-        """
-        creds = f"{self._token}:api_token"
-        creds64 = b64encode(bytes(creds, "utf-8")).decode("utf-8")
-        return {"Authorization": "Basic {}".format(creds64)}
-
-    def _make_report_url(self, endpoint: str):
-        """Get a fully formed URL for the Toggl Reports API v3 endpoint.
-
-        See https://engineering.toggl.com/docs/reports_start.
-        """
-        return "/".join((Toggl.API_BASE_URL, Toggl.API_REPORTS_BASE_URL, self.workspace_url_fragment, endpoint))
-
-    def post_reports(self, endpoint: str, **kwargs) -> requests.Response:
-        """Send a POST request to the Reports v3 `endpoint`.
-
-        Extra `kwargs` are passed through as a POST json body.
-
-        Will raise for non-200 status codes.
-
-        See https://engineering.toggl.com/docs/reports_start.
-        """
-        url = self._make_report_url(endpoint)
-
-        response = requests.post(url, json=kwargs, headers=self.headers, timeout=self.timeout)
-        response.raise_for_status()
-
-        return response
-
-    def detailed_time_entries(self, start_date: datetime, end_date: datetime, **kwargs):
-        """Request a CSV report from Toggl of detailed time entries for the given date range.
-
-        Args:
-            start_date (datetime): The beginning of the reporting period.
-
-            end_date (str): The end of the reporting period.
-
-        Extra `kwargs` are passed through as a POST json body.
-
-        By default, requests a report with the following configuration:
-            * `billable=True`
-            * `rounding=1` (True, but this is an int param)
-            * `rounding_minutes=15`
-
-        See https://engineering.toggl.com/docs/reports/detailed_reports#post-export-detailed-report.
-
-        Returns:
-            response (requests.Response): The HTTP response.
-        """
-        # ensure start_date precedes end_date
-        start_date, end_date = min(start_date, end_date), max(start_date, end_date)
-        start = start_date.strftime("%Y-%m-%d")
-        end = end_date.strftime("%Y-%m-%d")
-
-        # calculate a timeout based on the size of the reporting period in days
-        # approximately 5 seconds per month of query size, with a minimum of 5 seconds
-        range_days = (end_date - start_date).days
-        current_timeout = self.timeout
-        dynamic_timeout = int((max(30, range_days) / 30.0) * 5)
-        self.timeout = max(current_timeout, dynamic_timeout)
-
-        params = dict(
-            billable=True,
-            start_date=start,
-            end_date=end,
-            rounding=1,
-            rounding_minutes=15,
-        )
-        params.update(kwargs)
-
-        response = self.post_reports("search/time_entries.csv", **params)
-        self.timeout = current_timeout
-
-        return response
-
-
 def _get_first_name(email: str) -> str:
     """Get cached first name or derive from email."""
     user = USER_INFO.get(email)

tests/api/__init__.py

@@ -0,0 +1,95 @@
+from datetime import datetime
+from pathlib import Path
+
+import pytest
+
+from compiler_admin import __version__
+from compiler_admin.api.toggl import __name__ as MODULE, Toggl
+
+
+@pytest.fixture
+def mock_requests(mocker):
+    return mocker.patch(f"{MODULE}.requests")
+
+
+@pytest.fixture
+def toggl():
+    return Toggl("token", 1234)
+
+
+@pytest.fixture
+def toggl_mock_post_reports(mocker, toggl, toggl_file):
+    # setup a mock response to a requests.post call
+    mock_csv_bytes = Path(toggl_file).read_bytes()
+    mock_post_response = mocker.Mock()
+    mock_post_response.raise_for_status.return_value = None
+    # prepend the BOM to the mock content
+    mock_post_response.content = b"\xef\xbb\xbf" + mock_csv_bytes
+    # override the requests.post call to return the mock response
+    mocker.patch.object(toggl, "post_reports", return_value=mock_post_response)
+    return toggl
+
+
+def test_toggl_init(toggl):
+    token64 = "dG9rZW46YXBpX3Rva2Vu"
+
+    assert toggl._token == "token"
+    assert toggl.workspace_id == 1234
+    assert toggl.workspace_url_fragment == "workspace/1234"
+
+    assert toggl.headers["Content-Type"] == "application/json"
+
+    user_agent = toggl.headers["User-Agent"]
+    assert "compilerla/compiler-admin" in user_agent
+    assert __version__ in user_agent
+
+    assert toggl.headers["Authorization"] == f"Basic {token64}"
+
+    assert toggl.timeout == 5
+
+
+def test_toggl_make_report_url(toggl):
+    url = toggl._make_report_url("endpoint")
+
+    assert url.startswith(toggl.API_BASE_URL)
+    assert toggl.API_REPORTS_BASE_URL in url
+    assert toggl.workspace_url_fragment in url
+    assert "/endpoint" in url
+
+
+def test_toggl_post_reports(mock_requests, toggl):
+    url = toggl._make_report_url("endpoint")
+    response = toggl.post_reports("endpoint", kwarg1=1, kwarg2="two")
+
+    response.raise_for_status.assert_called_once()
+
+    mock_requests.post.assert_called_once_with(
+        url, json=dict(kwarg1=1, kwarg2="two"), headers=toggl.headers, timeout=toggl.timeout
+    )
+
+
+def test_toggl_detailed_time_entries(toggl_mock_post_reports):
+    dt = datetime(2024, 9, 25)
+    toggl_mock_post_reports.detailed_time_entries(dt, dt, kwarg1=1, kwarg2="two")
+
+    toggl_mock_post_reports.post_reports.assert_called_once_with(
+        "search/time_entries.csv",
+        billable=True,
+        start_date="2024-09-25",
+        end_date="2024-09-25",
+        rounding=1,
+        rounding_minutes=15,
+        kwarg1=1,
+        kwarg2="two",
+    )
+
+
+def test_toggl_detailed_time_entries_dynamic_timeout(mock_requests, toggl):
+    # range of 6 months
+    # timeout should be 6 * 5 = 30
+    start = datetime(2024, 1, 1)
+    end = datetime(2024, 6, 30)
+    toggl.detailed_time_entries(start, end)
+
+    mock_requests.post.assert_called_once()
+    assert mock_requests.post.call_args.kwargs["timeout"] == 30