DEV-14971: Python SDK storage-service compatibility unit tests

pyproject.toml

@@ -56,4 +56,10 @@ module = "importlib_metadata"
 ignore_missing_imports = true
 
 [dependency-groups]
-dev = ["pytest>=8.4.2", "pytest-mock>=3.15.1"]
+dev = [
+    "pytest>=8.4.2",
+    "pytest-asyncio>0.21",
+    "pytest-mock>=3.15.1",
+    "requests-mock>=1.8.0",
+    "msgpack>=0.5.6",
+]

tests/integration/queries/test_gallery.py

@@ -77,9 +77,9 @@ def test_list_gallery_pagination(indico):
     if len(blueprints) >= 2:
         # Check that we have unique blueprints
         blueprint_ids = [bp.id for bp in blueprints]
-        assert len(blueprint_ids) == len(
-            set(blueprint_ids)
-        ), "Duplicate blueprints found in pagination"
+        assert len(blueprint_ids) == len(set(blueprint_ids)), (
+            "Duplicate blueprints found in pagination"
+        )
 
 
 def test_get_gallery_tags(indico):

tests/unit/test_release_version_pattern.py

@@ -0,0 +1,35 @@
+import re
+from pathlib import Path
+
+try:  # pragma: no cover - exercised indirectly on Python <3.11 tox envs
+    import tomllib
+except ModuleNotFoundError:  # pragma: no cover
+    import tomli as tomllib
+
+
+def _load_release_pattern() -> str:
+    pyproject_path = Path(__file__).resolve().parents[2] / "pyproject.toml"
+    with pyproject_path.open("rb") as f:
+        pyproject = tomllib.load(f)
+    return pyproject["tool"]["uv-dynamic-versioning"]["pattern"]
+
+
+def test_release_pattern_matches_standard_tag() -> None:
+    pattern = re.compile(_load_release_pattern())
+    match = pattern.match("7.9.0")
+    assert match is not None
+    assert match.groupdict() == {"base": "7.9.0", "stage": None, "revision": None}
+
+
+def test_release_pattern_matches_hyphen_prerelease_tag() -> None:
+    pattern = re.compile(_load_release_pattern())
+    match = pattern.match("7.9.0-rc1")
+    assert match is not None
+    assert match.groupdict() == {"base": "7.9.0", "stage": "rc", "revision": "1"}
+
+
+def test_release_pattern_matches_post_release_tag() -> None:
+    pattern = re.compile(_load_release_pattern())
+    match = pattern.match("7.9.0.post1")
+    assert match is not None
+    assert match.groupdict() == {"base": "7.9.0", "stage": "post", "revision": "1"}

tests/unit/test_storage_compat.py

