test(gateway): add unit test for gateway manager

Author: cyclimse

pyproject.toml

@@ -60,6 +60,12 @@ myst_parser = ">=0.18.1,<1.1.0"
 sphinx = "^5.3.0"
 sphinx_rtd_theme = "^1.1.1"
 
+[tool.pytest.ini_options]
+filterwarnings = [
+    "ignore:.*pkg_resources\\.declare_namespace.*:DeprecationWarning",
+    "ignore:::pkg_resources",
+]
+
 [tool.pylint]
 load-plugins = ["pylint_per_file_ignores"]
 disable = "missing-module-docstring"

scw_serverless/config/route.py

@@ -1,6 +1,6 @@
 from dataclasses import dataclass
 from enum import Enum
-from typing import Optional
+from typing import Any, Optional
 
 from scw_serverless.config.utils import _SerializableDataClass
 
@@ -27,8 +27,17 @@ class GatewayRoute(_SerializableDataClass):
     target: Optional[str] = None
 
     def validate(self) -> None:
-        """Validates a route."""
+        """Validate a route."""
         if not self.relative_url:
             raise RuntimeError("Route relative_url must be defined")
         if not self.target:
             raise RuntimeError("Route target must be defined")
+        for method in self.http_methods or []:
+            if method not in HTTPMethod:
+                raise RuntimeError(f"Route contains invalid method {method.value}")
+
+    def asdict(self) -> dict[str, Any]:
+        serialized = super().asdict()
+        if self.http_methods:
+            serialized["http_methods"] = [method.value for method in self.http_methods]
+        return serialized

scw_serverless/gateway/gateway_manager.py

@@ -65,7 +65,7 @@ def update_routes(self) -> None:
         for function in routed_functions:
             if function.name not in created_functions:
                 raise RuntimeError(
-                    f"Could not update route to function {function.name}"
+                    f"Could not update route to function {function.name} "
                     + "because it was not deployed"
                 )
 

tests/test_gateway/test_gateway_manager.py

@@ -1,6 +1,7 @@
 import pytest
 import responses
 import scaleway.function.v1beta1 as sdk
+from responses.matchers import header_matcher, json_params_matcher, query_param_matcher
 from scaleway import Client
 
 from scw_serverless.app import Serverless
@@ -31,7 +32,7 @@ def app_gateway_manager() -> GatewayManager:
     client = Client(
         access_key="SCWXXXXXXXXXXXXXXXXX",
         # The uuid is validated
-        secret_key="498cce73-2a07-4e8c-b8ef-8f988e3c6929",
+        secret_key="498cce73-2a07-4e8c-b8ef-8f988e3c6929",  # nosec # false positive
         default_region=constants.DEFAULT_REGION,
     )
     return GatewayManager(app, MOCK_GATEWAY_URL, MOCK_GATEWAY_API_KEY, client)
@@ -60,15 +61,50 @@ def test_gateway_manager_update_routes(
         constants.SCALEWAY_FNC_API_URL + "/namespaces",
         json={"namespaces": [namespace]},
     )
+    # We have to provide a stop gap otherwise list_namepaces_all() will keep
+    # making API calls.
+    mocked_responses.get(
+        constants.SCALEWAY_FNC_API_URL + "/namespaces",
+        json={"namespaces": []},
+    )
 
     mocked_responses.get(
         constants.SCALEWAY_FNC_API_URL + "/functions",
+        match=[query_param_matcher({"namespace_id": namespace["id"], "page": 1})],
+        json={
+            "functions": [
+                {
+                    "name": function.name,
+                    "domain_name": HELLO_WORLD_MOCK_DOMAIN,
+                    "secret_environment_variables": [],
+                }
+            ]
+        },
+    )
+    mocked_responses.get(
+        constants.SCALEWAY_FNC_API_URL + "/functions",
+        match=[query_param_matcher({"namespace_id": namespace["id"], "page": 2})],
+        json={"functions": []},
+    )
+
+    # We should attempt to delete the route
+    mocked_responses.delete(
+        MOCK_GATEWAY_URL + "/scw",  # type: ignore
         match=[
-            responses.matchers.query_param_matcher(
-                {"namespace_id": namespace["id"], "page": 1}
-            )
+            header_matcher({"X-Auth-Token": MOCK_GATEWAY_API_KEY}),
+            json_params_matcher(params=function.gateway_route.asdict()),  # type: ignore
+        ],
+    )
+    # We should attempt to create the route
+    mocked_responses.post(
+        MOCK_GATEWAY_URL + "/scw",  # type: ignore
+        match=[
+            header_matcher({"X-Auth-Token": MOCK_GATEWAY_API_KEY}),
+            json_params_matcher(
+                params=function.gateway_route.asdict()  # type: ignore
+                | {"target": "https://" + HELLO_WORLD_MOCK_DOMAIN}
+            ),
         ],
-        json={"functions": []},
     )
 
     app_gateway_manager.update_routes()