#77: Add projects bulk editing (#89)

* Move Bulk Edit schemas and validation to base schemas
* Fix invalid query param type for bulk edit operations

Author: nifadyev

tests/factories/base.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Union
+from typing import TYPE_CHECKING, Dict, List, Union
 
 from tests.conftest import fake
 
@@ -22,3 +22,13 @@ def datetime_repr_factory(timezone: Union[ZoneInfo, TzInfo, None] = None) -> str
         timezone = zoneinfo.ZoneInfo(timezone_name)
 
     return fake.date_time_this_decade(tzinfo=timezone).isoformat(timespec="seconds")
+
+
+def bulk_edit_response_factory() -> Dict[str, List[Union[int, Dict[str, Union[int, str]]]]]:
+    return {
+        "success": [fake.random_int() for _ in range(fake.random_int(10))],
+        "failure": [
+            {"id": fake.random_int(), "message": fake.text(max_nb_chars=64)}
+            for _ in range(fake.random_int(4))
+        ],
+    }

tests/integration/test_project.py

@@ -6,7 +6,8 @@
 
 import pytest
 from toggl_python.exceptions import BadRequest
-from toggl_python.schemas.project import ProjectResponse
+from toggl_python.schemas.base import BulkEditOperation, BulkEditOperations, BulkEditResponse
+from toggl_python.schemas.project import BulkEditProjectsFieldNames, ProjectResponse
 
 from tests.conftest import fake
 from tests.factories.project import project_request_factory
@@ -442,6 +443,52 @@ def test_update_project__all_params(i_authed_workspace: Workspace) -> None:
     _ = i_authed_workspace.delete_project(workspace_id, project.id)
 
 
+def test_bulk_edit_projects(i_authed_workspace: Workspace) -> None:
+    workspace_id = int(os.environ["WORKSPACE_ID"])
+    project = i_authed_workspace.create_project(workspace_id, name=fake.uuid4(), active=True)
+    another_project = i_authed_workspace.create_project(
+        workspace_id, name=fake.uuid4(), active=True
+    )
+    full_request_body = project_request_factory()
+    allowed_fields = {item.value for item in BulkEditProjectsFieldNames}
+    random_param = fake.random_element(allowed_fields)
+    edit_operation = BulkEditOperation(
+        operation=BulkEditOperations.change,
+        field_name=random_param,
+        field_value=full_request_body[random_param],
+    )
+    expected_result = set(BulkEditResponse.model_fields.keys())
+
+    result = i_authed_workspace.bulk_edit_projects(
+        workspace_id, project_ids=[project.id, another_project.id], operations=[edit_operation]
+    )
+
+    assert result.model_fields_set == expected_result
+    assert project.id in result.success
+    assert another_project.id in result.success
+
+    _ = i_authed_workspace.delete_project(workspace_id, project.id)
+    _ = i_authed_workspace.delete_project(workspace_id, another_project.id)
+
+
+def test_bulk_edit_projects__forbid_to_edit_currency(i_authed_workspace: Workspace) -> None:
+    workspace_id = int(os.environ["WORKSPACE_ID"])
+    project = i_authed_workspace.create_project(workspace_id, name=fake.uuid4(), active=True)
+    edit_operation = BulkEditOperation(
+        operation=BulkEditOperations.change,
+        field_name="currency",
+        field_value=fake.currency_code(),
+    )
+    error_message = "User can not edit project billing info"
+
+    with pytest.raises(BadRequest, match=error_message):
+        _ = i_authed_workspace.bulk_edit_projects(
+            workspace_id, project_ids=[project.id], operations=[edit_operation]
+        )
+
+    _ = i_authed_workspace.delete_project(workspace_id, project.id)
+
+
 def test_delete_project(i_authed_workspace: Workspace) -> None:
     workspace_id = int(os.environ["WORKSPACE_ID"])
     project = i_authed_workspace.create_project(workspace_id, name=fake.uuid4())

tests/responses/time_entry_put_and_patch.py

@@ -7,9 +7,11 @@
 import pytest
 from httpx import Response as HttpxResponse
 from pydantic import ValidationError
-from toggl_python.schemas.project import ProjectResponse
+from toggl_python.schemas.base import BulkEditOperation, BulkEditOperations, BulkEditResponse
+from toggl_python.schemas.project import BulkEditProjectsFieldNames, ProjectResponse
 
 from tests.conftest import fake
+from tests.factories.base import bulk_edit_response_factory, datetime_repr_factory
 from tests.factories.project import project_request_factory, project_response_factory
 from tests.responses.project_get import PROJECT_RESPONSE
 
@@ -335,6 +337,76 @@ def test_update_project__invalid_timeframe(authed_workspace: Workspace) -> None:
         )
 
 