@@ -0,0 +1,293 @@
+"""
+Storage-service compatibility unit tests.
+
+Validates that Python SDK storage query classes remain compatible with the
+response shapes produced by storage-service (the Rainbow replacement).  These
+tests mock at the HTTP level and require no running service.
+
+Covered flows:
+  * UploadDocument: POST /storage/files/store, LegacyUploadResponseItem shape
+  * CreateStorageURLs: indico-file:///storage<path> URI construction
+  * RetrieveStorageObject: indico-file:// prefix stripping and GET path
+"""
+
+import io
+import json
+import threading
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from pathlib import Path
+
+import pytest
+
+from indico.client import IndicoClient
+from indico.client.request import HTTPMethod
+from indico.config import IndicoConfig
+from indico.errors import IndicoRequestError
+from indico.queries.model_import import _UploadSMExport
+from indico.queries.storage import (
+    CreateStorageURLs,
+    RetrieveStorageObject,
+    UploadDocument,
+)
+
+# ---------------------------------------------------------------------------
+# Response shape produced by storage-service /files/store endpoint
+# (mirrors LegacyUploadResponseItem from storage_service/routes/blob_routes.py)
+# ---------------------------------------------------------------------------
+STORAGE_SERVICE_UPLOAD_RESPONSE = [
+    {
+        "path": "/uploads/42/abc-uuid",
+        "name": "document.pdf",
+        "size": 12345,
+        "upload_type": "user",
+    }
+]
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def cfg():
+    return IndicoConfig(protocol="mock", host="mock", api_token="test-token")
+
+
+@pytest.fixture
+def mock_request(requests_mock, cfg):
+    """Register a URL on requests_mock using the test config base URL."""
+
+    def _register(method, path, **kwargs):
+        url = f"{cfg.protocol}://{cfg.host}{path}"
+        getattr(requests_mock, method)(
+            url, **kwargs, headers={"Content-Type": "application/json"}
+        )
+
+    return _register
+
+
+@pytest.fixture
+def client(mock_request, cfg):
+    mock_request("post", "/auth/users/refresh_token", json={"auth_token": "tok"})
+    return IndicoClient(config=cfg)
+
+
+# ---------------------------------------------------------------------------
+# UploadDocument — request shape and response parsing
+# ---------------------------------------------------------------------------
+
+
+def test_upload_document_posts_to_storage_files_store(mock_request, client):
+    """UploadDocument sends POST to /storage/files/store."""
+    captured = []
+
+    def capture(request, context):
+        captured.append(request.path)
+        context.status_code = 200
+        context.headers["Content-Type"] = "application/json"
+        import json as _json
+
+        return _json.dumps(STORAGE_SERVICE_UPLOAD_RESPONSE)
+
+    mock_request("post", "/storage/files/store", text=capture)
+    client.call(UploadDocument(streams={"test.pdf": io.BytesIO(b"data")}))
+    assert captured == ["/storage/files/store"]
+
+
+def test_upload_document_processes_path_name_upload_type(mock_request, client):
+    """UploadDocument.process_response reads path/name/upload_type from storage-service."""
+    mock_request("post", "/storage/files/store", json=STORAGE_SERVICE_UPLOAD_RESPONSE)
+    result = client.call(UploadDocument(streams={"test.pdf": io.BytesIO(b"data")}))
+
+    assert len(result) == 1
+    assert result[0]["filename"] == "document.pdf"
+    meta = json.loads(result[0]["filemeta"])
+    assert meta["path"] == "/uploads/42/abc-uuid"
+    assert meta["name"] == "document.pdf"
+    assert meta["uploadType"] == "user"
+
+
+def test_upload_document_handles_multiple_files(mock_request, client):
+    """Multiple files in one upload are each parsed correctly."""
+    multi_response = [
+        {
+            "path": "/uploads/42/uuid-1",
+            "name": "a.pdf",
+            "size": 100,
+            "upload_type": "user",
+        },
+        {
+            "path": "/uploads/42/uuid-2",
+            "name": "b.pdf",
+            "size": 200,
+            "upload_type": "user",
+        },
+    ]
+    mock_request("post", "/storage/files/store", json=multi_response)
+    result = client.call(
+        UploadDocument(
+            streams={
+                "a.pdf": io.BytesIO(b"aaa"),
+                "b.pdf": io.BytesIO(b"bbb"),
+            }
+        )
+    )
+    assert len(result) == 2
+    assert result[0]["filename"] == "a.pdf"
+    assert result[1]["filename"] == "b.pdf"
+
+
+# ---------------------------------------------------------------------------
+# CreateStorageURLs — indico-file URI construction
+# ---------------------------------------------------------------------------
+
+
+def test_create_storage_urls_builds_indico_file_uris(mock_request, client):
+    """CreateStorageURLs returns indico-file:///storage<path> from storage-service response."""
+    mock_request("post", "/storage/files/store", json=STORAGE_SERVICE_UPLOAD_RESPONSE)
+    result = client.call(CreateStorageURLs(streams={"test.pdf": io.BytesIO(b"data")}))
+    assert result == ["indico-file:///storage/uploads/42/abc-uuid"]
+
+
+def test_create_storage_urls_round_trips_through_retrieve():
+    """A URI from CreateStorageURLs can be fed directly into RetrieveStorageObject."""
+    uri = "indico-file:///storage/uploads/42/abc-uuid"
+    req = RetrieveStorageObject(uri)
+    assert req.path == "/storage/uploads/42/abc-uuid"
+    assert req.method == HTTPMethod.GET
+
+
+# ---------------------------------------------------------------------------
+# RetrieveStorageObject — path construction
+# ---------------------------------------------------------------------------
+
+
+def test_retrieve_storage_object_strips_indico_file_scheme():
+    """indico-file:// prefix is stripped; remaining path becomes the GET path."""
+    req = RetrieveStorageObject("indico-file:///storage/submissions/1/2/result.json")
+    assert req.path == "/storage/submissions/1/2/result.json"
+    assert req.method == HTTPMethod.GET
+
+
+def test_retrieve_storage_object_accepts_dict_with_url_key():
+    """Accepts a dict with 'url' key (as returned by GraphQL result objects)."""
+    req = RetrieveStorageObject({"url": "indico-file:///storage/extractions/99.json"})
+    assert req.path == "/storage/extractions/99.json"
+
+
+def test_retrieve_storage_object_fetches_content(mock_request, client):
+    """GET /storage/<path> is issued and the response body is returned."""
+    payload = {"status": "complete", "results": [{"text": "hello"}]}
+    mock_request("get", "/storage/submissions/1/2/result.json", json=payload)
+    result = client.call(
+        RetrieveStorageObject("indico-file:///storage/submissions/1/2/result.json")
+    )
+    assert result == payload
+
+
+def test_retrieve_storage_object_follows_redirects():
+    """Storage GET requests follow redirects in redirect-mode deployments."""
+    payload = {"status": "complete", "results": [{"text": "redirected"}]}
+    refresh_path = "/auth/users/refresh_token"
+    source_path = "/storage/submissions/1/2/result.json"
+    redirected_path = "/storage/signed/submissions/1/2/result.json"
+
+    class Handler(BaseHTTPRequestHandler):
+        def do_POST(self):  # noqa: N802
+            if self.path != refresh_path:
+                self.send_response(404)
+                self.end_headers()
+                return
+            body = json.dumps({"auth_token": "tok"}).encode("utf-8")
+            self.send_response(200)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+
+        def do_GET(self):  # noqa: N802
+            if self.path == source_path:
+                self.send_response(302)
+                self.send_header(
+                    "Location",
+                    f"http://{self.server.server_address[0]}:{self.server.server_address[1]}{redirected_path}",
+                )
+                self.end_headers()
+                return
+            if self.path == redirected_path:
+                body = json.dumps(payload).encode("utf-8")
+                self.send_response(200)
+                self.send_header("Content-Type", "application/json")
+                self.send_header("Content-Length", str(len(body)))
+                self.end_headers()
+                self.wfile.write(body)
+                return
+            self.send_response(404)
+            self.end_headers()
+
+        def log_message(self, format, *args):  # noqa: A003
+            return
+
+    with HTTPServer(("127.0.0.1", 0), Handler) as server:
+        thread = threading.Thread(target=server.serve_forever, daemon=True)
+        thread.start()
+        try:
+            host = f"{server.server_address[0]}:{server.server_address[1]}"
+            client = IndicoClient(
+                config=IndicoConfig(protocol="http", host=host, api_token="test-token")
+            )
+            result = client.call(
+                RetrieveStorageObject(
+                    "indico-file:///storage/submissions/1/2/result.json"
+                )
+            )
+            assert result == payload
+        finally:
+            server.shutdown()
+            thread.join(timeout=5)
+
+
+def test_upload_static_model_export_puts_zip_to_signed_url(tmp_path, requests_mock):
+    """Static model export upload uses the signed URL with zip content-type."""
+    export_path = tmp_path / "model.zip"
+    export_bytes = b"zip-bytes"
+    export_path.write_bytes(export_bytes)
+    signed_url = "https://signed.example/upload"
+    storage_uri = "indico-file:///storage/exports/model.zip"
+
+    requests_mock.put(signed_url, status_code=200, text="")
+
+    request = _UploadSMExport(str(export_path))
+    result = request.process_response(
+        {"data": {"exportUpload": {"signedUrl": signed_url, "storageUri": storage_uri}}}
+    )
+
+    assert result == storage_uri
+    assert len(requests_mock.request_history) == 1
+    put_call = requests_mock.request_history[0]
+    assert put_call.method == "PUT"
+    assert put_call.headers["Content-Type"] == "application/zip"
+    assert put_call.body == export_bytes
+
+
+def test_upload_static_model_export_raises_on_put_failure(tmp_path, requests_mock):
+    """A failing signed-url PUT raises IndicoRequestError."""
+    export_path = tmp_path / "model.zip"
+    export_path.write_bytes(b"zip-bytes")
+    signed_url = "https://signed.example/upload"
+
+    requests_mock.put(signed_url, status_code=403, text="forbidden")
+
+    request = _UploadSMExport(str(Path(export_path)))
+    with pytest.raises(IndicoRequestError):
+        request.process_response(
+            {
+                "data": {
+                    "exportUpload": {
+                        "signedUrl": signed_url,
+                        "storageUri": "indico-file:///storage/exports/model.zip",
+                    }
+                }
+            }
+        )

tox.ini

@@ -10,6 +10,7 @@ deps =
     pytest < 8
     pytest-asyncio > 0.21
     requests-mock >= 1.8.0
+    tomli; python_version < "3.11"
     mypy == 1.8
     typing_extensions
     pandas-stubs