diff --git a/src/lettermint/types.py b/src/lettermint/types.py index 0353a67..25faa05 100644 --- a/src/lettermint/types.py +++ b/src/lettermint/types.py @@ -2,7 +2,7 @@ from __future__ import annotations -from typing import Any, Literal, TypedDict +from typing import Any, Literal, Optional, TypedDict from typing_extensions import NotRequired, Required, TypeAlias @@ -576,6 +576,14 @@ ] WebhookScope: TypeAlias = Literal["team", "project", "route"] WebhookDeliveryModeFilter: TypeAlias = Literal["live", "sandbox", "both"] +WebhookBasicAuthData = TypedDict( + "WebhookBasicAuthData", + { + "username": "Required[str]", + "password": "Required[str]", + }, +) + StoreWebhookData = TypedDict( "StoreWebhookData", { @@ -589,6 +597,7 @@ "route_ids": "NotRequired[list[str]]", "route_id": "NotRequired[str | None]", "delivery_mode_filter": "NotRequired[WebhookDeliveryModeFilter | None]", + "basic_auth": "NotRequired[Optional[WebhookBasicAuthData]]", # noqa: UP045 - Python 3.9 runtime hint resolution. }, ) @@ -779,6 +788,7 @@ "route_ids": "NotRequired[list[str]]", "route_id": "NotRequired[str | None]", "delivery_mode_filter": "NotRequired[WebhookDeliveryModeFilter]", + "basic_auth": "NotRequired[Optional[WebhookBasicAuthData]]", # noqa: UP045 - Python 3.9 runtime hint resolution. }, ) @@ -792,6 +802,7 @@ "route_id": "Required[str | None]", "name": "Required[str]", "url": "Required[str]", + "has_basic_auth": "Required[bool]", "events": "Required[list[str]]", "enabled": "Required[bool]", "include_machine_events": "Required[bool]", @@ -864,6 +875,7 @@ "last_called_at": "Required[str | None]", "created_at": "Required[str]", "updated_at": "Required[str]", + "has_basic_auth": "Required[bool]", }, ) @@ -886,6 +898,7 @@ "created_at": "Required[str]", "updated_at": "Required[str]", "delivery_mode_filter": "Required[WebhookDeliveryModeFilter]", + "has_basic_auth": "Required[bool]", }, ) diff --git a/tests/test_webhook_basic_auth.py b/tests/test_webhook_basic_auth.py new file mode 100644 index 0000000..3d4c779 --- /dev/null +++ b/tests/test_webhook_basic_auth.py @@ -0,0 +1,135 @@ +from __future__ import annotations + +import json +import os +import subprocess +import sys +from pathlib import Path +from typing import get_args, get_origin, get_type_hints + +import pytest +import respx +from httpx import Response +from typing_extensions import NotRequired, Required + +from lettermint import AsyncLettermint, HttpRequestError, Lettermint +from lettermint import types as lm_types + + +@pytest.mark.parametrize( + "state", + [{}, {"basic_auth": {"username": " fixture user ", "password": ""}}, {"basic_auth": None}], +) +@pytest.mark.parametrize("asynchronous", [False, True]) +@respx.mock +@pytest.mark.asyncio +async def test_webhook_credential_states_keep_bearer_auth(state: dict, asynchronous: bool) -> None: + create = respx.post("https://api.lettermint.co/v1/webhooks").mock( + return_value=Response(201, json={"data": {"has_basic_auth": True}}) + ) + update = respx.put("https://api.lettermint.co/v1/webhooks/webhook-id").mock( + return_value=Response(200, json={"data": {"has_basic_auth": True}}) + ) + payload = { + "name": "Fixture", + "url": "https://example.test/hook", + "events": ["message.sent"], + **state, + } + if asynchronous: + async with AsyncLettermint.api("fixture-token") as api: + assert (await api.webhooks.create(payload))["data"]["has_basic_auth"] is True + assert (await api.webhooks.update("webhook-id", state))["data"][ + "has_basic_auth" + ] is True + else: + with Lettermint.api("fixture-token") as sync_api: + assert sync_api.webhooks.create(payload)["data"]["has_basic_auth"] is True + assert sync_api.webhooks.update("webhook-id", state)["data"]["has_basic_auth"] is True + for route, expected in [(create, payload), (update, state)]: + request = route.calls.last.request + assert json.loads(request.content) == expected + assert request.headers["authorization"] == "Bearer fixture-token" + assert "x-lettermint-token" not in request.headers + + +def test_webhook_types_expose_required_read_flag_and_optional_nullable_credentials() -> None: + def field_hint(model: type, field: str): + selected = type( + "SelectedField", (), {"__annotations__": {field: model.__annotations__[field]}} + ) + return get_type_hints(selected, globalns=vars(lm_types), include_extras=True)[field] + + for model in [lm_types.WebhookData, lm_types.WebhookListData, lm_types.WebhookSecretData]: + hint = field_hint(model, "has_basic_auth") + assert get_origin(hint) is Required and get_args(hint) == (bool,) + for model in [lm_types.StoreWebhookData, lm_types.UpdateWebhookData]: + hint = field_hint(model, "basic_auth") + assert get_origin(hint) is NotRequired + assert set(get_args(get_args(hint)[0])) == {lm_types.WebhookBasicAuthData, type(None)} + assert "has_basic_auth" not in model.__annotations__ + credentials: lm_types.WebhookBasicAuthData = {"username": "fixture", "password": ""} + assert credentials["password"] == "" + + +@respx.mock +def test_free_plan_sandbox_keeps_403_response() -> None: + body = { + "error": { + "code": "FEATURE_NOT_AVAILABLE", + "message": "Sandbox mode is available only on paid plans.", + } + } + respx.post("https://api.lettermint.co/v1/send").mock(return_value=Response(403, json=body)) + with Lettermint.email("fixture-token") as email, pytest.raises(HttpRequestError) as caught: + email.from_("from@example.test").to("to@example.test").subject("Fixture").send() + assert caught.value.status_code == 403 + assert caught.value.response_body == body + + +@pytest.mark.parametrize("model", ["WebhookData", "WebhookListData", "WebhookSecretData"]) +def test_old_typed_webhook_fixtures_need_safe_read_flag(tmp_path: Path, model: str) -> None: + value = { + "id": "fixture", + "scope": "route", + "project_ids": [], + "route_ids": [], + "route_id": None, + "name": "Fixture", + "url": "https://example.test/hook", + "events": [], + "enabled": True, + "last_called_at": None, + "created_at": "", + "updated_at": "", + "delivery_mode_filter": "both", + } + if model != "WebhookListData": + value["include_machine_events"] = False + if model == "WebhookSecretData": + value["secret"] = "synthetic-signing-secret" + caller = tmp_path / "old_caller.py" + environment = {**os.environ, "MYPYPATH": str(Path(__file__).resolve().parents[1] / "src")} + command = [ + sys.executable, + "-m", + "mypy", + "--python-version", + "3.10", + "--follow-imports=silent", + "--no-incremental", + "--cache-dir", + str(tmp_path / "mypy-cache"), + str(caller), + ] + caller.write_text(f"from lettermint.types import {model}\nfixture: {model} = {value!r}\n") + old = subprocess.run(command, env=environment, capture_output=True, text=True, check=False) + assert ( + old.returncode == 1 + and f'Missing key "has_basic_auth" for TypedDict "{model}"' in old.stdout + ) + assert "Found 1 error" in old.stdout + value["has_basic_auth"] = False + caller.write_text(f"from lettermint.types import {model}\nfixture: {model} = {value!r}\n") + migrated = subprocess.run(command, env=environment, capture_output=True, text=True, check=False) + assert migrated.returncode == 0, migrated.stdout + migrated.stderr