tests/test_tasks.py

"""Cloud Tasks dispatcher (tasks.CloudTasksDispatcher) — env gating + the shared-secret guard.

No google-cloud-tasks, no network: the client is imported lazily, so constructing a disabled
(or even enabled) dispatcher never touches the SDK. The endpoint-level behaviour (routing +
/tasks/quote) is covered in test_api.py against the app.
"""

from __future__ import annotations

from lawn_estimator.tasks import CloudTasksDispatcher

_ENV = ["CLOUD_TASKS_QUEUE", "CLOUD_TASKS_LOCATION", "CLOUD_TASKS_PROJECT",
        "GOOGLE_CLOUD_PROJECT", "TASKS_TARGET_URL", "TASKS_AUTH_TOKEN"]


def _clear(monkeypatch):
    for k in _ENV:
        monkeypatch.delenv(k, raising=False)


def test_disabled_without_env(monkeypatch):
    _clear(monkeypatch)
    d = CloudTasksDispatcher()
    assert d.enabled is False
    # The public /tasks/quote endpoint is CLOSED when no token is configured.
    assert d.verify_request("anything") is False
    assert d.verify_request("") is False


def test_enabled_requires_all_four_env(monkeypatch):
    _clear(monkeypatch)
    monkeypatch.setenv("CLOUD_TASKS_PROJECT", "p")
    monkeypatch.setenv("CLOUD_TASKS_LOCATION", "us-central1")
    monkeypatch.setenv("CLOUD_TASKS_QUEUE", "lawn-batch")
    # target url still missing -> not enabled
    assert CloudTasksDispatcher().enabled is False
    monkeypatch.setenv("TASKS_TARGET_URL", "https://svc.run.app/")
    assert CloudTasksDispatcher().enabled is True


def test_project_falls_back_to_google_cloud_project(monkeypatch):
    _clear(monkeypatch)
    monkeypatch.setenv("GOOGLE_CLOUD_PROJECT", "auto-proj")   # no CLOUD_TASKS_PROJECT
    monkeypatch.setenv("CLOUD_TASKS_LOCATION", "us-central1")
    monkeypatch.setenv("CLOUD_TASKS_QUEUE", "lawn-batch")
    monkeypatch.setenv("TASKS_TARGET_URL", "https://svc.run.app")
    d = CloudTasksDispatcher()
    assert d.enabled is True and d._project == "auto-proj"


def test_verify_request_is_constant_time_exact_match(monkeypatch):
    _clear(monkeypatch)
    monkeypatch.setenv("TASKS_AUTH_TOKEN", "s3cret-token")
    d = CloudTasksDispatcher()
    assert d.verify_request("s3cret-token") is True
    assert d.verify_request("s3cret-toke") is False   # prefix is not enough
    assert d.verify_request("wrong") is False
    assert d.verify_request("") is False