python: Add unit tests for Svix API client (WIP)

Author: anlambert

.github/workflows/python-tests.yml

@@ -0,0 +1,43 @@
+name: Python Tests
+on:
+  push:
+    paths:
+      - "python/**"
+      - "openapi.json"
+  pull_request:
+    paths:
+      - "python/**"
+      - "openapi.json"
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Build svix server image
+        run: docker compose build
+        working-directory: ./server
+
+      - uses: actions/setup-python@v2
+        name: Install Python
+        with:
+          python-version: "3.11"
+
+      - name: Install deps
+        run: |
+          python -m pip install --upgrade pip
+          python -m pip install -r requirements.txt .
+          python -m pip install -r requirements-dev.txt .
+        working-directory: ./python
+
+      - name: Regen openapi libs
+        run: ./scripts/generate_openapi.sh
+        working-directory: ./python
+
+      - name: Check typing on client tests
+        run: mypy tests/test_client.py
+        working-directory: ./python
+
+      - name: Run Python tests
+        run: pytest -sv
+        working-directory: ./python

python/requirements-dev.txt

@@ -1,13 +1,17 @@
 #
-# This file is autogenerated by pip-compile with Python 3.11
+# This file is autogenerated by pip-compile with Python 3.12
 # by the following command:
 #
 #    pip-compile --output-file=requirements-dev.txt requirements.in/development.txt
 #
-anyio==4.4.0
-    # via httpcore
+anyio==4.6.0
+    # via
+    #   httpcore
+    #   httpx
 attrs==24.2.0
-    # via openapi-python-client
+    # via
+    #   openapi-python-client
+    #   pytest-docker
 autoflake==2.3.1
     # via openapi-python-client
 black==24.8.0
@@ -18,6 +22,9 @@ certifi==2024.8.30
     # via
     #   httpcore
     #   httpx
+    #   requests
+charset-normalizer==3.3.2
+    # via requests
 click==8.1.7
     # via
     #   black
@@ -35,6 +42,7 @@ idna==3.10
     # via
     #   anyio
     #   httpx
+    #   requests
 iniconfig==2.0.0
     # via pytest
 isort==5.13.2
@@ -62,7 +70,7 @@ pathspec==0.12.1
     # via black
 pip-tools==7.4.1
     # via -r requirements.in/development.txt
-platformdirs==4.3.3
+platformdirs==4.3.6
     # via black
 pluggy==1.5.0
     # via pytest
@@ -75,12 +83,18 @@ pyproject-hooks==1.1.0
     #   build
     #   pip-tools
 pytest==8.3.3
+    # via
+    #   -r requirements.in/development.txt
+    #   pytest-docker
+pytest-docker==3.1.1
     # via -r requirements.in/development.txt
 python-dateutil==2.9.0.post0
     # via openapi-python-client
 pyyaml==6.0.2
     # via openapi-python-client
-ruff==0.6.5
+requests==2.32.3
+    # via -r requirements.in/development.txt
+ruff==0.6.8
     # via -r requirements.in/development.txt
 shellingham==1.5.4
     # via openapi-python-client
@@ -98,6 +112,8 @@ typing-extensions==4.12.2
     #   mypy
     #   pydantic
     #   typer
+urllib3==2.2.3
+    # via requests
 wheel==0.44.0
     # via pip-tools
 

python/requirements.in/development.txt

@@ -5,3 +5,5 @@ pytest
 httpx>=0.23.0
 openapi-python-client>=0.14.1,<0.15 # I think version 0.15 is now dangerous for us? https://github.com/openapi-generators/openapi-python-client/pull/775#issuecomment-1646977834
 jinja2>=3.1.3
+pytest-docker
+requests

python/tests/conftest.py

