feat: add test setup with cloud-tasks-emulator

.github/workflows/test.yml

@@ -0,0 +1,57 @@
+name: Tests
+
+on:
+  push:
+    branches: [ master ]
+  pull_request:
+    branches: [ master ]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+
+    services:
+      cloud-tasks-emulator:
+        image: ghcr.io/aertje/cloud-tasks-emulator:latest
+        ports:
+          - 8123:8123
+
+    steps:
+    - uses: actions/checkout@v4
+
+    - name: Set up Python
+      uses: actions/setup-python@v5
+      with:
+        python-version: "3.11"
+
+    - name: Install Poetry
+      uses: snok/install-poetry@v1
+      with:
+        version: "1.8.2"
+
+    - name: Install dependencies
+      run: |
+        poetry install --with test
+        poetry install --with dev
+
+    - name: Run linting
+      run: poetry run sh scripts/lint.sh
+
+    - name: Run type checking
+      run: poetry run mypy fastapi_gcp_tasks
+
+    - name: Run tests
+      run: poetry run pytest --cov=fastapi_gcp_tasks --cov-report=xml
+      env:
+        IS_LOCAL: "true"
+        CLOUD_TASKS_EMULATOR_URL: http://localhost:8123
+        TASK_LISTENER_BASE_URL: http://localhost:8000
+        TASK_PROJECT_ID: test-project
+        TASK_LOCATION: us-central1
+        TASK_QUEUE: test-queue
+
+    - name: Upload coverage reports to Codecov
+      uses: codecov/codecov-action@v4
+      with:
+        file: ./coverage.xml
+        fail_ci_if_error: true

Dockerfile.emulator

@@ -0,0 +1,2 @@
+FROM ghcr.io/aertje/cloud-tasks-emulator:latest
+RUN apk add --no-cache wget socat

docker-compose.yml

@@ -0,0 +1,38 @@
+version: '3'
+services:
+  cloud-tasks-emulator:
+    build:
+      context: .
+      dockerfile: Dockerfile.emulator
+    command: sh -c "/emulator -host 0.0.0.0 -port 8123 -grpc_port 8123 -grpc_host 0.0.0.0 -grpc_insecure -grpc_bind_all -bind_all"
+    ports:
+      - "8123:8123"
+      - "8123:8123/udp"
+    network_mode: "bridge"
+
+    healthcheck:
+      test: ["CMD", "wget", "-q", "--spider", "http://0.0.0.0:8123/v1/projects/test-project/locations/us-central1/queues"]
+      interval: 1s
+      timeout: 3s
+      retries: 30
+  tests:
+    image: python:3.11
+    depends_on:
+      cloud-tasks-emulator:
+        condition: service_healthy
+    volumes:
+      - .:/app
+    working_dir: /app
+    environment:
+      - IS_LOCAL=true
+      - CLOUD_TASKS_EMULATOR_HOST=cloud-tasks-emulator
+      - CLOUD_TASKS_EMULATOR_PORT=8123
+      - CLOUD_TASKS_EMULATOR_URL=http://cloud-tasks-emulator:8123
+      - TASK_LISTENER_BASE_URL=http://localhost:8000
+      - TASK_PROJECT_ID=test-project
+      - TASK_LOCATION=us-central1
+      - TASK_QUEUE=test-queue
+    command: >
+      bash -c "pip install poetry &&
+               poetry install &&
+               poetry run pytest tests/ -v"

docs/tests.md

