style: fix docstring and linting issues

Co-Authored-By: Rushil Srivastava <[email protected]>

Author: devin-ai-integration[bot]

tests/test_delayed_route.py

@@ -1,4 +1,5 @@
-"""Tests for DelayedRouteBuilder functionality.
+"""
+Tests for DelayedRouteBuilder functionality.
 
 This module verifies the core functionality of DelayedRouteBuilder, including:
 - Basic task creation and execution
@@ -9,7 +10,7 @@
 """
 
 import pytest
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends
 from google.protobuf import duration_pb2
 from pydantic import BaseModel
 
@@ -87,8 +88,9 @@ async def test_task(payload: TestPayload):
 
 
 def test_delayed_task_with_task_id(app, delayed_route, test_client):
-    """Test task creation with task ID.
-    
+    """
+    Test task creation with task ID.
+
     This test verifies that:
     1. Tasks can be created with unique IDs
     2. Duplicate task IDs are handled correctly
@@ -103,24 +105,21 @@ async def test_task(payload: TestPayload):
     app.include_router(router)
 
     # Test with unique task ID
-    task1 = test_task.options(task_id="unique-task-1").delay(
-        payload=TestPayload(message="test1")
-    )
+    task1 = test_task.options(task_id="unique-task-1").delay(payload=TestPayload(message="test1"))
     assert task1 is not None
 
     # Test with duplicate task ID (should be idempotent)
-    task2 = test_task.options(task_id="unique-task-1").delay(
-        payload=TestPayload(message="test1")
-    )
+    task2 = test_task.options(task_id="unique-task-1").delay(payload=TestPayload(message="test1"))
     assert task2 is not None
 
     response = test_client.post("/test-task-id", json={"message": "test"})
     assert response.status_code == 200
 
 
 def test_delayed_task_error_handling(app, test_client):
-    """Test error handling in delayed routes.
-    
+    """
+    Test error handling in delayed routes.
+
     This test verifies that:
     1. Invalid configurations are caught
     2. Task creation failures are handled
@@ -158,8 +157,9 @@ def test_delayed_task_error_handling(app, test_client):
 
 
 def test_delayed_task_queue_creation(app, test_client):
-    """Test queue auto-creation functionality.
-    
+    """
+    Test queue auto-creation functionality.
+
     This test verifies that:
     1. Queue is created if it doesn't exist
     2. DelayedRouteBuilder handles existing queues

tests/test_examples.py

@@ -1,19 +1,18 @@
-"""Tests for example implementations.
+"""
+Tests for example implementations.
 
 This test module verifies that the example implementations work as expected,
 covering both simple and full examples from the examples directory.
 """
-import os
+
 from unittest.mock import patch
 
 import pytest
-from fastapi import FastAPI
 from fastapi.testclient import TestClient
 from google.protobuf import duration_pb2
 
-from examples.simple.main import app as simple_app
 from examples.full.tasks import app as full_app
-from fastapi_gcp_tasks.utils import emulator_client, queue_path
+from examples.simple.main import app as simple_app
 
 
 @pytest.fixture
@@ -29,8 +28,9 @@ def full_client():
 
 
 def test_simple_example_local_mode(simple_client, monkeypatch):
-    """Test simple example in local mode.
-    
+    """
+    Test simple example in local mode.
+
     This test verifies that:
     1. Emulator client is used in local mode
     2. Tasks are properly queued
@@ -40,23 +40,21 @@ def test_simple_example_local_mode(simple_client, monkeypatch):
     # Ensure we're in local mode
     monkeypatch.setenv("IS_LOCAL", "true")
     monkeypatch.setenv("TASK_LISTENER_BASE_URL", "http://localhost:8000")
-    
+
     # Test trigger endpoint
     response = simple_client.get("/trigger")
     assert response.status_code == 200
     assert response.json() == {"message": "Basic hello task triggered"}
 
     # Test hello task endpoint
-    response = simple_client.post(
-        "/delayed/hello",
-        json={"message": "test message"}
-    )
+    response = simple_client.post("/delayed/hello", json={"message": "test message"})
     assert response.status_code == 200
 
 
 def test_full_example_chained_hooks(full_client, monkeypatch):
-    """Test full example with chained hooks.
-    
+    """
+    Test full example with chained hooks.
+
     This test verifies that:
     1. OIDC and deadline hooks work together
     2. Hook order is preserved
@@ -67,32 +65,25 @@ def test_full_example_chained_hooks(full_client, monkeypatch):
     monkeypatch.setenv("IS_LOCAL", "true")
     monkeypatch.setenv("CLOUD_TASKS_EMULATOR_URL", "http://localhost:8123")
     monkeypatch.setenv("TASK_LISTENER_BASE_URL", "http://localhost:8000")
-    
+
     # Test hello task with chained hooks
-    response = full_client.post(
-        "/delayed/hello",
-        json={"message": "test with hooks"}
-    )
+    response = full_client.post("/delayed/hello", json={"message": "test with hooks"})
     assert response.status_code == 200
 
     # Test fail_twice with retries
     response = full_client.post("/delayed/fail_twice")
     assert response.status_code == 500  # Should fail after 2 retries
 
     # Test scheduled hello with hooks