@@ -0,0 +1,112 @@
+import os
+import shutil
+from subprocess import CalledProcessError, check_output
+
+import pytest
+import requests
+from requests.adapters import HTTPAdapter
+from svix.api import Svix, SvixOptions
+from urllib3.util.retry import Retry
+
+SVIX_ORG_ID = "org_svix_python_tests"
+
+
+def pytest_collection_modifyitems(config, items):
+    """Client tests require docker compose (v2 or v1) so skip them if it is not
+    installed on host."""
+    skipper = None
+    if shutil.which("docker") is None:
+        skipper = pytest.mark.skip(reason="skipping test as docker command is missing")
+    else:
+        docker_compose_available = False
+        try:
+            # check if docker compose v2 if available
+            check_output(["docker", "compose", "version"])
+            docker_compose_available = True
+        except CalledProcessError:
+            # check if docker compose v1 if available
+            docker_compose_available = shutil.which("docker-compose") is not None
+        finally:
+            if not docker_compose_available:
+                skipper = pytest.mark.skip(
+                    reason="skipping test as docker compose is missing"
+                )
+    if skipper is not None:
+        for item in items:
+            if item.module.__name__ == "tests.test_client":
+                item.add_marker(skipper)
+
+
+@pytest.fixture(scope="session")
+def docker_compose_command():
+    try:
+        # use docker compose v2 if available
+        check_output(["docker", "compose", "version"])
+        return "docker compose"
+    except Exception:
+        # fallback on v1 otherwise
+        return "docker-compose"
+
+
+@pytest.fixture(scope="session")
+def docker_compose_file():
+    return [
+        os.path.join(os.path.dirname(__file__), "../../server/docker-compose.yml"),
+        os.path.join(
+            os.path.dirname(__file__), "../../server/docker-compose.override.yml"
+        ),
+    ]
+
+
+@pytest.fixture(scope="session")
+def docker_compose(docker_services):
+    return docker_services._docker_compose
+
+
+@pytest.fixture(scope="session")
+def svix_server_url(docker_services):
+    # svix server container exposes a free port to the docker host,
+    # we use the docker network gateway IP in case the tests are also
+    # executed in a container
+    svix_server_port = docker_services.port_for("backend", 8071)
+    return f"http://172.17.0.1:{svix_server_port}"
+
+
+@pytest.fixture(autouse=True, scope="session")
+def svix_server(svix_server_url):
+    """Spawn a Svix server for the tests session using docker compose"""
+    # wait for the svix backend service to be up and responding
+    request_session = requests.Session()
+    retries = Retry(total=10, backoff_factor=0.1, status_forcelist=[500, 502, 503, 504])
+    request_session.mount("http://", HTTPAdapter(max_retries=retries))
+    api_url = f"{svix_server_url}/api/v1/health/"
+    response = request_session.get(api_url)
+    assert response
+
+
+@pytest.fixture(autouse=True)
+def svix_wiper(docker_compose):
+    """Ensure stateless tests"""
+    yield
+    # wipe svix database after each test to ensure stateless tests
+    docker_compose.execute(
+        f"exec -T backend svix-server wipe --yes-i-know-what-im-doing {SVIX_ORG_ID}"
+    )
+
+
+@pytest.fixture(scope="session")
+def svix_api(svix_server_url, docker_compose):
+    # generate bearer token to authorize communication with the svix server
+    exec_output = docker_compose.execute(
+        f"exec -T backend svix-server jwt generate {SVIX_ORG_ID}"
+    )
+    svix_auth_token = (
+        exec_output.decode()
+        .replace("Token (Bearer): ", "")
+        .replace("\r", "")
+        .replace("\n", "")
+    )
+    return Svix(
+        svix_auth_token,
+        SvixOptions(server_url=svix_server_url),
+    )

python/tests/test_client.py

