diff --git a/google_health_api/cli/auth.py b/google_health_api/cli/auth.py index 1580d7f..b683025 100644 --- a/google_health_api/cli/auth.py +++ b/google_health_api/cli/auth.py @@ -63,17 +63,22 @@ class CredentialsAuth(AbstractAuth): """Auth wrapper that uses google-auth credentials.""" def __init__( - self, websession: aiohttp.ClientSession, credentials, host: str | None = None + self, + websession: aiohttp.ClientSession, + credentials, + host: str | None = None, + token_file: str | None = None, ) -> None: super().__init__(websession, host) self._credentials = credentials + self._token_file = token_file async def async_get_access_token(self) -> str: if not self._credentials.valid: loop = asyncio.get_running_loop() req = Request() await loop.run_in_executor(None, self._credentials.refresh, req) - save_credentials(self._credentials) + save_credentials(self._credentials, token_file=self._token_file) return self._credentials.token @@ -90,22 +95,28 @@ async def async_get_access_token(self) -> str: return self._token -def save_credentials(credentials) -> None: +def save_credentials(credentials, token_file: str | None = None) -> None: """Save credentials to local token file.""" - with open(TOKEN_FILE, "w") as f: + path = token_file or os.environ.get("GOOGLE_HEALTH_CLI_TOKEN_FILE") or TOKEN_FILE + with open(path, "w") as f: f.write(credentials.to_json()) -def load_credentials_or_env(): +def load_credentials_or_env( + token_file: str | None = None, + environ: dict[str, str] | None = None, +): """Load credentials from environment or token.json.""" - token_env = os.environ.get("GOOGLE_HEALTH_CLI_TOKEN") + env = environ if environ is not None else os.environ + token_env = env.get("GOOGLE_HEALTH_CLI_TOKEN") if token_env: return ("env", token_env) - if not os.path.exists(TOKEN_FILE): + path = token_file or env.get("GOOGLE_HEALTH_CLI_TOKEN_FILE") or TOKEN_FILE + if not os.path.exists(path): return None - with open(TOKEN_FILE, "r") as f: + with open(path, "r") as f: data = json.load(f) expiry_str = data.get("expiry") diff --git a/google_health_api/cli/subcommands/login.py b/google_health_api/cli/subcommands/login.py index 2893ca4..ee953b1 100644 --- a/google_health_api/cli/subcommands/login.py +++ b/google_health_api/cli/subcommands/login.py @@ -3,6 +3,7 @@ import json import os import sys +from collections.abc import Callable from google_auth_oauthlib.flow import Flow, InstalledAppFlow @@ -10,21 +11,29 @@ from ..utils import print_error_json, print_json -def cmd_login(args) -> None: +def cmd_login( + args, + client_secret_file: str | None = None, + token_file: str | None = None, + is_tty: bool | None = None, + input_func: Callable[[str], str] | None = None, +) -> None: """Execute interactive OAuth login flow.""" - if not os.path.exists(CLIENT_SECRET_FILE): + secret_path = client_secret_file or CLIENT_SECRET_FILE + if not os.path.exists(secret_path): print_error_json( - f"Client secrets file '{CLIENT_SECRET_FILE}' not found.", + f"Client secrets file '{secret_path}' not found.", status="NOT_FOUND", ) - if not sys.stdin.isatty(): + tty = is_tty if is_tty is not None else sys.stdin.isatty() + if not tty: print_error_json( "Cannot run interactive login in a headless environment.", status="FAILED_PRECONDITION", ) - with open(CLIENT_SECRET_FILE, "r") as f: + with open(secret_path, "r") as f: client_secrets_data = json.load(f) is_web = "web" in client_secrets_data @@ -34,7 +43,7 @@ def cmd_login(args) -> None: redirect_uri = redirect_uris[0] if redirect_uris else "http://localhost:8080/" flow = Flow.from_client_secrets_file( - CLIENT_SECRET_FILE, + secret_path, scopes=SCOPES, redirect_uri=redirect_uri, ) @@ -44,7 +53,8 @@ def cmd_login(args) -> None: ) print("Web-based authentication flow:") print(f"URL: {authorization_url}") - redirect_response = input("Redirected URL or auth code: ").strip() + inp = input_func or input + redirect_response = inp("Redirected URL or auth code: ").strip() if not redirect_response: print_error_json( @@ -61,10 +71,10 @@ def cmd_login(args) -> None: credentials = flow.credentials else: flow = InstalledAppFlow.from_client_secrets_file( - CLIENT_SECRET_FILE, + secret_path, scopes=SCOPES, ) credentials = flow.run_local_server(port=0) - save_credentials(credentials) + save_credentials(credentials, token_file=token_file) print_json({"status": "SUCCESS", "message": "Logged in successfully."}) diff --git a/tests/cli/subcommands/test_login.py b/tests/cli/subcommands/test_login.py index 66575c1..81d3502 100644 --- a/tests/cli/subcommands/test_login.py +++ b/tests/cli/subcommands/test_login.py @@ -1,6 +1,7 @@ """Tests for login and auth flow CLI commands.""" import json +from collections.abc import Callable from pathlib import Path from unittest.mock import MagicMock @@ -15,13 +16,29 @@ from tests.cli.conftest import run_cli -@pytest.mark.asyncio -async def test_credentials_auth_refresh( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch +def run_test_login( + args: MagicMock | None = None, + client_secret_file: str | None = None, + token_file: str | None = None, + is_tty: bool = True, + input_func: Callable[[str], str] | None = None, ) -> None: + """Helper wrapper to run cmd_login with default test settings.""" + if args is None: + args = MagicMock() + cmd_login( + args, + client_secret_file=client_secret_file, + token_file=token_file, + is_tty=is_tty, + input_func=input_func, + ) + + +@pytest.mark.asyncio +async def test_credentials_auth_refresh(tmp_path: Path) -> None: """Test CredentialsAuth refreshing expired credentials.""" token_file = tmp_path / "token.json" - monkeypatch.setattr("google_health_api.cli.auth.TOKEN_FILE", str(token_file)) creds_mock = MagicMock() creds_mock.valid = False @@ -29,7 +46,7 @@ async def test_credentials_auth_refresh( creds_mock.to_json.return_value = '{"token": "refreshed-token"}' session_mock = MagicMock() - auth = CredentialsAuth(session_mock, creds_mock) + auth = CredentialsAuth(session_mock, creds_mock, token_file=str(token_file)) token = await auth.async_get_access_token() assert token == "refreshed-token" @@ -46,20 +63,15 @@ async def test_env_auth() -> None: assert token == "env-secret-token" -def test_load_credentials_or_env( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> None: +def test_load_credentials_or_env(tmp_path: Path) -> None: """Test loading credentials from env vs token file.""" # 1. Test env var set - monkeypatch.setenv("GOOGLE_HEALTH_CLI_TOKEN", "env-tok") - res = load_credentials_or_env() + res = load_credentials_or_env(environ={"GOOGLE_HEALTH_CLI_TOKEN": "env-tok"}) assert res == ("env", "env-tok") # 2. Test no file and no env var - monkeypatch.delenv("GOOGLE_HEALTH_CLI_TOKEN", raising=False) token_file = tmp_path / "token.json" - monkeypatch.setattr("google_health_api.cli.auth.TOKEN_FILE", str(token_file)) - assert load_credentials_or_env() is None + assert load_credentials_or_env(token_file=str(token_file), environ={}) is None # 3. Test token file exists with Z expiry token_data = { @@ -68,7 +80,7 @@ def test_load_credentials_or_env( "expiry": "2026-12-31T23:59:59Z", } token_file.write_text(json.dumps(token_data)) - res_file = load_credentials_or_env() + res_file = load_credentials_or_env(token_file=str(token_file), environ={}) assert res_file is not None assert res_file[0] == "file" assert res_file[1].token == "file-tok" @@ -76,38 +88,28 @@ def test_load_credentials_or_env( # 4. Token file with non-Z iso string with offset token_data["expiry"] = "2026-12-31T23:59:59+00:00" token_file.write_text(json.dumps(token_data)) - res_file2 = load_credentials_or_env() + res_file2 = load_credentials_or_env(token_file=str(token_file), environ={}) assert res_file2 is not None assert res_file2[0] == "file" def test_cmd_login_missing_client_secret( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] + tmp_path: Path, capsys: pytest.CaptureFixture[str] ) -> None: """Test login failure when client secret file is missing.""" - monkeypatch.setattr( - "google_health_api.cli.subcommands.login.CLIENT_SECRET_FILE", - str(tmp_path / "nonexistent.json"), - ) with pytest.raises(SystemExit): - cmd_login(MagicMock()) + run_test_login(client_secret_file=str(tmp_path / "nonexistent.json")) captured = capsys.readouterr() assert "not found" in captured.out -def test_cmd_login_not_tty( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] -) -> None: +def test_cmd_login_not_tty(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: """Test login failure in non-interactive environment.""" secret_file = tmp_path / "client_secret.json" secret_file.write_text(json.dumps({"web": {}})) - monkeypatch.setattr( - "google_health_api.cli.subcommands.login.CLIENT_SECRET_FILE", str(secret_file) - ) - monkeypatch.setattr("sys.stdin.isatty", lambda: False) with pytest.raises(SystemExit): - cmd_login(MagicMock()) + run_test_login(client_secret_file=str(secret_file), is_tty=False) captured = capsys.readouterr() assert "headless environment" in captured.out @@ -115,7 +117,6 @@ def test_cmd_login_not_tty( def test_cmd_login_web_flow( mock_flow_cls: MagicMock, tmp_path: Path, - monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: """Test login web OAuth flow.""" @@ -123,48 +124,44 @@ def test_cmd_login_web_flow( secret_file.write_text( json.dumps({"web": {"redirect_uris": ["http://localhost:8080/"]}}) ) - monkeypatch.setattr( - "google_health_api.cli.subcommands.login.CLIENT_SECRET_FILE", str(secret_file) - ) - monkeypatch.setattr( - "google_health_api.cli.auth.TOKEN_FILE", str(tmp_path / "token.json") - ) - monkeypatch.setattr("sys.stdin.isatty", lambda: True) + token_file = tmp_path / "token.json" mock_flow = MagicMock() mock_flow_cls.from_client_secrets_file.return_value = mock_flow - mock_flow.authorization_url.return_value = ("https://auth.example.com", "state") + mock_flow.authorization_url.return_value = ( + "https://auth.example.com", + "state", + ) mock_creds = MagicMock() mock_creds.to_json.return_value = '{"token": "abc"}' mock_flow.credentials = mock_creds # Empty response -> error - monkeypatch.setattr("builtins.input", lambda prompt="": "") with pytest.raises(SystemExit): - cmd_login(MagicMock()) + run_test_login( + client_secret_file=str(secret_file), + token_file=str(token_file), + input_func=lambda prompt="": "", + ) # Valid auth code response - monkeypatch.setattr("builtins.input", lambda prompt="": "code=auth123") - cmd_login(MagicMock()) + run_test_login( + client_secret_file=str(secret_file), + token_file=str(token_file), + input_func=lambda prompt="": "code=auth123", + ) assert mock_flow.fetch_token.called def test_cmd_login_installed_app_flow( mock_installed_flow_cls: MagicMock, tmp_path: Path, - monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: """Test login installed app flow.""" secret_file = tmp_path / "client_secret.json" secret_file.write_text(json.dumps({"installed": {}})) - monkeypatch.setattr( - "google_health_api.cli.subcommands.login.CLIENT_SECRET_FILE", str(secret_file) - ) - monkeypatch.setattr( - "google_health_api.cli.auth.TOKEN_FILE", str(tmp_path / "token.json") - ) - monkeypatch.setattr("sys.stdin.isatty", lambda: True) + token_file = tmp_path / "token.json" mock_flow = MagicMock() mock_installed_flow_cls.from_client_secrets_file.return_value = mock_flow @@ -172,19 +169,19 @@ def test_cmd_login_installed_app_flow( mock_creds.to_json.return_value = '{"token": "abc"}' mock_flow.run_local_server.return_value = mock_creds - cmd_login(MagicMock()) + run_test_login( + client_secret_file=str(secret_file), + token_file=str(token_file), + ) captured = capsys.readouterr() assert "Logged in successfully" in captured.out def test_unauthenticated_cli_setup( - monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] + mock_load_credentials: MagicMock, capsys: pytest.CaptureFixture[str] ) -> None: """Test error when setup_client finds no auth token/credentials.""" - monkeypatch.delenv("GOOGLE_HEALTH_CLI_TOKEN", raising=False) - monkeypatch.setattr( - "google_health_api.cli.auth.TOKEN_FILE", "non_existent_token.json" - ) + mock_load_credentials.return_value = None with pytest.raises(SystemExit): run_cli(["userinfo"])