@@ -0,0 +1,112 @@
+# Testing Guide
+
+## Prerequisites
+- Python 3.11 or higher
+- Poetry for dependency management
+- cloud-tasks-emulator (install from https://github.com/aertje/cloud-tasks-emulator)
+
+## Environment Variables
+Required environment variables for testing:
+```bash
+# Local development flag
+IS_LOCAL=true
+
+# Task emulator settings
+CLOUD_TASKS_EMULATOR_URL=http://localhost:8123
+TASK_LISTENER_BASE_URL=http://localhost:8000
+
+# Task queue settings
+TASK_PROJECT_ID=test-project
+TASK_LOCATION=us-central1
+TASK_QUEUE=test-queue
+```
+
+## Running Tests
+
+1. Install dependencies:
+```bash
+poetry install --with test
+```
+
+2. Start the cloud-tasks-emulator:
+```bash
+cloud-tasks-emulator
+```
+
+3. Run tests:
+```bash
+# Run all tests
+poetry run pytest
+
+# Run with coverage report
+poetry run pytest --cov=fastapi_gcp_tasks
+
+# Run specific test file
+poetry run pytest tests/test_delayed_route.py
+```
+
+## Test Structure
+
+The test suite is organized into several key areas:
+
+### Core Functionality Tests
+- DelayedRouteBuilder
+  * Basic task creation and execution
+  * Queue auto-creation
+  * Task options (countdown, task_id)
+  * Error handling and validation
+
+- ScheduledRouteBuilder
+  * Basic job creation
+  * Cron schedule validation
+  * Time zone handling
+  * Job updates and uniqueness
+
+### Hook and Dependency Tests
+- Task Hooks
+  * OIDC token hooks
+  * Deadline hooks
+  * Chained hooks
+  * Custom hook creation
+- Dependencies
+  * max_retries functionality
+  * CloudTasksHeaders integration
+  * Error propagation
+
+### Example Implementation Tests
+- Simple Example
+  * Local mode functionality
+  * Task queueing
+  * Environment variable handling
+  * Default settings
+
+- Full Example
+  * Chained hook configuration
+  * Retry mechanisms
+  * Scheduled tasks
+  * Environment-specific behavior
+
+### Environment-Specific Tests
+- Local Development
+  * Emulator integration
+  * Default configurations
+  * Environment variable handling
+
+- Deployed Environment
+  * OIDC token integration
+  * Cloud Scheduler integration
+  * Job scheduling
+  * Production settings
+
+## Contributing Tests
+
+When adding new tests:
+1. Follow existing test patterns
+2. Ensure proper type hints are used
+3. Run formatting and linting:
+```bash
+sh scripts/format.sh
+sh scripts/lint.sh
+```
+4. Add appropriate docstrings for complex test cases
+5. Consider edge cases and error scenarios

examples/full/settings.py

@@ -19,7 +19,7 @@
 SCHEDULED_LOCATION = os.getenv("SCHEDULED_LOCATION", default="us-central1")
 TASK_QUEUE = os.getenv("TASK_QUEUE", default="test-queue")
 
-CLOUD_TASKS_EMULATOR_URL = os.getenv("CLOUD_TASKS_EMULATOR_URL", "localhost:8123")
+CLOUD_TASKS_EMULATOR_HOST = os.getenv("CLOUD_TASKS_EMULATOR_HOST", "localhost:8124")
 
 TASK_SERVICE_ACCOUNT = os.getenv(
     "TASK_SERVICE_ACCOUNT",

examples/full/tasks.py

@@ -9,7 +9,6 @@
 # Imports from this repository
 from examples.full.serializer import Payload
 from examples.full.settings import (
-    CLOUD_TASKS_EMULATOR_URL,
     IS_LOCAL,
     SCHEDULED_LOCATION_PATH,
     SCHEDULED_OIDC_TOKEN,
@@ -36,7 +35,7 @@
 
 delayed_client = None
 if IS_LOCAL:
-    delayed_client = emulator_client(host=CLOUD_TASKS_EMULATOR_URL)
+    delayed_client = emulator_client()
 
 DelayedRoute = DelayedRouteBuilder(
     client=delayed_client,

fastapi_gcp_tasks/utils.py

@@ -1,6 +1,8 @@
-# Third Party Imports
+# Standard Library Imports
+import os
 from typing import Any
 
+# Third Party Imports
 import grpc
 from google.api_core.exceptions import AlreadyExists
 from google.cloud import scheduler_v1, tasks_v2
@@ -31,8 +33,9 @@ def ensure_queue(
     """
     # We extract information from the queue path to make the public api simpler
     parsed_queue_path = client.parse_queue_path(path=path)
+    location_args = {k: v for k, v in parsed_queue_path.items() if k in ("project", "location")}
     create_req = tasks_v2.CreateQueueRequest(
-        parent=location_path(**parsed_queue_path),
+        parent=location_path(**location_args),
         queue=tasks_v2.Queue(name=path, **kwargs),
     )
     try:
@@ -41,8 +44,27 @@ def ensure_queue(
         pass
 
 
-def emulator_client(*, host: str = "localhost:8123") -> tasks_v2.CloudTasksClient:
+def emulator_client() -> tasks_v2.CloudTasksClient:
     """Helper function to create a CloudTasksClient from an emulator host."""
-    channel = grpc.insecure_channel(host)
+    host = os.getenv("CLOUD_TASKS_EMULATOR_HOST", "localhost")
+    port = os.getenv("CLOUD_TASKS_EMULATOR_PORT", "8123")
+    target = f"{host}:{port}"
+
+    # Configure DNS resolution for Docker networking
+    options = [
+        ('grpc.enable_http_proxy', 0),
+        ('grpc.enable_retries', 0),
+        ('grpc.max_receive_message_length', -1),
+        ('grpc.max_send_message_length', -1),
+        ('grpc.keepalive_time_ms', 30000),
+        ('grpc.dns_resolver_query_timeout_ms', 1000),
+        ('grpc.dns_resolver_backoff_multiplier', 1.0),
+        ('grpc.dns_resolver_backoff_jitter', 0.0),
+        ('grpc.dns_resolver_backoff_min_seconds', 1),
+        ('grpc.dns_resolver_backoff_max_seconds', 5),
+    ]
+
+    # Create channel with DNS resolution options
+    channel = grpc.insecure_channel(target, options=options)
     transport = transports.CloudTasksGrpcTransport(channel=channel)
     return tasks_v2.CloudTasksClient(transport=transport)