test: Adding Initial Consumer Contract tests for QNexus-Nexus

.github/workflows/ci.yml

@@ -7,6 +7,14 @@ on:
   push:
     branches:
       - main
+  workflow_call:
+    secrets:
+      PACT_BROKER_BASE_URL:
+        required: true
+      PACT_BROKER_USERNAME:
+        required: true
+      PACT_BROKER_PASSWORD:
+        required: true
 env:
   UV_VERSION: "0.6.6"
 
@@ -88,3 +96,8 @@ jobs:
           name: pytest-results-${{ matrix.python-version }}
           path: htmlcov/
 
+  test-contracts:
+    name: Verify Consumer Contracts
+    uses: ./.github/workflows/contract.yml
+    secrets: inherit
+

.github/workflows/contract.yml

@@ -0,0 +1,85 @@
+name: Contract Tests
+
+on:
+  workflow_dispatch:
+    inputs:
+      pr_number:
+        description: "Pull request number (for external PRs)"
+        required: false
+        type: number
+  workflow_call:
+    secrets:
+      PACT_BROKER_BASE_URL:
+        required: true
+      PACT_BROKER_USERNAME:
+        required: true
+      PACT_BROKER_PASSWORD:
+        required: true
+env:
+  UV_VERSION: "0.6.6"
+
+jobs:
+  test-qnexus-consumer:
+    name: Consumer Contract Tests
+    runs-on: ubuntu-24.04
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.12"]
+    env:
+      NEXUS_QA_USER_EMAIL: ${{ secrets.NEXUS_QA_USER_EMAIL }}
+      NEXUS_QA_USER_PASSWORD: ${{ secrets.NEXUS_QA_USER_PASSWORD }}
+      NEXUS_DOMAIN: "qa.myqos.com"
+      NEXUS_STORE_TOKENS: "false"
+      NEXUS_QA_QSYS_DEVICE: ${{ secrets.NEXUS_QA_QSYS_DEVICE }}
+
+    steps:
+      - name: Comment action run link on PR
+        if: ${{ github.event.inputs.pr_number && github.event.inputs.pr_number != '' }}
+        uses: thollander/actions-comment-pull-request@v3
+        with:
+          message: |
+            Contract tests are running on this PR at:
+            ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}.
+          pr-number: ${{ github.event.inputs.pr_number }}
+
+      - name: Checkout code
+        uses: actions/checkout@v4
+
+      - name: Set up uv & venv
+        uses: astral-sh/setup-uv@v5
+        with:
+          version: ${{ env.UV_VERSION }}
+          enable-cache: true
+          python-version: ${{ matrix.python-version }}
+          cache-dependency-glob: uv.lock
+
+      - name: Run contract tests
+        run: |
+          uv run pytest tests/contract/
+
+      - uses: actions/upload-artifact@v4
+        with:
+          name: qnexus-consumer-pacts
+          path: tests/pacts/
+
+  publish-pact-files:
+    runs-on: ubuntu-latest
+    needs: test-qnexus-consumer
+    steps:
+      - uses: actions/checkout@v5
+      - uses: actions/download-artifact@v4
+        with:
+          name: qnexus-consumer-pacts
+          path: tests/pacts/
+      - uses: pactflow/actions/publish-pact-files@v2
+        with:
+          pactfiles: tests/pacts/
+          # version: "1.2.3" # optional, defaults to git sha if not specified
+          # branch: main # optional, defaults to git branch if not specified
+          # tag: main # optional, defaults to not set if not specified
+          tag_with_git_branch: true # optional, defaults to not false if not set. will auto tag with user specified branch, or set to auto detected branch
+          broker_url: ${{ secrets.PACT_BROKER_BASE_URL }}
+        env:
+          PACT_BROKER_USERNAME: ${{ secrets.PACT_BROKER_USERNAME }}
+          PACT_BROKER_PASSWORD: ${{ secrets.PACT_BROKER_PASSWORD }}

DEPENDENCIES.md