@@ -0,0 +1,146 @@
+import itertools
+import uuid
+from typing import Any, Dict, Optional
+
+import pytest
+from svix.api import (
+    ApplicationIn,
+    ApplicationOut,
+    EndpointIn,
+    EndpointOut,
+    EventTypeIn,
+    EventTypeOut,
+    Svix,
+)
+
+
+def _gen_uuid(name: str) -> str:
+    return str(uuid.uuid5(uuid.NAMESPACE_DNS, name))
+
+
+@pytest.fixture
+def svix_app_name() -> str:
+    return "svix_python_tests"
+
+
+@pytest.fixture
+def event_type_schema() -> Dict[str, Any]:
+    return {
+        "type": "object",
+        "title": "event.test",
+        "description": "A dummy event type",
+        "properties": {
+            "value": {
+                "type": "string",
+                "description": "A simple string value",
+            }
+        },
+        "required": ["value"],
+    }
+
+
+@pytest.fixture
+def endpoint_url() -> str:
+    return "http://localhost/webhook/receiver"
+
+
+def create_svix_app(
+    svix_api: Svix, svix_app_name: str, svix_app_uid: str
+) -> ApplicationOut:
+    return svix_api.application.get_or_create(
+        ApplicationIn(name=svix_app_name, uid=svix_app_uid)
+    )
+
+
+def create_svix_event_type(
+    svix_api: Svix, event_type_schema: Dict[str, Any]
+) -> EventTypeOut:
+    return svix_api.event_type.create(
+        EventTypeIn(
+            name=event_type_schema["title"],
+            description=event_type_schema["description"],
+            schemas={"1": event_type_schema},
+        )
+    )
+
+
+def create_svix_endpoint(
+    svix_api: Svix,
+    app_uid: str,
+    event_type_name: str,
+    endpoint_url: str,
+    endpoint_uid: str,
+    channel: Optional[str] = None,
+    metadata: Optional[Dict[str, Any]] = None,
+    secret: Optional[str] = None,
+) -> EndpointOut:
+    return svix_api.endpoint.create(
+        app_uid,
+        EndpointIn(
+            url=endpoint_url,
+            uid=endpoint_uid,
+            version=1,
+            filter_types=[event_type_name],
+            channels=[channel] if channel else None,
+            metadata=metadata,  # type: ignore[arg-type]
+            secret=secret,
+        ),
+    )
+
+
+def test_svix_application_create(svix_api: Svix, svix_app_name: str) -> None:
+    svix_app_uid = _gen_uuid(svix_app_name)
+    app = create_svix_app(svix_api, svix_app_name, svix_app_uid)
+    assert app.name == svix_app_name
+    assert app.uid == svix_app_uid
+
+
+def test_svix_event_type_create(
+    svix_api: Svix, event_type_schema: Dict[str, Any]
+) -> None:
+    event_type = create_svix_event_type(svix_api, event_type_schema)
+    assert event_type.name == event_type_schema["title"]
+    assert event_type.description == event_type_schema["description"]
+    assert event_type.schemas == {"1": event_type_schema}
+
+
+@pytest.mark.parametrize(
+    "with_channel,with_metadata,with_secret",
+    list(itertools.product([False, True], repeat=3)),
+)
+def test_svix_endpoint_create(
+    svix_api: Svix,
+    svix_app_name: str,
+    event_type_schema: Dict[str, Any],
+    endpoint_url: str,
+    with_channel: bool,
+    with_metadata: bool,
+    with_secret: bool,
+) -> None:
+    svix_app_uid = _gen_uuid(svix_app_name)
+    app = create_svix_app(svix_api, svix_app_name, svix_app_uid)
+    event_type = create_svix_event_type(svix_api, event_type_schema)
+    endpoint_uid = _gen_uuid(endpoint_url)
+    channel = "test" if with_channel else None
+    metadata = {"test": "test"} if with_metadata else None
+    secret = "whsec_" + "e" * 32 if with_secret else None
+    assert app.uid
+    endpoint = create_svix_endpoint(
+        svix_api,
+        app.uid,
+        event_type.name,
+        endpoint_url,
+        endpoint_uid,
+        channel,
+        metadata,
+        secret,
+    )
+    assert endpoint.url == endpoint_url
+    assert endpoint.uid == endpoint_uid
+    assert endpoint.filter_types == [event_type.name]
+    if with_channel:
+        assert endpoint.channels == [channel]
+    if with_metadata:
+        assert endpoint.metadata == metadata
+    if with_secret:
+        assert svix_api.endpoint.get_secret(app.uid, endpoint_uid).key == secret