Add user journey for logging in and deleting shared environment

conda-store-server/tests/user_journeys/test_user_journeys.py

@@ -29,15 +29,74 @@ def test_admin_user_can_create_environment(
     base_url: str, token: str, specification_path: str
 ) -> None:
     """Test that an admin user can create an environment."""
-    namespace = utils.API.gen_random_namespace()
     api = utils.API(base_url=base_url, token=token)
-    api.create_namespace(namespace)
-    response = api.create_environment(namespace, specification_path)
-    data = response.json()["data"]
-    assert "build_id" in data
-    build_id = data["build_id"]
-    assert build_id is not None
-    build = api.wait_for_successful_build(build_id)
-    environment_name = build.json()["data"]["specification"]["name"]
-    api.delete_environment(namespace, environment_name)
+    namespace = "default"
+    environment = api.create_environment(namespace, specification_path).json()["data"][
+        "specification"
+    ]["name"]
+    api.delete_environment(namespace, environment)
+
+
+@pytest.mark.user_journey
+@pytest.mark.parametrize(
+    "specification_path",
+    [
+        ("tests/user_journeys/test_data/simple_environment.yaml"),
+    ],
+)
+def test_admin_login_and_delete_shared_environment(
+    base_url: str, specification_path: str
+) -> None:
+    """Test that an admin can login and create/delete an env in a shared namespace."""
+    api = utils.API(base_url=base_url)
+
+    # Create a shared namespace; default permissions for namepace/environment
+    # */* is admin
+    namespace = api.create_namespace().json()["data"]["name"]
+    environment = api.create_environment(
+        namespace,
+        specification_path,
+    ).json()["data"][
+        "specification"
+    ]["name"]
+
+    api.delete_environment(namespace, environment)
+    api.delete_namespace(namespace)
+
+
+@pytest.mark.user_journey
+@pytest.mark.parametrize(
+    "specification_path",
+    [
+        ("tests/user_journeys/test_data/simple_environment.yaml"),
+    ],
+)
+def test_user_login_and_create_shared_environment(
+    base_url: str, specification_path: str
+) -> None:
+    """Test that a user can login and create an environment in a shared namespace."""
+    api = utils.API(base_url=base_url)
+
+    # Create a shared namespace; default permissions for namepace/environment
+    # */* is admin
+    namespace = api.create_namespace().json()["data"]["name"]
+
+    dev_api = utils.API(
+        base_url=base_url,
+        token=api.create_token(
+            namespace,
+            "developer",
+        ).json()[
+            "data"
+        ]["token"],
+    )
+
+    environment = dev_api.create_environment(
+        namespace,
+        specification_path,
+    ).json()[
+        "data"
+    ]["specification"]["name"]
+
+    api.delete_environment(namespace, environment)
     api.delete_namespace(namespace)

conda-store-server/tests/user_journeys/utils/api_utils.py

@@ -2,6 +2,7 @@
 import time
 import uuid
 from enum import Enum
+from typing import Union
 
 import requests
 import utils.time_utils as time_utils
@@ -19,6 +20,11 @@ class BuildStatus(Enum):
     CANCELED = "CANCELED"
 
 
+class NamespaceStatus(Enum):
+    OK = "ok"
+    ERROR = "error"
+
+
 class API:
     """
     Helper class for making requests to the API.
@@ -69,9 +75,44 @@ def _login(self, username: str, password: str) -> None:
         data = token_response.json()
         self.token = data["data"]["token"]
 
-    def create_namespace(self, namespace: str) -> requests.Response:
-        """Create a namespace."""
-        return self._make_request(f"api/v1/namespace/{namespace}", method="POST")
+    def create_namespace(
+        self,
+        namespace: Union[str, None] = None,
+        max_iterations: int = 100,
+        sleep_time: int = 5,
+    ) -> requests.Response:
+        """Create a namespace.
+
+        Parameters
+        ----------
+        namespace : str
+            Name of the namespace to create. If None, use a random namespace name
+        max_iterations : int
+            Max number of times to check whether the namespace was created before failing
+        sleep_time : int
+            Seconds to wait between each status check
+
+        Returns
+        -------
+        requests.Response
+            Response from the conda-store server
+        """
+        if namespace is None:
+            namespace = self.gen_random_namespace()
+
+        self._make_request(f"api/v1/namespace/{namespace}", method="POST")
+        for i in range(max_iterations):
+            response = self._make_request(f"api/v1/namespace/{namespace}")
+            status = NamespaceStatus(response.json()["status"])
+            if status in [NamespaceStatus.OK, NamespaceStatus.ERROR]:
+                return response
+
+            time.sleep(sleep_time)
+
+        raise TimeoutError(
+            f"Timed out waiting to create namespace {namespace}. Current response: "
+            f"{response.json()}"
+        )
 
     def create_token(
         self, namespace: str, role: str, default_namespace: str = "default"
@@ -85,37 +126,56 @@ def create_token(
         return self._make_request("api/v1/token", method="POST", json_data=json_data)
 
     def create_environment(
-        self, namespace: str, specification_path: str
+        self,
+        namespace: str,
+        specification_path: str,
+        max_iterations: int = 100,
+        sleep_time: int = 5,
     ) -> requests.Response:
-        """
-        Create an environment.
-        The environment specification is read
-        from a conda environment.yaml file.
+        """Create an environment.
+
+        Parameters
+        ----------
+        namespace : str
+            Namespace the environment should be written to
+        specification_path : str
+            Path to conda environment specification file
+        max_iterations : int
+            Max number of times to check whether the build completed before failing
+        sleep_time : int
+            Seconds to wait between each status check
+
+        Returns
+        -------
+        requests.Response
+            Response from the conda-store server
         """
         with open(specification_path, "r", encoding="utf-8") as file:
             specification_content = file.read()
 
-        json_data = {"namespace": namespace, "specification": specification_content}
-
-        return self._make_request(
-            "api/v1/specification", method="POST", json_data=json_data
+        response = self._make_request(
+            "api/v1/specification",
+            method="POST",
+            json_data={"namespace": namespace, "specification": specification_content},
         )
+        build_id = response.json()["data"]["build_id"]
+        for i in range(max_iterations):
+            response = self._make_request(f"api/v1/build/{build_id}/")
+            status = BuildStatus(response.json()["data"]["status"])
+
+            if status in [
+                BuildStatus.COMPLETED,
+                BuildStatus.FAILED,
+                BuildStatus.CANCELED,
+            ]:
+                return response
 
-    def wait_for_successful_build(
-        self, build_id: str, max_iterations: int = 100, sleep_time: int = 5
-    ) -> requests.Response:
-        """Wait for a build to complete."""
-        status = BuildStatus.QUEUED.value
-        iterations = 0
-        while status != BuildStatus.COMPLETED.value:
-            if iterations > max_iterations:
-                raise TimeoutError("Timed out waiting for build")
-            response = self._make_request(f"api/v1/build/{build_id}", method="GET")
-            status = response.json()["data"]["status"]
-            assert status != BuildStatus.FAILED.value, "Build failed"
-            iterations += 1
             time.sleep(sleep_time)
-        return response
+
+        raise TimeoutError(
+            f"Timed out waiting to create namespace {namespace}. Current response: "
+            f"{response.json()}"
+        )
 
     def delete_environment(
         self, namespace: str, environment_name: str