feat(cli): fdp login/logout and auto-login on fdp run

Wire auth.login/logout into new `fdp login [--write]` and `fdp logout`
subcommands; thread auto_login=True into setup_environment for `fdp run`;
fix do_env to emit the device-declared bearer env-var name instead of
hardcoded BEARER_TOKEN.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: sammuli

fdp/cli.py

@@ -18,8 +18,10 @@
 import shlex
 import subprocess
 import sys
+from datetime import datetime, timezone
 from pathlib import Path
 
+from . import auth
 from .catalog import catalog
 from .environment import (
     build_device_config, resolve_bearer_token, _resolve_device_handle,
@@ -45,7 +47,34 @@ def do_env(args) -> None:
         print(f"export {key}={shlex.quote(str(value))}")
     token = resolve_bearer_token(handle)
     if token:
-        print(f"export BEARER_TOKEN={shlex.quote(token)}")
+        env_var = auth.bearer_env(handle) or "BEARER_TOKEN"
+        print(f"export {env_var}={shlex.quote(token)}")
+
+
+def do_login(args) -> None:
+    handle = _resolve_device_handle(args.default_device)
+    try:
+        result = auth.login(handle, write=args.write)
+    except auth.AuthError as exc:
+        print(f"Login failed: {exc}", file=sys.stderr)
+        sys.exit(1)
+    if result is None:
+        print(f"Device '{handle.schema.name}' needs no bearer token.")
+        return
+    if result.exp:
+        when = datetime.fromtimestamp(
+            result.exp, tz=timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
+    else:
+        when = "unknown"
+    print(f"Logged in to {result.device} ({result.scope}); "
+          f"token valid until {when}.")
+
+
+def do_logout(args) -> None:
+    handle = _resolve_device_handle(args.default_device)
+    removed = auth.logout(handle)
+    print("Removed cached token."
+          if removed else "No cached token to remove.")
 
 
 def do_run(args) -> None:
@@ -212,12 +241,22 @@ def build_parser() -> argparse.ArgumentParser:
                             help="Run a command with FDP env applied")
     p_run.add_argument("command_args", nargs=argparse.REMAINDER,
                         help="Command and args to pass through")
-    p_run.set_defaults(func=do_run)
+    p_run.set_defaults(func=do_run, auto_login=True)
 
     p_env = sub.add_parser("env",
                             help="Print env vars for shell eval")
     p_env.set_defaults(func=do_env)
 
+    p_login = sub.add_parser("login",
+                             help="Acquire/refresh a bearer token via pelican")
+    p_login.add_argument("--write", action="store_true",
+                         help="Request a write-scoped token (default: read).")
+    p_login.set_defaults(func=do_login, needs_env=False)
+
+    p_logout = sub.add_parser("logout",
+                              help="Delete the cached bearer token")
+    p_logout.set_defaults(func=do_logout, needs_env=False)
+
     p_ls = sub.add_parser("ls", help="List files on the FDP")
     p_ls.add_argument("--dirs-only", "-d", action="store_true",
                        help="Only show subdirectories")
@@ -288,6 +327,7 @@ def main(argv=None) -> None:
         setup_environment(
             device=args.default_device,
             bearer_token=args.bearer_token or None,
+            auto_login=getattr(args, "auto_login", False),
         )
 
     args.func(args)

tests/test_cli.py

@@ -22,6 +22,7 @@
 import sys
 import unittest
 from contextlib import ExitStack, redirect_stdout
+from types import SimpleNamespace
 from unittest import mock
 
 
@@ -329,5 +330,79 @@ def test_default_device_passed_to_setup_environment(self):
         self.assertEqual(kwargs.get("device"), "d3d")
 
 
+class TestCliLoginLogout(unittest.TestCase):
+    def test_login_dispatches_to_auth_login(self):
+        from fdp import cli, auth
+        ep = _make_catalog_ep("d3d", _D3D_TEST_YAML)
+        ct = auth.CachedToken(device="d3d", scope="read", exp=None)
+        with ExitStack() as stack:
+            stack.enter_context(mock.patch.object(
+                sys, "argv", ["fdp", "login"]))
+            stack.enter_context(mock.patch(
+                "fdp.catalog.entry_points", return_value=[ep]))
+            from fdp.catalog import catalog as _cat
+            _cat._cache = None
+            login_mock = stack.enter_context(
+                mock.patch.object(cli.auth, "login", return_value=ct))
+            buf = io.StringIO()
+            with redirect_stdout(buf):
+                cli.main()
+            login_mock.assert_called_once()
+            self.assertEqual(login_mock.call_args.kwargs.get("write"), False)
+
+    def test_login_write_flag(self):
+        from fdp import cli, auth
+        ep = _make_catalog_ep("d3d", _D3D_TEST_YAML)
+        ct = auth.CachedToken(device="d3d", scope="write", exp=None)
+        with ExitStack() as stack:
+            stack.enter_context(mock.patch.object(
+                sys, "argv", ["fdp", "login", "--write"]))
+            stack.enter_context(mock.patch(
+                "fdp.catalog.entry_points", return_value=[ep]))
+            from fdp.catalog import catalog as _cat
+            _cat._cache = None
+            login_mock = stack.enter_context(
+                mock.patch.object(cli.auth, "login", return_value=ct))
+            with redirect_stdout(io.StringIO()):
+                cli.main()
+            self.assertEqual(login_mock.call_args.kwargs.get("write"), True)
+
+    def test_logout_dispatches(self):
+        from fdp import cli
+        ep = _make_catalog_ep("d3d", _D3D_TEST_YAML)
+        with ExitStack() as stack:
+            stack.enter_context(mock.patch.object(
+                sys, "argv", ["fdp", "logout"]))
+            stack.enter_context(mock.patch(
+                "fdp.catalog.entry_points", return_value=[ep]))
+            from fdp.catalog import catalog as _cat
+            _cat._cache = None
+            logout_mock = stack.enter_context(
+                mock.patch.object(cli.auth, "logout", return_value=True))
+            with redirect_stdout(io.StringIO()):
+                cli.main()
+            logout_mock.assert_called_once()
+
+    def test_run_sets_auto_login_true(self):
+        from fdp import cli
+        ep = _make_catalog_ep("d3d", _D3D_TEST_YAML)
+        with ExitStack() as stack:
+            stack.enter_context(mock.patch.object(
+                sys, "argv", ["fdp", "run", "true"]))
+            stack.enter_context(mock.patch(
+                "fdp.catalog.entry_points", return_value=[ep]))
+            from fdp.catalog import catalog as _cat
+            _cat._cache = None
+            setup_mock = stack.enter_context(
+                mock.patch.object(cli, "setup_environment"))
+            stack.enter_context(mock.patch.object(
+                cli.subprocess, "run",
+                return_value=SimpleNamespace(returncode=0)))
+            with self.assertRaises(SystemExit):
+                cli.main()
+            self.assertEqual(
+                setup_mock.call_args.kwargs.get("auto_login"), True)
+
+
 if __name__ == "__main__":
     unittest.main()