+def test_bulk_edit_projects__too_much_ids(authed_workspace: Workspace) -> None:
+    workspace_id = fake.random_int()
+    project_ids = [fake.random_int() for _ in range(101)]
+    error_message = "List should have at most 100 items after validation"
+
+    with pytest.raises(ValueError, match=error_message):
+        _ = authed_workspace.bulk_edit_projects(workspace_id, project_ids, operations=[])
+
+
+def test_bulk_edit_projects__empty_projects_ids(authed_workspace: Workspace) -> None:
+    workspace_id = fake.random_int()
+    error_message = "List should have at least 1 item after validation"
+
+    with pytest.raises(ValueError, match=error_message):
+        _ = authed_workspace.bulk_edit_projects(workspace_id, project_ids=[], operations=[])
+
+
+def test_bulk_edit_projects__empty_operations(authed_workspace: Workspace) -> None:
+    workspace_id = fake.random_int()
+    project_ids = [fake.random_int()]
+    error_message = "List should have at least 1 item after validation"
+
+    with pytest.raises(ValueError, match=error_message):
+        _ = authed_workspace.bulk_edit_projects(workspace_id, project_ids, operations=[])
+
+
+@pytest.mark.parametrize(
+    argnames=("operation"), argvalues=[item.value for item in BulkEditOperations]
+)
+@pytest.mark.parametrize(
+    argnames=("field_name", "field_value"),
+    argvalues=[
+        (BulkEditProjectsFieldNames.auto_estimates.value, fake.boolean()),
+        (BulkEditProjectsFieldNames.end_date.value, datetime_repr_factory()),
+        (BulkEditProjectsFieldNames.estimated_hours.value, fake.random_int()),
+        (BulkEditProjectsFieldNames.is_private.value, fake.boolean()),
+        (BulkEditProjectsFieldNames.project_name.value, fake.uuid4()),
+        (BulkEditProjectsFieldNames.start_date.value, fake.date()),
+        (BulkEditProjectsFieldNames.template.value, fake.boolean()),
+    ],
+)
+def test_bulk_edit_time_entries__ok(
+    field_name: BulkEditProjectsFieldNames,
+    field_value: Union[str, int],
+    operation: BulkEditOperations,
+    response_mock: MockRouter,
+    authed_workspace: Workspace,
+) -> None:
+    workspace_id = fake.random_int()
+    project_ids = [fake.random_int(), fake.random_int()]
+    project_ids_repr = ",".join(str(item) for item in project_ids)
+    edit_operation = BulkEditOperation(
+        operation=operation, field_name=field_name, field_value=field_value
+    )
+    response = bulk_edit_response_factory()
+    mocked_route = response_mock.patch(
+        f"/workspaces/{workspace_id}/projects/{project_ids_repr}"
+    ).mock(
+        return_value=HttpxResponse(status_code=200, json=response),
+    )
+    expected_result = BulkEditResponse.model_validate(response)
+
+    result = authed_workspace.bulk_edit_projects(
+        workspace_id, project_ids, operations=[edit_operation]
+    )
+
+    assert mocked_route.called is True
+    assert result == expected_result
+
+
 def test_delete_project(response_mock: MockRouter, authed_workspace: Workspace) -> None:
     workspace_id = fake.random_int()
     project_id = fake.random_int()

tests/test_project.py

@@ -1,33 +1,30 @@
 from __future__ import annotations
 
 from datetime import date, datetime, timezone
-from random import randint
 from typing import TYPE_CHECKING, Dict, List, 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.base import BulkEditOperation, BulkEditOperations, BulkEditResponse
 from toggl_python.schemas.time_entry import (
     BulkEditTimeEntriesFieldNames,
-    BulkEditTimeEntriesOperation,
-    BulkEditTimeEntriesOperations,
-    BulkEditTimeEntriesResponse,
     MeTimeEntryResponse,
     MeTimeEntryWithMetaResponse,
     MeWebTimerResponse,
 )
 
 from tests.conftest import fake
+from tests.factories.base import bulk_edit_response_factory, datetime_repr_factory
 from tests.factories.time_entry import (
     time_entry_extended_request_factory,
     time_entry_request_factory,
     time_entry_response_factory,
 )
 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
-from tests.responses.time_entry_put_and_patch import BULK_EDIT_TIME_ENTRIES_RESPONSE
 
 
 if TYPE_CHECKING:
@@ -119,6 +116,7 @@ def test_create_time_entry__invalid_start_stop_and_duration(authed_workspace: Wo
             stop=request_body["stop"],
         )
 
+
 def test_get_time_entry__without_query_params(
     response_mock: MockRouter, authed_current_user: CurrentUser
 ) -> None:
