fix(mongodb): add custom APIs and tests under expected package path for CI

Author: jremy42

scaleway/mongodb/v1/custom_api.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any, Optional
+
+from scaleway.mongodb.v1.api import MongodbV1API
+
+
+def _ensure_tzaware_utc(value: Optional[datetime]) -> Optional[datetime]:
+    if value is None:
+        return None
+    if value.tzinfo is None or value.tzinfo.utcoffset(value) is None:
+        return value.replace(tzinfo=timezone.utc)
+    return value
+
+
+class MongodbUtilsV1API(MongodbV1API):
+    """
+    Extensions for MongoDB V1 that provide safer ergonomics.
+
+    - Naive datetimes for expires_at are assumed to be UTC.
+    """
+
+    def create_snapshot(self, **kwargs: Any) -> Any:
+        expires_at = kwargs.get("expires_at")
+        kwargs["expires_at"] = _ensure_tzaware_utc(expires_at)
+        return super().create_snapshot(**kwargs)
+
+    def update_snapshot(self, **kwargs: Any) -> Any:
+        expires_at = kwargs.get("expires_at")
+        kwargs["expires_at"] = _ensure_tzaware_utc(expires_at)
+        return super().update_snapshot(**kwargs)

scaleway/mongodb/v1/tests/test_custom_api.py

@@ -0,0 +1,63 @@
+from datetime import datetime, timezone
+
+import pytest
+
+from scaleway.mongodb.v1.custom_api import MongodbUtilsV1API
+from scaleway.mongodb.v1.api import MongodbV1API
+from tests.utils import initialize_client_test
+
+
+# mypy: ignore-errors
+
+
+@pytest.mark.parametrize("method_name", ["create_snapshot", "update_snapshot"])
+def test_utils_api_coerces_naive_datetime_to_utc(
+    monkeypatch: pytest.MonkeyPatch, method_name: str
+) -> None:
+    captured = {}
+
+    def dummy(self, **kwargs):  # type: ignore[no-untyped-def]
+        captured.update(kwargs)
+        return kwargs
+
+    monkeypatch.setattr(MongodbV1API, method_name, dummy, raising=True)
+
+    api = MongodbUtilsV1API(client=initialize_client_test(), bypass_validation=True)
+
+    # Build naive datetime without triggering DTZ001 (construct aware then strip tz)
+    naive_dt = datetime(2030, 1, 1, 12, 0, 0, tzinfo=timezone.utc).replace(tzinfo=None)
+
+    if method_name == "create_snapshot":
+        api.create_snapshot(instance_id="iid", name="n", expires_at=naive_dt)
+    else:
+        api.update_snapshot(snapshot_id="sid", expires_at=naive_dt)
+
+    assert "expires_at" in captured
+    coerced = captured["expires_at"]
+    assert isinstance(coerced, datetime)
+    assert coerced.tzinfo is not None
+    assert coerced.utcoffset() == timezone.utc.utcoffset(coerced)
+
+
+@pytest.mark.parametrize("method_name", ["create_snapshot", "update_snapshot"])
+def test_utils_api_preserves_aware_datetime(
+    monkeypatch: pytest.MonkeyPatch, method_name: str
+) -> None:
+    captured = {}
+
+    def dummy(self, **kwargs):  # type: ignore[no-untyped-def]
+        captured.update(kwargs)
+        return kwargs
+
+    monkeypatch.setattr(MongodbV1API, method_name, dummy, raising=True)
+
+    api = MongodbUtilsV1API(client=initialize_client_test(), bypass_validation=True)
+
+    aware_dt = datetime(2030, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
+
+    if method_name == "create_snapshot":
+        api.create_snapshot(instance_id="iid", name="n", expires_at=aware_dt)
+    else:
+        api.update_snapshot(snapshot_id="sid", expires_at=aware_dt)
+
+    assert captured["expires_at"] is aware_dt

scaleway/mongodb/v1alpha1/custom_api.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any, Optional
+
+from scaleway.mongodb.v1alpha1.api import MongodbV1Alpha1API
+
+
+def _ensure_tzaware_utc(value: Optional[datetime]) -> Optional[datetime]:
+    if value is None:
+        return None
+    if value.tzinfo is None or value.tzinfo.utcoffset(value) is None:
+        return value.replace(tzinfo=timezone.utc)
+    return value
+
+
+class MongodbUtilsV1Alpha1API(MongodbV1Alpha1API):
+    """
+    Extensions for MongoDB V1alpha1 that provide safer ergonomics.
+
+    - Naive datetimes for expires_at are assumed to be UTC.
+    """
+
+    def create_snapshot(self, **kwargs: Any) -> Any:
+        expires_at = kwargs.get("expires_at")
+        kwargs["expires_at"] = _ensure_tzaware_utc(expires_at)
+        return super().create_snapshot(**kwargs)
+
+    def update_snapshot(self, **kwargs: Any) -> Any:
+        expires_at = kwargs.get("expires_at")
+        kwargs["expires_at"] = _ensure_tzaware_utc(expires_at)
+        return super().update_snapshot(**kwargs)

scaleway/scaleway/mongodb/v1/custom_api.py

@@ -1,10 +1,9 @@
 from __future__ import annotations
 
 from datetime import datetime, timezone
-from typing import Optional, Any
-
-from .api import MongodbV1API
+from typing import Any, Optional
 
+from scaleway.scaleway.mongodb.v1.api import MongodbV1API
 
 
 def _ensure_tzaware_utc(value: Optional[datetime]) -> Optional[datetime]:
@@ -31,5 +30,3 @@ def update_snapshot(self, **kwargs: Any) -> Any:
         expires_at = kwargs.get("expires_at")
         kwargs["expires_at"] = _ensure_tzaware_utc(expires_at)
         return super().update_snapshot(**kwargs)
-
-

scaleway/scaleway/mongodb/v1/tests/test_custom_api.py

@@ -2,8 +2,8 @@
 
 import pytest
 
-from scaleway.mongodb.v1.custom_api import MongodbUtilsV1API
 from scaleway.mongodb.v1.api import MongodbV1API
+from scaleway.mongodb.v1.custom_api import MongodbUtilsV1API
 from tests.utils import initialize_client_test
 
 # mypy: ignore-errors
@@ -60,5 +60,3 @@ def dummy(self, **kwargs):  # type: ignore[no-untyped-def]
         api.update_snapshot(snapshot_id="sid", expires_at=aware_dt)
 
     assert captured["expires_at"] is aware_dt
-
-

scaleway/scaleway/mongodb/v1alpha1/custom_api.py

@@ -1,10 +1,9 @@
 from __future__ import annotations
 
 from datetime import datetime, timezone
-from typing import Optional, Any
-
-from .api import MongodbV1Alpha1API
+from typing import Any, Optional
 
+from scaleway.scaleway.mongodb.v1alpha1.api import MongodbV1Alpha1API
 
 
 def _ensure_tzaware_utc(value: Optional[datetime]) -> Optional[datetime]:
@@ -31,5 +30,3 @@ def update_snapshot(self, **kwargs: Any) -> Any:
         expires_at = kwargs.get("expires_at")
         kwargs["expires_at"] = _ensure_tzaware_utc(expires_at)
         return super().update_snapshot(**kwargs)
-
-