Python client tests

.github/workflows/python-tests.yml

@@ -0,0 +1,49 @@
+name: python-tests
+
+on:
+  pull_request:
+    branches:
+      - main
+  push:
+    branches:
+      - main
+
+permissions:
+  contents: read
+
+jobs:
+  test:
+    name: Python (${{ matrix.python-version }})
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.11", "3.12", "3.13"]
+
+    steps:
+      - name: Check out code
+        uses: actions/checkout@v7
+
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v6
+        with:
+          python-version: ${{ matrix.python-version }}
+
+      - name: Install buf
+        uses: bufbuild/buf-setup-action@v1
+        with:
+          version: "1.71.0"
+
+      - name: Generate protobuf Python code (including buf/validate)
+        run: |
+          cd proto
+          buf generate
+
+      - name: Install test dependencies
+        run: |
+          pip install connectrpc>=0.10.0 protovalidate werkzeug pytest
+
+      - name: Run Python client tests
+        run: |
+          PYTHONPATH="$GITHUB_WORKSPACE/python" python -m pytest python/metalstack/client/tests/ -v

python/metalstack/client/tests/conftest.py

@@ -0,0 +1,172 @@
+from collections.abc import Iterator
+from typing import Any
+
+import pyqwest
+import pytest
+from connectrpc.request import RequestContext
+from pyqwest.testing import WSGITransport
+
+from metalstack.api.v2 import ip_pb2
+from metalstack.api.v2 import version_pb2
+from metalstack.api.v2.ip_connect import (
+    IPServiceWSGIApplication,
+)
+from metalstack.api.v2.version_connect import (
+    VersionServiceWSGIApplication,
+)
+from metalstack.client.client import Client
+from metalstack.infra.v2 import bmc_pb2
+from metalstack.infra.v2 import component_pb2 as infra_component_pb2
+from metalstack.infra.v2.bmc_connect import (
+    BMCServiceWSGIApplication,
+)
+from metalstack.infra.v2.component_connect import (
+    ComponentServiceWSGIApplication,
+)
+
+
+class MockVersionService:
+    def __init__(self, version: str = "1.0"):
+        self._version = version
+
+    def get(
+        self, request: version_pb2.VersionServiceGetRequest, ctx: RequestContext
+    ) -> version_pb2.VersionServiceGetResponse:
+        return version_pb2.VersionServiceGetResponse(
+            version=version_pb2.Version(version=self._version)
+        )
+
+
+class MockIPService:
+    def __init__(self):
+        self.received_auth: str | None = None
+
+    def get(
+        self, request: ip_pb2.IPServiceGetRequest, ctx: RequestContext
+    ) -> ip_pb2.IPServiceGetResponse:
+        self.received_auth = ctx.request_headers().get("authorization")
+        return ip_pb2.IPServiceGetResponse(
+            ip=ip_pb2.IP(ip=request.ip, project=request.project)
+        )
+
+    def create(
+        self, request: ip_pb2.IPServiceCreateRequest, ctx: RequestContext
+    ) -> ip_pb2.IPServiceCreateResponse:
+        return ip_pb2.IPServiceCreateResponse()
+
+    def update(
+        self, request: ip_pb2.IPServiceUpdateRequest, ctx: RequestContext
+    ) -> ip_pb2.IPServiceUpdateResponse:
+        return ip_pb2.IPServiceUpdateResponse()
+
+    def list(
+        self, request: ip_pb2.IPServiceListRequest, ctx: RequestContext
+    ) -> ip_pb2.IPServiceListResponse:
+        return ip_pb2.IPServiceListResponse()
+
+    def delete(
+        self, request: ip_pb2.IPServiceDeleteRequest, ctx: RequestContext
+    ) -> ip_pb2.IPServiceDeleteResponse:
+        return ip_pb2.IPServiceDeleteResponse()
+
+
+class MockBMCService:
+    def __init__(self):
+        self.received_auth: str | None = None
+
+    def update_b_m_c_info(
+        self, request: bmc_pb2.UpdateBMCInfoRequest, ctx: RequestContext
+    ) -> bmc_pb2.UpdateBMCInfoResponse:
+        self.received_auth = ctx.request_headers().get("authorization")
+        return bmc_pb2.UpdateBMCInfoResponse()
+
+    def wait_for_b_m_c_command(
+        self, request: bmc_pb2.WaitForBMCCommandRequest, ctx: RequestContext
+    ) -> Iterator[bmc_pb2.WaitForBMCCommandResponse]:
+        self.received_auth = ctx.request_headers().get("authorization")
+        yield bmc_pb2.WaitForBMCCommandResponse(uuid="test-uuid", command_id="cmd-1")
+
+    def b_m_c_command_done(
+        self, request: bmc_pb2.BMCCommandDoneRequest, ctx: RequestContext
+    ) -> bmc_pb2.BMCCommandDoneResponse:
+        return bmc_pb2.BMCCommandDoneResponse()
+
+
+class MockComponentService:
+    def __init__(self):
+        self.pings: list[infra_component_pb2.ComponentServicePingRequest] = []
+
+    def ping(
+        self,
+        request: infra_component_pb2.ComponentServicePingRequest,
+        ctx: RequestContext,
+    ) -> infra_component_pb2.ComponentServicePingResponse:
+        self.pings.append(
+            infra_component_pb2.ComponentServicePingRequest(
+                type=request.type,
+                identifier=request.identifier,
+                started_at=request.started_at,
+                interval=request.interval,
+                version=request.version,
+            )
+        )
+        return infra_component_pb2.ComponentServicePingResponse()
+
+
+def _build_combined_wsgi_app(services: dict[str, Any]):
+    """Build a combined WSGI app that dispatches by path prefix."""
+
+    def application(environ, start_response):
+        path = environ.get("PATH_INFO", "/")
+        for prefix, app in services.items():
+            if path.startswith(prefix):
+                return app(environ, start_response)
+        from werkzeug.exceptions import NotFound
+
+        return NotFound()(environ, start_response)
+
+    return application
+
+
+@pytest.fixture
+def mock_version_service():
+    return MockVersionService()
+
+
+@pytest.fixture
+def mock_ip_service():
+    return MockIPService()
+
+
+@pytest.fixture
+def mock_bmc_service():
+    return MockBMCService()
+
+
+@pytest.fixture
+def mock_component_service():
+    return MockComponentService()
+
+
+@pytest.fixture
+def test_client(
+    mock_version_service,
+    mock_ip_service,
+    mock_bmc_service,
+    mock_component_service,
+):
+    """Create a Client with WSGI-backed transport for testing."""
+    services = {
+        "/metalstack.api.v2.VersionService": VersionServiceWSGIApplication(
+            mock_version_service
+        ),
+        "/metalstack.api.v2.IPService": IPServiceWSGIApplication(mock_ip_service),
+        "/metalstack.infra.v2.BMCService": BMCServiceWSGIApplication(mock_bmc_service),
+        "/metalstack.infra.v2.ComponentService": ComponentServiceWSGIApplication(
+            mock_component_service
+        ),
+    }
+    transport = WSGITransport(_build_combined_wsgi_app(services))
+    c = Client(baseurl="http://test", timeout=10)
+    c._client = pyqwest.SyncClient(transport=transport)
+    return c

