Add basic integration tests

Use pytest-docker to run docker-compose as a pytest fixture, bringing up Cadence, Cassandra, and statsd. If possible I'd like to switch to using SQLite in the future, but I need to explore that more. For the moment this is similar to what the Java and Go clients do.

Unlike the Java and Go clients we orchestrate docker via the testing framework, rather than running the tests themselves via docker. This seems much easier to use and debug. With a bit of tinkering we should have a lot more flexibility to do things like running tests against an already running set of services to speed up iteration time.

Additionally change the pytest `asyncio_mode` to `auto` so that async tests and fixtures are automatically run in an eventloop. One major issue with the current version of pytest-asyncio that we're using is its limited control over the eventloop used for running tests/fixtures. It's currently running each in their own eventloop, which means you can't pass certain objects between them, such as GRPC's async channel objects.

Author: natemort

.github/workflows/ci_checks.yml

@@ -73,4 +73,24 @@ jobs:
           uv sync --extra dev
       - name: Run unit tests
         run: |
-          uv run python -m pytest tests/ -v
+          uv run pytest -v
+
+  integration_test:
+    name: Integration Tests
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          submodules: true
+      - name: Set up Python
+        uses: actions/setup-python@v4
+        with:
+          python-version: "3.13"
+      - name: Set up uv
+        uses: astral-sh/setup-uv@v1
+      - name: Install dependencies
+        run: |
+          uv sync --extra dev
+      - name: Run unit tests
+        run: |
+          uv run pytest -v --integration-tests

cadence/_internal/rpc/yarpc.py