@@ -419,68 +417,79 @@ def test_delete_time_entry__ok(response_mock: MockRouter, authed_workspace: Work
 
 
 def test_bulk_edit_time_entries__too_much_ids(authed_workspace: Workspace) -> None:
-    workspace_id = 123
-    time_entry_ids = [randint(100000, 999999) for _ in range(101)]  # noqa: S311
-    error_message = "Limit to max TimeEntry IDs exceeded. "
+    workspace_id = fake.random_int()
+    time_entry_ids = [fake.random_int() for _ in range(101)]
+    error_message = "List should have at most 100 items after validation"
 
     with pytest.raises(ValueError, match=error_message):
         _ = authed_workspace.bulk_edit_time_entries(workspace_id, time_entry_ids, operations=[])
 
 
 def test_bulk_edit_time_entries__empty_time_entry_ids(authed_workspace: Workspace) -> None:
-    workspace_id = 123
-    error_message = "Specify at least one TimeEntry ID"
+    workspace_id = fake.random_int()
+    error_message = "List should have at least 1 item after validation"
 
     with pytest.raises(ValueError, match=error_message):
         _ = authed_workspace.bulk_edit_time_entries(workspace_id, time_entry_ids=[], operations=[])
 
 
 def test_bulk_edit_time_entries__empty_operations(authed_workspace: Workspace) -> None:
-    workspace_id = 123
-    time_entry_ids = [12345677]
-    error_message = "Specify at least one edit operation"
+    workspace_id = fake.random_int()
+    time_entry_ids = [fake.random_int()]
+    error_message = "List should have at least 1 item after validation"
 
     with pytest.raises(ValueError, match=error_message):
         _ = authed_workspace.bulk_edit_time_entries(workspace_id, time_entry_ids, operations=[])
 
 
 @pytest.mark.parametrize(
-    argnames=("operation"), argvalues=[item.value for item in BulkEditTimeEntriesOperations]
+    argnames=("operation"), argvalues=[item.value for item in BulkEditOperations]
 )
 @pytest.mark.parametrize(
     argnames=("field_name", "field_value"),
     argvalues=[
-        (BulkEditTimeEntriesFieldNames.billable.value, True),
-        (BulkEditTimeEntriesFieldNames.description.value, "updated description"),
-        (BulkEditTimeEntriesFieldNames.duration.value, -1),
-        (BulkEditTimeEntriesFieldNames.project_id.value, 757542305),
-        (BulkEditTimeEntriesFieldNames.shared_with_user_ids.value, [1243543643, 676586868]),
-        (BulkEditTimeEntriesFieldNames.start.value, datetime(2024, 5, 10, tzinfo=timezone.utc)),
-        (BulkEditTimeEntriesFieldNames.stop.value, datetime(2022, 4, 15, tzinfo=timezone.utc)),
-        (BulkEditTimeEntriesFieldNames.tag_ids.value, [24032, 354742502]),
-        (BulkEditTimeEntriesFieldNames.tags.value, ["new tag"]),
-        (BulkEditTimeEntriesFieldNames.task_id.value, 1593268409),
-        (BulkEditTimeEntriesFieldNames.user_id.value, 573250897),
+        (BulkEditTimeEntriesFieldNames.billable.value, fake.boolean()),
+        (BulkEditTimeEntriesFieldNames.description.value, fake.text(max_nb_chars=32)),
+        (BulkEditTimeEntriesFieldNames.duration.value, fake.random_int(min=-1, max=128)),
+        (BulkEditTimeEntriesFieldNames.project_id.value, fake.random_int()),
+        (
+            BulkEditTimeEntriesFieldNames.shared_with_user_ids.value,
+            [fake.random_int() for _ in range(fake.random_int(max=10))],
+        ),
+        (BulkEditTimeEntriesFieldNames.start.value, datetime_repr_factory()),
+        (BulkEditTimeEntriesFieldNames.stop.value, datetime_repr_factory()),
+        (
+            BulkEditTimeEntriesFieldNames.tag_ids.value,
+            [fake.random_int() for _ in range(fake.random_int(max=10))],
+        ),
+        (
+            BulkEditTimeEntriesFieldNames.tags.value,
+            [fake.word() for _ in range(fake.random_int(max=10))],
+        ),
+        (BulkEditTimeEntriesFieldNames.task_id.value, fake.random_int()),
+        (BulkEditTimeEntriesFieldNames.user_id.value, fake.random_int()),
     ],
 )
 def test_bulk_edit_time_entries__ok(
     field_name: BulkEditTimeEntriesFieldNames,
     field_value: Union[str, int],
-    operation: BulkEditTimeEntriesOperations,
+    operation: BulkEditOperations,
     response_mock: MockRouter,
     authed_workspace: Workspace,
 ) -> None:
-    workspace_id = 123
-    time_entry_ids = [98765, 43210]
-    edit_operation = BulkEditTimeEntriesOperation(
+    workspace_id = fake.random_int()
+    time_entry_ids = [fake.random_int() for _ in range(fake.random_int(min=1, max=8))]
+    time_entry_ids_repr = ",".join(str(item) for item in time_entry_ids)
+    edit_operation = BulkEditOperation(
         operation=operation, field_name=field_name, field_value=field_value
     )
+    response = bulk_edit_response_factory()
     mocked_route = response_mock.patch(
-        f"/workspaces/{workspace_id}/time_entries/{time_entry_ids}"
+        f"/workspaces/{workspace_id}/time_entries/{time_entry_ids_repr}"
     ).mock(
-        return_value=Response(status_code=200, json=BULK_EDIT_TIME_ENTRIES_RESPONSE),
+        return_value=Response(status_code=200, json=response),
     )
-    expected_result = BulkEditTimeEntriesResponse.model_validate(BULK_EDIT_TIME_ENTRIES_RESPONSE)
+    expected_result = BulkEditResponse.model_validate(response)
 
     result = authed_workspace.bulk_edit_time_entries(
         workspace_id, time_entry_ids, operations=[edit_operation]