python/metalstack/client/tests/test_client.py

@@ -0,0 +1,122 @@
+"""Tests mirroring go/client/client_test.go.
+
+Test_Client -> test_client_apiv2_version_get
+Test_ClientInterceptors -> test_client_auth_header_unary, test_client_auth_header_streaming
+"""
+
+import pyqwest
+import pytest
+from pyqwest.testing import WSGITransport
+
+from metalstack.api.v2 import ip_pb2
+from metalstack.api.v2 import version_pb2
+from metalstack.api.v2.ip_connect import (
+    IPServiceClientSync,
+    IPServiceWSGIApplication,
+)
+from metalstack.api.v2.version_connect import (
+    VersionServiceClientSync,
+    VersionServiceWSGIApplication,
+)
+from metalstack.client.client import Client
+from metalstack.infra.v2 import bmc_pb2
+from metalstack.infra.v2.bmc_connect import (
+    BMCServiceClientSync,
+    BMCServiceWSGIApplication,
+)
+from metalstack.infra.v2 import component_pb2 as infra_component_pb2
+from metalstack.infra.v2.component_connect import (
+    ComponentServiceClientSync,
+    ComponentServiceWSGIApplication,
+)
+
+from .conftest import (
+    MockBMCService,
+    MockComponentService,
+    MockIPService,
+    MockVersionService,
+    _build_combined_wsgi_app,
+)
+
+
+class TestClient:
+    """Mirrors Test_Client from client_test.go.
+
+    Tests client instantiation and basic RPC calls through the generated Client wrapper.
+    """
+
+    def test_client_creates_service_groups(self, test_client):
+        """Client exposes adminv2, apiv2, infrav2 service groups."""
+        assert test_client.apiv2() is not None
+        assert test_client.infrav2() is not None
+
+    def test_client_apiv2_version_get(self, test_client, mock_version_service):
+        """Version.Get returns the expected version through the Client wrapper."""
+        resp = test_client.apiv2().version().get(
+            request=version_pb2.VersionServiceGetRequest()
+        )
+        assert resp is not None
+        assert resp.version.version == "1.0"
+
+    def test_client_apiv2_version_get_custom_version(self, mock_version_service):
+        """Client can be configured with a custom version response."""
+        mock_version_service._version = "2.5.3"
+        vs = MockVersionService(version="2.5.3")
+        ips = MockIPService()
+        bmc = MockBMCService()
+        cs = MockComponentService()
+
+        services = {
+            "/metalstack.api.v2.VersionService": VersionServiceWSGIApplication(vs),
+            "/metalstack.api.v2.IPService": IPServiceWSGIApplication(ips),
+            "/metalstack.infra.v2.BMCService": BMCServiceWSGIApplication(bmc),
+            "/metalstack.infra.v2.ComponentService": ComponentServiceWSGIApplication(cs),
+        }
+        transport = WSGITransport(_build_combined_wsgi_app(services))
+        c = Client(baseurl="http://test", timeout=10)
+        c._client = pyqwest.SyncClient(transport=transport)
+
+        resp = c.apiv2().version().get(
+            request=version_pb2.VersionServiceGetRequest()
+        )
+        assert resp.version.version == "2.5.3"
+
+
+class TestClientInterceptors:
+    """Mirrors Test_ClientInterceptors from client_test.go.
+
+    Tests that auth headers are correctly passed for both unary and streaming calls.
+    """
+
+    def test_client_auth_header_unary(self, test_client, mock_bmc_service):
+        """Unary call has Authorization header set."""
+        token = "test-bearer-token"
+        resp = test_client.infrav2().bmc().update_b_m_c_info(
+            request=bmc_pb2.UpdateBMCInfoRequest(),
+            headers={"authorization": f"Bearer {token}"},
+        )
+        assert resp is not None
+        assert mock_bmc_service.received_auth == f"Bearer {token}"
+
+    def test_client_auth_header_streaming(self, test_client, mock_bmc_service):
+        """Streaming call has Authorization header set."""
+        token = "test-stream-token"
+        stream = test_client.infrav2().bmc().wait_for_b_m_c_command(
+            request=bmc_pb2.WaitForBMCCommandRequest(),
+            headers={"authorization": f"Bearer {token}"},
+        )
+        assert stream is not None
+        for cmd in stream:
+            assert cmd.uuid == "test-uuid"
+        assert mock_bmc_service.received_auth == f"Bearer {token}"
+
+    def test_client_auth_header_ip_get(self, test_client, mock_ip_service):
+        """IP Get call has Authorization header set."""
+        token = "ip-service-token"
+        resp = test_client.apiv2().ip().get(
+            request=ip_pb2.IPServiceGetRequest(ip="10.0.0.1"),
+            headers={"authorization": f"Bearer {token}"},
+        )
+        assert resp is not None
+        assert resp.ip.ip == "10.0.0.1"
+        assert mock_ip_service.received_auth == f"Bearer {token}"