diff --git a/tests/cli/__init__.py b/tests/cli/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/cli/conftest.py b/tests/cli/conftest.py new file mode 100644 index 000000000..e34dce96c --- /dev/null +++ b/tests/cli/conftest.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +import pytest +import requests +import yaml + +from diracx.core.preferences import get_diracx_preferences + + +@pytest.fixture +def cli_env(monkeypatch, tmp_path, demo_dir): + """Set up the environment for the CLI""" + # HACK: Find the URL of the demo DiracX instance + helm_values = yaml.safe_load((demo_dir / "values.yaml").read_text()) + host_url = helm_values["dex"]["config"]["issuer"].rsplit(":", 1)[0] + diracx_url = f"{host_url}:8000" + + # Ensure the demo is working + r = requests.get(f"{diracx_url}/openapi.json") + r.raise_for_status() + assert r.json()["info"]["title"] == "Dirac" + + env = { + "DIRACX_URL": diracx_url, + "HOME": tmp_path, + } + for key, value in env.items(): + monkeypatch.setenv(key, value) + yield env + + # The DiracX preferences are cached however when testing this cache is invalid + get_diracx_preferences.cache_clear() + + +@pytest.fixture +async def with_cli_login(monkeypatch, capfd, cli_env, tmp_path): + from .test_login import test_login + + try: + credentials = await test_login(monkeypatch, capfd, cli_env) + except Exception: + pytest.skip("Login failed, fix test_login to re-enable this test") + + credentials_path = tmp_path / "credentials.json" + credentials_path.write_text(credentials) + monkeypatch.setenv("DIRACX_CREDENTIALS_PATH", str(credentials_path)) + yield diff --git a/tests/cli/test_jobs.py b/tests/cli/test_jobs.py new file mode 100644 index 000000000..3875fa7fe --- /dev/null +++ b/tests/cli/test_jobs.py @@ -0,0 +1,13 @@ +from __future__ import annotations + +import json + +from diracx import cli + + +async def test_search(with_cli_login, capfd): + await cli.jobs.search() + cap = capfd.readouterr() + assert cap.err == "" + # By default the output should be in JSON format as capfd is not a TTY + json.loads(cap.out) diff --git a/tests/cli/test_login.py b/tests/cli/test_login.py new file mode 100644 index 000000000..ea432b63a --- /dev/null +++ b/tests/cli/test_login.py @@ -0,0 +1,92 @@ +from __future__ import annotations + +import asyncio +import re +from html.parser import HTMLParser +from pathlib import Path +from urllib.parse import urljoin + +import requests + +from diracx import cli + + +def do_device_flow_with_dex(url: str) -> None: + """Do the device flow with dex""" + + class DexLoginFormParser(HTMLParser): + def handle_starttag(self, tag, attrs): + nonlocal action_url + if "form" in str(tag): + assert action_url is None + action_url = urljoin(login_page_url, dict(attrs)["action"]) + + # Get the login page + r = requests.get(url) + r.raise_for_status() + login_page_url = r.url # This is not the same as URL as we redirect to dex + login_page_body = r.text + + # Search the page for the login form so we know where to post the credentials + action_url = None + DexLoginFormParser().feed(login_page_body) + assert action_url is not None, login_page_body + + # Do the actual login + r = requests.post( + action_url, data={"login": "admin@example.com", "password": "password"} + ) + r.raise_for_status() + # This should have redirected to the DiracX page that shows the login is complete + assert "Please close the window" in r.text + + +async def test_login(monkeypatch, capfd, cli_env): + poll_attempts = 0 + + def fake_sleep(*args, **kwargs): + nonlocal poll_attempts + + # Keep track of the number of times this is called + poll_attempts += 1 + + # After polling 5 times, do the actual login + if poll_attempts == 5: + # The login URL should have been printed to stdout + captured = capfd.readouterr() + match = re.search(rf"{cli_env['DIRACX_URL']}[^\n]+", captured.out) + assert match, captured + + do_device_flow_with_dex(match.group()) + + # Ensure we don't poll forever + assert poll_attempts <= 10 + + # Reduce the sleep duration to zero to speed up the test + return unpatched_sleep(0) + + # We monkeypatch asyncio.sleep to provide a hook to run the actions that + # would normally be done by a user. This includes capturing the login URL + # and doing the actual device flow with dex. + unpatched_sleep = asyncio.sleep + monkeypatch.setattr("asyncio.sleep", fake_sleep) + + expected_credentials_path = Path( + cli_env["HOME"], ".cache", "diracx", "credentials.json" + ) + + # Ensure the credentials file does not exist before logging in + assert not expected_credentials_path.exists() + + # Run the login command + await cli.login(vo="diracAdmin", group=None, property=None) + captured = capfd.readouterr() + assert "Login successful!" in captured.out + assert captured.err == "" + + # Ensure the credentials file exists after logging in + assert expected_credentials_path.exists() + + # Return the credentials so this test can also be used by the + # "with_cli_login" fixture + return expected_credentials_path.read_text() diff --git a/tests/conftest.py b/tests/conftest.py index b3a743187..b0519e22f 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -222,13 +222,17 @@ def admin_user_client(test_client, test_auth_settings): @pytest.fixture(scope="session") -def demo_kubectl_env(request): - """Get the dictionary of environment variables for kubectl to control the demo""" +def demo_dir(request) -> Path: demo_dir = request.config.getoption("--demo-dir") if demo_dir is None: pytest.skip("Requires a running instance of the DiracX demo") demo_dir = (demo_dir / ".demo").resolve() + yield demo_dir + +@pytest.fixture(scope="session") +def demo_kubectl_env(demo_dir): + """Get the dictionary of environment variables for kubectl to control the demo""" kube_conf = demo_dir / "kube.conf" if not kube_conf.exists(): raise RuntimeError(f"Could not find {kube_conf}, is the demo running?")