-    response = full_client.post(
-        "/scheduled/timed_hello",
-        json={"message": "test scheduled with hooks"}
-    )
+    response = full_client.post("/scheduled/timed_hello", json={"message": "test scheduled with hooks"})
     assert response.status_code == 200
-    assert response.json() == {
-        "message": "Scheduled hello task ran with payload: test scheduled with hooks"
-    }
+    assert response.json() == {"message": "Scheduled hello task ran with payload: test scheduled with hooks"}
 
 
 def test_full_example_hook_configuration():
-    """Test hook configuration in full example.
-    
+    """
+    Test hook configuration in full example.
+
     This test verifies that:
     1. OIDC token is properly configured
     2. Deadline duration is set correctly
@@ -101,20 +92,18 @@ def test_full_example_hook_configuration():
     with patch("fastapi_gcp_tasks.hooks.oidc_delayed_hook") as mock_oidc_hook:
         with patch("fastapi_gcp_tasks.hooks.deadline_delayed_hook") as mock_deadline_hook:
             # Import here to trigger hook creation with mocks
-            from examples.full.tasks import DelayedRoute
-            
+
             # Verify OIDC hook was called
             mock_oidc_hook.assert_called_once()
-            
+
             # Verify deadline hook was called with correct duration
-            mock_deadline_hook.assert_called_once_with(
-                duration=duration_pb2.Duration(seconds=1800)
-            )
+            mock_deadline_hook.assert_called_once_with(duration=duration_pb2.Duration(seconds=1800))
 
 
 def test_simple_example_environment_handling(monkeypatch):
-    """Test environment handling in simple example.
-    
+    """
+    Test environment handling in simple example.
+
     This test verifies that:
     1. Local mode uses emulator client
     2. Environment variables are properly handled
@@ -123,21 +112,25 @@ def test_simple_example_environment_handling(monkeypatch):
     # Test local mode
     monkeypatch.setenv("IS_LOCAL", "true")
     from examples.simple.main import client
+
     assert client is not None
-    
+
     # Test non-local mode
     monkeypatch.setenv("IS_LOCAL", "false")
     with patch("examples.simple.main.tasks_v2.CloudTasksClient") as mock_client:
         # Reimport to trigger client creation
         from importlib import reload
+
         import examples.simple.main
+
         reload(examples.simple.main)
         mock_client.assert_called_once()
 
 
 def test_deployed_environment_scheduling(full_client, monkeypatch):
-    """Test scheduling in deployed environment.
-    
+    """
+    Test scheduling in deployed environment.
+
     This test verifies that:
     1. Scheduling only occurs when not local
     2. OIDC tokens are properly used
@@ -150,38 +143,35 @@ def test_deployed_environment_scheduling(full_client, monkeypatch):
         monkeypatch.setenv("IS_LOCAL", "false")
         monkeypatch.setenv("TASK_LISTENER_BASE_URL", "https://example.com")
         monkeypatch.setenv("SCHEDULED_OIDC_TOKEN", "test-token")
-        
+
         # Reload module to trigger scheduling
         from importlib import reload
+
         import examples.full.tasks
+
         reload(examples.full.tasks)
-        
+
         # Verify scheduler client was used
         mock_scheduler.assert_called_once()
-        
+
         # Verify scheduled task creation
         scheduler_instance = mock_scheduler.return_value
         create_job = scheduler_instance.create_job
         assert create_job.called
-        
+
         # Verify job configuration
         job_args = create_job.call_args[0]
         assert len(job_args) == 2  # parent and job args
         job = job_args[1]
-        
+
         # Verify schedule
         assert job.schedule == "*/5 * * * *"
         assert job.time_zone == "Asia/Kolkata"
-        
+
         # Verify OIDC token
         assert job.http_target.oidc_token.service_account_email == "test-token"
-        
+
         # Test endpoint still works
-        response = full_client.post(
-            "/scheduled/timed_hello",
-            json={"message": "test in deployed mode"}
-        )
+        response = full_client.post("/scheduled/timed_hello", json={"message": "test in deployed mode"})
         assert response.status_code == 200
-        assert response.json() == {
-            "message": "Scheduled hello task ran with payload: test in deployed mode"
-        }
+        assert response.json() == {"message": "Scheduled hello task ran with payload: test in deployed mode"}

tests/test_scheduled_route.py

@@ -1,4 +1,5 @@
-"""Tests for ScheduledRouteBuilder functionality.
+"""
+Tests for ScheduledRouteBuilder functionality.
 
 This module verifies the core functionality of ScheduledRouteBuilder, including:
 - Basic scheduled task creation
@@ -64,8 +65,9 @@ async def test_task(payload: TestPayload):
 
 
 def test_scheduled_route_cron_validation(app, test_client):
-    """Test cron schedule validation.
-    
+    """
+    Test cron schedule validation.
+
     This test verifies that:
     1. Valid cron expressions are accepted
     2. Invalid expressions are rejected
@@ -126,8 +128,9 @@ async def test_task(payload: TestPayload):
 
 
 def test_scheduled_route_job_creation(app, test_client):
-    """Test Cloud Scheduler job creation.
-    
+    """
+    Test Cloud Scheduler job creation.
+
     This test verifies that:
     1. Jobs are created with correct settings
     2. Job names are unique