@@ -14,3 +14,5 @@
 | `pydantic-settings` | `>=2, <3.0` | Settings management using Pydantic | [https://pypi.org/project/pydantic-settings/](https://pypi.org/project/pydantic-settings/) |
 | `quantinuum-schemas` | `>=7.3, <8.0` | Shared data models for Quantinuum. | [https://github.com/CQCL/quantinuum-schemas](https://github.com/CQCL/quantinuum-schemas) |
 | `hugr` | `>=0.14.0, <1.0.0` | Quantinuum's common representation for quantum programs | [https://github.com/CQCL/hugr/tree/main/hugr-py](https://github.com/CQCL/hugr/tree/main/hugr-py) |
+| `numpy` | `>= 2.2.4` | Fundamental package for array computing in Python | [https://numpy.org](https://numpy.org) |
+| `pact-python` | `>= 3.1.0` | Tool for creating and verifying consumer-driven contracts using the Pact framework. | [https://pact.io](https://pact.io) |

pyproject.toml

@@ -23,6 +23,8 @@ dependencies = [
     "pydantic-settings >=2, <3.0",
     "quantinuum-schemas>=7.3, <8.0",
     "hugr >=0.14.0, <1.0.0",
+    "numpy >= 2.2.4",
+    "pact-python >= 3.1.0",
 ]
 
 [project.optional-dependencies]

scripts/run_unit_tests.sh

@@ -11,6 +11,6 @@ uv run pytest tests/test_auth.py::test_token_refresh_expired
 
 
 echo "Running non-auth tests"
-uv run pytest tests/ -v --ignore=tests/test_auth.py
+uv run pytest tests/ -v --ignore=tests/test_auth.py --ignore=tests/contract/
 
 echo -e "\n🎉 All tests passed successfully!"

tests/contract/conftest.py

@@ -0,0 +1,12 @@
+from pathlib import Path
+from typing import Generator
+from pact import Pact
+
+import pytest
+
+
+@pytest.fixture
+def pact() -> Generator[Pact, None, None]:
+    pact = Pact("QNexus", "Nexus-API").with_specification("V3")
+    yield pact
+    pact.write_file(Path(__file__).parent.parent / "pacts")

tests/contract/projects/test_list_projects_consumer.py

@@ -0,0 +1,108 @@
+from collections.abc import Sequence
+from datetime import datetime
+from typing import Any
+
+import httpx
+from pact import Pact, match
+
+PY_DT_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ"
+DT_STRING = "2025-10-21T11:23:47.489670Z"
+DT_SAMPLE = datetime.strptime(DT_STRING, PY_DT_FORMAT)
+
+PROJECT_DATA = {
+    "id": "550e8400-e29b-41d4-a716-446655440000",
+    "attributes": {
+        "contents_modified": DT_SAMPLE,
+        "archived": False,
+        "name": "project name",
+        "description": "project description",
+        "properties": {},
+        "timestamps": {
+            "created": DT_SAMPLE,
+            "modified": DT_SAMPLE,
+        },
+    },
+}
+
+PROJECT_DATA_MATCHERS = {"id": match.uuid(str(PROJECT_DATA["id"])), "attributes": {}}
+
+
+def _get_matcher_value(v: Any) -> match.AbstractMatcher[Any]:
+    if isinstance(v, datetime):
+        return match.datetime(DT_STRING, PY_DT_FORMAT)
+    else:
+        return match.type(v)
+
+
+assert isinstance(PROJECT_DATA["attributes"], dict)
+assert isinstance(PROJECT_DATA_MATCHERS["attributes"], dict)
+for k, v in PROJECT_DATA["attributes"].items():
+    if not isinstance(v, dict):
+        PROJECT_DATA_MATCHERS["attributes"][k] = _get_matcher_value(v)
+    else:
+        PROJECT_DATA_MATCHERS["attributes"][k] = {}
+        for k2, v2 in v.items():
+            assert isinstance(PROJECT_DATA_MATCHERS["attributes"][k], dict)
+            PROJECT_DATA_MATCHERS["attributes"][k][k2] = _get_matcher_value(v2)
+
+
+PROJECT_RESPONSE = {
+    "data": match.each_like(PROJECT_DATA_MATCHERS),
+    "meta": {
+        "page_number": match.integer(0, min=0),
+        "page_size": match.integer(1, min=0),
+        "total_pages": match.integer(1, min=0),
+        "total_count": match.integer(1, min=0),
+    },
+}
+
+
+def test_list_projects(pact: Pact) -> None:
+    (
+        pact.upon_receiving("a request to list projects")
+        .with_request("GET", "/api/projects/v1beta2")
+        .with_query_parameter("scope", "user")
+        .will_respond_with(200)
+        .with_body(PROJECT_RESPONSE)
+    )
+
+    with pact.serve() as srv:
+        client = client = httpx.Client(base_url=str(srv.url), timeout=None)
+        response = client.get("/api/projects/v1beta2?scope=user")
+        response_data = response.json()["data"]
+
+        # Check that the project response is the correct type
+        assert isinstance(PROJECT_RESPONSE["data"], match.matcher.GenericMatcher)
+        assert isinstance(PROJECT_RESPONSE["data"].value, Sequence)
+
+        # Unlike PactJS, there is no "match.reify" function, so we have to get the sample values ourselves
+        # Confirm that the response has the correct number of elements
+        assert len(response_data) == len(PROJECT_RESPONSE["data"].value)
+
+        # Check attribute values of the first response element
+        assert "id" in response_data[0].keys()
+        assert response_data[0]["id"] == PROJECT_DATA["id"]
+        assert "attributes" in response_data[0].keys()
+        for k, v in response_data[0]["attributes"].items():
+            if not isinstance(v, dict):
+                assert isinstance(PROJECT_DATA["attributes"], dict)
+                _datetime_safe_assert(v, PROJECT_DATA["attributes"][k])
+            else:
+                assert isinstance(PROJECT_DATA["attributes"], dict)
+                assert isinstance(PROJECT_DATA["attributes"][k], dict)
+                # For each k2/v2 in project_data[attributes][k] (str/GenericMatcher)
+                #     response_data[0][attributes][k][k2] == v2.value
+                #     v = response_data[0][attributes][k]
+                for k2, v2 in PROJECT_DATA["attributes"][k].items():
+                    _datetime_safe_assert(v[k2], v2)
+
+
+def _datetime_safe_assert(v1: str | datetime, v2: str | datetime) -> None:
+    """
+    Quick helper function to check if 2 values are the same, regardless of datetime/non-datetime conversions
+    """
+    if isinstance(v1, datetime):
+        v1 = datetime.strftime(v1, PY_DT_FORMAT)
+    if isinstance(v2, datetime):
+        v2 = datetime.strftime(v2, PY_DT_FORMAT)
+    assert v1 == v2