fix: fix deploy to same namespace if organization_id incorrectly set

scw_serverless/utils/credentials.py

@@ -18,6 +18,7 @@ def get_scw_client(
     """Attempts to load the profile. Will raise on invalid profiles."""
     client = Client.from_config_file_and_env(profile_name)
     _update_client_from_cli(client, secret_key, project_id, region)
+
     return _validate_client(client)
 
 
@@ -47,3 +48,7 @@ def _update_client_from_cli(
     if not client.default_region:
         get_logger().info(f"No region was configured, using {DEFAULT_REGION}")
         client.default_region = DEFAULT_REGION
+
+    # Not used by the API framework
+    # Can lead to issues if project_id does not belong to organization
+    client.default_organization_id = None

tests/app_fixtures/app.py

@@ -0,0 +1,23 @@
+from typing import Any
+
+from scw_serverless.app import Serverless
+
+DESCRIPTION = "Say hello to the whole world."
+MESSAGE = "Hello World!"
+NAMESPACE_NAME = "integration-tests"
+
+app = Serverless(NAMESPACE_NAME)
+
+
+@app.func(
+    description=DESCRIPTION,
+    privacy="public",
+    env={"key": "value"},
+    secret={},
+    min_scale=0,
+    max_scale=20,
+    memory_limit=256,
+    timeout="300s",
+)
+def hello_world(_event: dict[str, Any], _context: dict[str, Any]):
+    return MESSAGE

tests/dev/app.py → tests/app_fixtures/app_updated.py

@@ -2,24 +2,21 @@
 
 from scw_serverless.app import Serverless
 
+MESSAGE = "Hello galaxy!"
+DESCRIPTION = "Say hello to the whole galaxy."
+
 app = Serverless("integration-tests")
 
 
 @app.func(
-    description="This is a description",
+    description=DESCRIPTION,
     privacy="public",
     env={"key": "value"},
     secret={},
     min_scale=0,
-    max_scale=20,
+    max_scale=10,  # Differs from app.py
     memory_limit=256,
     timeout="300s",
 )
 def hello_world(_event: dict[str, Any], _context: dict[str, Any]):
-    """handle a request to the function
-    Args:
-        event (dict): request params
-        context (dict): function call metadata
-    """
-
-    return "Hello World!"
+    return MESSAGE

tests/dev/multiple_functions.py → tests/app_fixtures/multiple_functions.py

@@ -2,19 +2,25 @@
 
 from scw_serverless.app import Serverless
 
+MESSAGES = {
+    "hello-world": "Hello World!",
+    "cloud-of-choice": "The cloud of choice!",
+    "scaleway": "Scaleway",
+}
+
 app = Serverless("integration-tests")
 
 
 @app.func(memory_limit=256)
 def hello_world(_event: dict[str, Any], _context: dict[str, Any]):
-    return "Hello World!"
+    return MESSAGES["hello-world"]
 
 
 @app.func(memory_limit=256)
 def cloud_of_choice(_event: dict[str, Any], _context: dict[str, Any]):
-    return "The cloud of choice"
+    return MESSAGES["cloud-of-choice"]
 
 
 @app.func(memory_limit=256)
 def scaleway(_event: dict[str, Any], _context: dict[str, Any]):
-    return "Scaleway"
+    return MESSAGES["scaleway"]

tests/constants.py

@@ -0,0 +1,14 @@
+import os
+from pathlib import Path
+
+DEFAULT_REGION = "pl-waw"
+SCALEWAY_API_URL = "https://api.scaleway.com/"
+
+COLD_START_TIMEOUT = 20
+
+TESTS_DIR = os.path.realpath(os.path.dirname(__file__))
+
+APP_FIXTURES_PATH = Path(TESTS_DIR, "app_fixtures")
+
+APP_PY_PATH = APP_FIXTURES_PATH / "app.py"
+MULTIPLE_FUNCTIONS = APP_FIXTURES_PATH / "multiple_functions.py"

tests/integrations/deploy/deploy_wrapper.py

@@ -0,0 +1,45 @@
+import re
+import shutil
+import tempfile
+from pathlib import Path
+from typing import Literal
+
+from scaleway import Client
+
+from ..utils import run_cli, trigger_function
+
+FunctionUrl = str
+
+
+def run_deploy_command(
+    client: Client, app_path: Path, backend: Literal["serverless", "api"] = "api"
+) -> list[FunctionUrl]:
+    """Run deploy command with a specific backend."""
+
+    app_dir = app_path.parent.resolve()
+
+    # Run the command inside a temporary directory
+    with tempfile.TemporaryDirectory() as directory:
+        shutil.copytree(src=app_dir, dst=directory, dirs_exist_ok=True)
+
+        ret = run_cli(client, directory, ["deploy", app_path.name, "-b", backend])
+
+    assert ret.returncode == 0, f"Non-null return code: {ret}"
+
+    output = ret.stderr if backend == "serverless" else ret.stdout
+    output = str(output.decode("UTF-8")).strip()
+
+    # Parse the functions URL from the program output
+    pattern = re.compile(
+        r"(Function [a-z0-9-]+ (?:has been )?deployed to:? (https://.+))"
+    )
+    groups = re.findall(pattern, output)
+
+    function_urls = []
+    for group in groups:
+        function_urls.append(group[1])
+
+        # Call the actual function
+        trigger_function(group[1])
+
+    return function_urls

tests/integrations/deploy/test_api_backend.py

@@ -1,34 +1,80 @@
 # pylint: disable=unused-import,redefined-outer-name # fixture
-from tests.integrations.utils import serverless_test_project  # noqa: F401
-from tests.integrations.utils import ServerlessTestProject
-from tests.utils import APP_PY_PATH
-
-
-def test_integration_deploy(
-    serverless_test_project: ServerlessTestProject,  # noqa: F811
-):
-    serverless_test_project.deploy(APP_PY_PATH, backend="api")
-
-
-# Due to external factors these test will randomly fail.
-# def test_integration_deploy_multiple_functions():
-#     deploy(test_utils.MULTIPLE_FUNCTIONS_PY_PATH)
-#
-#
-# def test_integration_deploy_existing_function():
-#     project_id = deploy(test_utils.APP_PY_PATH, do_cleanup=False)
-#     deploy(test_utils.APP_PY_PATH, do_cleanup=True, project_id=project_id)
-#
-#
-# def test_integration_deploy_multiple_existing_functions():
-#     project_id = deploy(test_utils.MULTIPLE_FUNCTIONS_PY_PATH, do_cleanup=False)
-#     deploy(
-#         test_utils.MULTIPLE_FUNCTIONS_PY_PATH, do_cleanup=True, project_id=project_id
-#     )
-#
-#
-# def test_integration_deploy_one_existing_function():
-#     project_id = deploy(test_utils.APP_PY_PATH, do_cleanup=False)
-#     deploy(
-#         test_utils.MULTIPLE_FUNCTIONS_PY_PATH, do_cleanup=True, project_id=project_id
-#     )
+
+import scaleway.function.v1beta1 as sdk
+
+from tests import constants
+from tests.app_fixtures import app, app_updated, multiple_functions
+from tests.integrations.deploy.deploy_wrapper import run_deploy_command
+from tests.integrations.project_fixture import scaleway_project  # noqa
+from tests.integrations.utils import create_client, trigger_function
+
+
+def test_integration_deploy(scaleway_project: str):  # noqa
+    client = create_client()
+    client.default_project_id = scaleway_project
+
+    url, *_ = run_deploy_command(
+        client,
+        app_path=constants.APP_PY_PATH,
+    )
+
+    # Check message content
+    resp = trigger_function(url)
+    assert resp.text == app.MESSAGE
+
+
+def test_integration_deploy_existing_function(scaleway_project: str):  # noqa
+    client = create_client()
+    client.default_project_id = scaleway_project
+
+    url, *_ = run_deploy_command(
+        client,
+        app_path=constants.APP_PY_PATH,
+    )
+
+    # Check message content
+    resp = trigger_function(url)
+    assert resp.text == app.MESSAGE
+
+    # Get function_id
+    api = sdk.FunctionV1Beta1API(client)
+    namespace, *_ = api.list_namespaces_all(name=app.NAMESPACE_NAME)
+
+    # Check description
+    function, *_ = api.list_functions_all(namespace_id=namespace.id, name="hello-world")
+    assert function.description == app.DESCRIPTION
+
+    # Deploy twice in a row
+    url, *_ = run_deploy_command(
+        client,
+        app_path=constants.APP_FIXTURES_PATH / "app_updated.py",
+    )
+
+    # Check updated message content
+    resp = trigger_function(url)
+    assert resp.text == app_updated.MESSAGE
+
+    # Check updated description
+    function = api.get_function(function_id=function.id)
+    assert function.description == app_updated.DESCRIPTION
+
+
+def test_integration_deploy_multiple_functions(scaleway_project: str):  # noqa
+    client = create_client()
+    client.default_project_id = scaleway_project
+
+    urls = run_deploy_command(
+        client,
+        app_path=constants.MULTIPLE_FUNCTIONS,
+    )
+
+    for url in urls:
+        # Get the function_name from the url
+        function_name = None
+        for name in multiple_functions.MESSAGES:
+            if name in url:
+                function_name = name
+        assert function_name
+
+        resp = trigger_function(url)
+        assert resp.text == multiple_functions.MESSAGES[function_name]

tests/integrations/deploy/test_sf_backend.py

@@ -1,47 +1,59 @@
 # pylint: disable=unused-import,redefined-outer-name # fixture
-from tests.integrations.utils import serverless_test_project  # noqa: F401
-from tests.integrations.utils import ServerlessTestProject
-from tests.utils import APP_PY_PATH
 
+import scaleway.function.v1beta1 as sdk
 
-def test_integration_deploy_using_srvless_fw(
-    serverless_test_project: ServerlessTestProject,  # noqa: F811
+from tests import constants
+from tests.app_fixtures import app, app_updated
+from tests.integrations.deploy.deploy_wrapper import run_deploy_command
+from tests.integrations.project_fixture import scaleway_project  # noqa
+from tests.integrations.utils import create_client, trigger_function
+
+
+def test_integration_deploy_serverless_backend(scaleway_project: str):  # noqa
+    client = create_client()
+    client.default_project_id = scaleway_project
+
+    url, *_ = run_deploy_command(
+        client, app_path=constants.APP_PY_PATH, backend="serverless"
+    )
+
+    # Check message content
+    resp = trigger_function(url)
+    assert resp.text == app.MESSAGE
+
+
+def test_integration_deploy_existing_function_serverless_backend(
+    scaleway_project: str,  # noqa
 ):
-    serverless_test_project.deploy(APP_PY_PATH, backend="serverless")
-
-
-# Due to external factors these test will randomly fail.
-# def test_integration_deploy_multiple_functions_using_srvless_fw():
-#     deploy(test_utils.MULTIPLE_FUNCTIONS_PY_PATH, backend="serverless")
-#
-#
-# def test_integration_deploy_existing_function_using_srvless_fw():
-#     project_id = deploy(test_utils.APP_PY_PATH, do_cleanup=False, backend="serverless")
-#     deploy(
-#         test_utils.APP_PY_PATH,
-#         do_cleanup=True,
-#         project_id=project_id,
-#         backend="serverless",
-#     )
-#
-#
-# def test_integration_deploy_multiple_existing_functions_using_srvless_fw():
-#     project_id = deploy(
-#         test_utils.MULTIPLE_FUNCTIONS_PY_PATH, do_cleanup=False, backend="serverless"
-#     )
-#     deploy(
-#         test_utils.MULTIPLE_FUNCTIONS_PY_PATH,
-#         do_cleanup=True,
-#         project_id=project_id,
-#         backend="serverless",
-#     )
-#
-#
-# def test_integration_deploy_one_existing_function_using_srvless_fw():
-#     project_id = deploy(test_utils.APP_PY_PATH, do_cleanup=False, backend="serverless")
-#     deploy(
-#         test_utils.MULTIPLE_FUNCTIONS_PY_PATH,
-#         do_cleanup=True,
-#         project_id=project_id,
-#         backend="serverless",
-#     )
+    client = create_client()
+    client.default_project_id = scaleway_project
+
+    url, *_ = run_deploy_command(
+        client, app_path=constants.APP_PY_PATH, backend="serverless"
+    )
+
+    # Check message content
+    resp = trigger_function(url)
+    assert resp.text == app.MESSAGE
+
+    # Get function_id
+    api = sdk.FunctionV1Beta1API(client)
+    namespace, *_ = api.list_namespaces_all(name=app.NAMESPACE_NAME)
+
+    # Check description
+    function, *_ = api.list_functions_all(namespace_id=namespace.id, name="hello-world")
+    assert function.description == app.DESCRIPTION
+
+    # Deploy twice in a row
+    url, *_ = run_deploy_command(
+        client,
+        app_path=constants.APP_FIXTURES_PATH.joinpath("app_updated.py"),
+    )
+
+    # Check updated message content
+    resp = trigger_function(url)
+    assert resp.text == app_updated.MESSAGE
+
+    # Check updated description
+    function = api.get_function(function_id=function.id)
+    assert function.description == app_updated.DESCRIPTION