@@ -44,7 +44,8 @@ def _replace_details(self, client_call_details: ClientCallDetails) -> ClientCall
 
         return _ClientCallDetails(
             method=client_call_details.method,
-            timeout=client_call_details.timeout,
+            # YARPC seems to require a TTL value
+            timeout=client_call_details.timeout or 60.0,
             metadata=metadata,
             credentials=client_call_details.credentials,
             wait_for_ready=client_call_details.wait_for_ready,

cadence/client.py

@@ -6,6 +6,7 @@
 
 from cadence._internal.rpc.error import CadenceErrorInterceptor
 from cadence._internal.rpc.yarpc import YarpcMetadataInterceptor
+from cadence.api.v1.service_domain_pb2_grpc import DomainAPIStub
 from cadence.api.v1.service_worker_pb2_grpc import WorkerAPIStub
 from grpc.aio import Channel, ClientInterceptor, secure_channel, insecure_channel
 from cadence.data_converter import DataConverter, DefaultDataConverter
@@ -39,6 +40,7 @@ def __init__(self, **kwargs: Unpack[ClientOptions]) -> None:
         self._options = _validate_and_copy_defaults(ClientOptions(**kwargs))
         self._channel = _create_channel(self._options)
         self._worker_stub = WorkerAPIStub(self._channel)
+        self._domain_stub = DomainAPIStub(self._channel)
 
     @property
     def data_converter(self) -> DataConverter:
@@ -52,13 +54,26 @@ def domain(self) -> str:
     def identity(self) -> str:
         return self._options["identity"]
 
+    @property
+    def domain_stub(self) -> DomainAPIStub:
+        return self._domain_stub
+
     @property
     def worker_stub(self) -> WorkerAPIStub:
         return self._worker_stub
 
+    async def ready(self) -> None:
+        await self._channel.channel_ready()
+
     async def close(self) -> None:
         await self._channel.close()
 
+    async def __aenter__(self) -> 'Client':
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
+        await self.close()
+
 def _validate_and_copy_defaults(options: ClientOptions) -> ClientOptions:
     if "target" not in options:
         raise ValueError("target must be specified")

cadence/sample/client_example.py

@@ -6,9 +6,9 @@
 
 
 async def main():
-    client = Client(target="localhost:7833", domain="foo")
-    worker = Worker(client, "task_list", Registry())
-    await worker.run()
+    async with Client(target="localhost:7833", domain="foo") as client:
+        worker = Worker(client, "task_list", Registry())
+        await worker.run()
 
 if __name__ == '__main__':
     asyncio.run(main())

pyproject.toml

@@ -44,6 +44,7 @@ dev = [
     "flake8>=6.0.0",
     "mypy>=1.0.0",
     "pre-commit>=3.0.0",
+    "pytest-docker>=3.2.3",
 ]
 docs = [
     "sphinx>=6.0.0",
@@ -138,14 +139,14 @@ ignore_missing_imports = true
 
 [tool.pytest.ini_options]
 minversion = "7.0"
-addopts = "-ra -q --strict-markers --strict-config"
+addopts = "-ra -q --strict-markers --strict-config --import-mode=importlib"
+asyncio_mode = "auto"
 testpaths = ["tests"]
 python_files = ["test_*.py", "*_test.py"]
 python_classes = ["Test*"]
 python_functions = ["test_*"]
 markers = [
     "slow: marks tests as slow (deselect with '-m \"not slow\"')",
-    "integration: marks tests as integration tests",
     "unit: marks tests as unit tests",
     "asyncio: marks tests as async tests",
 ]

tests/conftest.py

@@ -0,0 +1,8 @@
+
+
+ENABLE_INTEGRATION_TESTS = "--integration_tests"
+
+# Need to define the option in the root conftest.py file
+def pytest_addoption(parser):
+    parser.addoption(ENABLE_INTEGRATION_TESTS, action="store_true",
+                     help="enables running integration tests, which rely on docker and docker-compose")

tests/integration_tests/conftest.py

@@ -0,0 +1,47 @@
+import asyncio
+import os
+from datetime import timedelta
+
+import pytest
+
+from google.protobuf.duration import from_timedelta
+from pytest_docker import Services
+
+from cadence.api.v1.service_domain_pb2 import RegisterDomainRequest
+from cadence.client import ClientOptions
+from tests.conftest import ENABLE_INTEGRATION_TESTS
+from tests.integration_tests.helper import CadenceHelper, DOMAIN_NAME
+
+# Run tests in this directory and lower only if integration tests are enabled
+def pytest_runtest_setup(item):
+    if not item.config.getoption(ENABLE_INTEGRATION_TESTS):
+        pytest.skip(f"{ENABLE_INTEGRATION_TESTS} not enabled")
+
+@pytest.fixture(scope="session")
+def docker_compose_file(pytestconfig):
+    return os.path.join(str(pytestconfig.rootdir), "tests", "integration_tests", "docker-compose.yml")
+
+@pytest.fixture(scope="session")
+def client_options(docker_ip: str, docker_services: Services) -> ClientOptions:
+    return ClientOptions(
+        domain=DOMAIN_NAME,
+        target=f'{docker_ip}:{docker_services.port_for("cadence", 7833)}',
+    )
+
+# We can't pass around Client objects between tests/fixtures without changing our pytest-asyncio version
+# to ensure that they use the same event loop.
+# Instead, we can wait for the server to be ready, create the common domain, and then provide a helper capable
+# of creating additional clients within each test as needed
+@pytest.fixture(scope="session")
+async def helper(client_options: ClientOptions) -> CadenceHelper:
+    helper = CadenceHelper(client_options)
+    async with helper.client() as client:
+        # It takes around a minute for the Cadence server to start up with Cassandra
+        async with asyncio.timeout(120):
+            await client.ready()
+
+        await client.domain_stub.RegisterDomain(RegisterDomainRequest(
+            name=DOMAIN_NAME,
+            workflow_execution_retention_period=from_timedelta(timedelta(days=1)),
+        ))
+    return CadenceHelper(client_options)

tests/integration_tests/docker-compose.yml

@@ -0,0 +1,47 @@
+version: "3.5"
+
+services:
+  cassandra:
+    image: cassandra:4.1.3
+    ports:
+      - "9042:9042"
+    networks:
+      services-network:
+        aliases:
+          - cassandra
+
+  statsd:
+    image: hopsoft/graphite-statsd
+    ports:
+      - "8080:80"
+      - "2003:2003"
+      - "8125:8125"
+      - "8126:8126"
+    networks:
+      services-network:
+        aliases:
+          - statsd
+
+  cadence:
+    image: ubercadence/server:master-auto-setup
+    ports:
+      - "7933:7933"
+      - "7833:7833"
+      - "7934:7934"
+      - "7935:7935"
+      - "7939:7939"
+    environment:
+      - "CASSANDRA_SEEDS=cassandra"
+      - "STATSD_ENDPOINT=statsd:8125"
+      - "DYNAMIC_CONFIG_FILE_PATH=config/dynamicconfig/development.yaml"
+    depends_on:
+      - cassandra
+      - statsd
+    networks:
+      services-network:
+        aliases:
+          - cadence
+networks:
+  services-network:
+    name: services-network
+    driver: bridge

tests/integration_tests/helper.py

@@ -0,0 +1,11 @@
+from cadence.client import ClientOptions, Client
+
+DOMAIN_NAME = "test-domain"
+
+
+class CadenceHelper:
+    def __init__(self, options: ClientOptions):
+        self.options = options
+
+    def client(self):
+        return Client(**self.options)

tests/integration_tests/test_client.py

@@ -0,0 +1,18 @@
+import pytest
+
+from cadence.api.v1.service_domain_pb2 import DescribeDomainRequest, DescribeDomainResponse
+from cadence.error import EntityNotExistsError
+from tests.integration_tests.helper import CadenceHelper, DOMAIN_NAME
+
+
+@pytest.mark.usefixtures("helper")
+async def test_domain_exists(helper: CadenceHelper):
+    async with helper.client() as client:
+        response: DescribeDomainResponse = await client.domain_stub.DescribeDomain(DescribeDomainRequest(name=DOMAIN_NAME))
+        assert response.domain.name == DOMAIN_NAME
+
+@pytest.mark.usefixtures("helper")
+async def test_domain_not_exists(helper: CadenceHelper):
+    with pytest.raises(EntityNotExistsError):
+        async with helper.client() as client:
+            await client.domain_stub.DescribeDomain(DescribeDomainRequest(name="unknown-domain"))