Add CLI tests

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

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)

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": "[email protected]", "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()

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?")