tests: added login tests

backend/tests/conftest.py

@@ -0,0 +1,98 @@
+import contextlib
+import os
+from typing import Iterator
+
+import pytest
+from pytest_docker.plugin import DockerComposeExecutor, Services
+
+
+def pytest_addoption(parser):
+    parser.addoption(
+        "--skip-docker",
+        action="store_true",
+        default=False,
+        help="Skip spinning up docker containers",
+    )
+
+
+@pytest.fixture(scope="session")
+def docker_compose_file(pytestconfig):
+    test_directory = os.path.dirname(os.path.abspath(__file__))
+    return os.path.join(test_directory, "docker-compose.yml")
+
+
+@pytest.fixture(scope="session")
+def docker_compose_project_name() -> str:
+    """Generate a project name using the current process PID. Override this
+    fixture in your tests if you need a particular project name."""
+
+    return "pytest_9070_atlas"
+
+
+@contextlib.contextmanager
+def get_docker_services(
+    docker_compose_command: str,
+    docker_compose_file: list[str] | str,
+    docker_compose_project_name: str,
+    docker_setup: list[str] | str,
+    docker_cleanup: list[str] | str,
+) -> Iterator[Services]:
+    docker_compose = DockerComposeExecutor(
+        docker_compose_command, docker_compose_file, docker_compose_project_name
+    )
+
+    try:
+        if docker_cleanup:
+            # Maintain backwards compatibility with the string format.
+            if isinstance(docker_cleanup, str):
+                docker_cleanup = [docker_cleanup]
+            for command in docker_cleanup:
+                docker_compose.execute(command)
+    except Exception:
+        pass
+
+    # setup containers.
+    if docker_setup:
+        # Maintain backwards compatibility with the string format.
+        if isinstance(docker_setup, str):
+            docker_setup = [docker_setup]
+        for command in docker_setup:
+            docker_compose.execute(command)
+
+    try:
+        # Let test(s) run.
+        yield Services(docker_compose)
+    finally:
+        # Clean up.
+        if docker_cleanup:
+            # Maintain backwards compatibility with the string format.
+            if isinstance(docker_cleanup, str):
+                docker_cleanup = [docker_cleanup]
+            for command in docker_cleanup:
+                docker_compose.execute(command)
+
+
+@pytest.fixture(scope="session")
+def docker_services(
+    docker_compose_command: str,
+    docker_compose_file: list[str] | str,
+    docker_compose_project_name: str,
+    docker_setup: str,
+    docker_cleanup: str,
+    request,
+) -> Iterator[Services]:
+    """Start all services from a docker compose file (`docker-compose up`).
+    After test are finished, shutdown all services (`docker-compose down`)."""
+
+    if request.config.getoption("--skip-docker"):
+        yield
+        return
+
+    with get_docker_services(
+        docker_compose_command,
+        docker_compose_file,
+        docker_compose_project_name,
+        docker_setup,
+        docker_cleanup,
+    ) as docker_service:
+        yield docker_service

backend/tests/docker-compose.yml

@@ -0,0 +1,6 @@
+version: '2'
+services:
+  scylla:
+    image: scylladb/scylla:latest
+    ports:
+      - "9070:9042"

backend/tests/test_login.py

@@ -0,0 +1,70 @@
+import os
+import socket
+
+import pytest
+from bec_atlas.main import AtlasApp
+from bec_atlas.utils.setup_database import setup_database
+from fastapi.testclient import TestClient
+
+
+@pytest.fixture(scope="session")
+def scylla_container(docker_ip, docker_services):
+    host = docker_ip
+    if os.path.exists("/.dockerenv"):
+        # if we are running in the CI, scylla was started as 'scylla' service
+        host = "scylla"
+    if docker_services is None:
+        port = 9042
+    else:
+        port = docker_services.port_for("scylla", 9042)
+
+    setup_database(host=host, port=port)
+    return host, port
+
+
+@pytest.fixture(scope="session")
+def client(scylla_container):
+    host, port = scylla_container
+    config = {"scylla": {"hosts": [(host, port)]}}
+
+    with TestClient(AtlasApp(config).app) as _client:
+        yield _client
+
+
+@pytest.mark.timeout(60)
+def test_login(client):
+    """
+    Test that the login endpoint returns a token.
+    """
+    response = client.post(
+        "/api/v1/user/login", json={"username": "admin@bec_atlas.ch", "password": "admin"}
+    )
+    assert response.status_code == 200
+    token = response.json()
+    assert isinstance(token, str)
+    assert len(token) > 20
+
+
+@pytest.mark.timeout(60)
+def test_login_wrong_password(client):
+    """
+    Test that the login returns a 401 when the password is wrong.
+    """
+    response = client.post(
+        "/api/v1/user/login", json={"username": "admin@bec_atlas.ch", "password": "wrong_password"}
+    )
+    assert response.status_code == 401
+    assert response.json() == {"detail": "Invalid password"}
+
+
+@pytest.mark.timeout(60)
+def test_login_unknown_user(client):
+    """
+    Test that the login returns a 404 when the user is unknown.
+    """
+    response = client.post(
+        "/api/v1/user/login",
+        json={"username": "no_user@bec_atlas.ch", "password": "wrong_password"},
+    )
+    assert response.status_code == 404
+    assert response.json() == {"detail": "User not found"}