17249640673

Committed 26 Aug 2025 08:27PM UTC coverage: 92.699% (+0.03%) from 92.672%

Build # 17249640673

Build Type

Pull #7

github

Committed by

web-flow

Commit Message

Merge 4144c5b04 into 653847af2

Pull Request Pull Request #7: Refactor CLI

Run Details

209 of 259 new or added lines in 5 files covered. (80.69%)

3 existing lines in 1 file now uncovered.

838 of 904 relevant lines covered (92.7%)

0.93 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

89.55

/src/monzoh/cli/token_cache.py

"""Token caching and refresh functionality."""

import json
import os
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any

from rich.console import Console

from ..auth import MonzoOAuth
from ..models import OAuthToken


def get_token_cache_path() -> Path:
    """Get path for token cache file."""
    import platform

    system = platform.system()

    if system == "Windows":
        # Use %LOCALAPPDATA% on Windows
        cache_dir = (
            Path(os.getenv("LOCALAPPDATA", Path.home() / "AppData" / "Local"))
            / "monzoh"
        )
    elif system == "Darwin":  # macOS
        # Use ~/Library/Caches on macOS
        cache_dir = Path.home() / "Library" / "Caches" / "monzoh"
    else:
        # Linux and other Unix-like systems: use XDG_CACHE_HOME or ~/.cache
        cache_dir = Path(os.getenv("XDG_CACHE_HOME", Path.home() / ".cache")) / "monzoh"

    cache_dir.mkdir(parents=True, exist_ok=True)
    return cache_dir / "tokens.json"


def save_token_to_cache(token: OAuthToken, console: Console) -> None:
    """Save token to cache file."""
    try:
        cache_path = get_token_cache_path()

        # Calculate expiry time
        expires_at = datetime.now() + timedelta(seconds=token.expires_in)

        cache_data = {
            "access_token": token.access_token,
            "refresh_token": token.refresh_token,
            "expires_at": expires_at.isoformat(),
            "user_id": token.user_id,
            "client_id": token.client_id,
        }

        with open(cache_path, "w") as f:
            json.dump(cache_data, f, indent=2)

        # Set restrictive permissions (readable only by owner)
        try:
            cache_path.chmod(0o600)
        except OSError:
            # On Windows, chmod might not work as expected
            # The file is still protected by the user's directory permissions
            pass

        console.print(f"💾 Token cached to [green]{cache_path}[/green]")

    except Exception as e:
        console.print(f"⚠️  [yellow]Warning: Could not cache token: {e}[/yellow]")


def load_token_from_cache(include_expired: bool = False) -> dict[str, Any] | None:
    """Load token from cache file.

    Args:
        include_expired: If True, return expired tokens (useful for refresh)
    """
    try:
        cache_path = get_token_cache_path()

        if not cache_path.exists():
            return None

        with open(cache_path) as f:
            cache_data: dict[str, Any] = json.load(f)

        # Check if token has expired
        if not include_expired:
            expires_at = datetime.fromisoformat(cache_data["expires_at"])
            if datetime.now() >= expires_at - timedelta(minutes=5):  # 5 min buffer
                return None

        return cache_data

    except (OSError, ValueError, TypeError, KeyError, FileNotFoundError):
        return None


def clear_token_cache() -> None:
    """Clear the token cache."""
    try:
        cache_path = get_token_cache_path()
        if cache_path.exists():
            cache_path.unlink()
    except (OSError, ValueError, TypeError, KeyError, FileNotFoundError):
        pass


def try_refresh_token(
    cached_token: dict[str, Any], oauth: MonzoOAuth, console: Console
) -> str | None:
    """Try to refresh an expired token."""
    if not cached_token.get("refresh_token"):
        return None

    try:
        console.print("🔄 Refreshing expired access token...")

        with oauth:
            new_token = oauth.refresh_token(cached_token["refresh_token"])

        save_token_to_cache(new_token, console)
        console.print("✅ [green]Token refreshed successfully![/green]")
        return new_token.access_token

    except Exception as e:
        console.print(f"⚠️  [yellow]Token refresh failed: {e}[/yellow]")
        clear_token_cache()
        return None

1	"""Token caching and refresh functionality."""
2
3	import json	1✔
4	import os	1✔
5	from datetime import datetime, timedelta	1✔
6	from pathlib import Path	1✔
7	from typing import Any	1✔
8
9	from rich.console import Console	1✔
10
11	from ..auth import MonzoOAuth	1✔
12	from ..models import OAuthToken	1✔
13
14
15	def get_token_cache_path() -> Path:	1✔
16	"""Get path for token cache file."""
17	import platform	1✔
18
19	system = platform.system()	1✔
20
21	if system == "Windows":	1✔
22	# Use %LOCALAPPDATA% on Windows
NEW 23	cache_dir = (	×
24	Path(os.getenv("LOCALAPPDATA", Path.home() / "AppData" / "Local"))
25	/ "monzoh"
26	)
27	elif system == "Darwin": # macOS	1✔
28	# Use ~/Library/Caches on macOS
29	cache_dir = Path.home() / "Library" / "Caches" / "monzoh"	1✔
30	else:
31	# Linux and other Unix-like systems: use XDG_CACHE_HOME or ~/.cache
32	cache_dir = Path(os.getenv("XDG_CACHE_HOME", Path.home() / ".cache")) / "monzoh"	1✔
33
34	cache_dir.mkdir(parents=True, exist_ok=True)	1✔
35	return cache_dir / "tokens.json"	1✔
36
37
38	def save_token_to_cache(token: OAuthToken, console: Console) -> None:	1✔
39	"""Save token to cache file."""
40	try:	1✔
41	cache_path = get_token_cache_path()	1✔
42
43	# Calculate expiry time
44	expires_at = datetime.now() + timedelta(seconds=token.expires_in)	1✔
45
46	cache_data = {	1✔
47	"access_token": token.access_token,
48	"refresh_token": token.refresh_token,
49	"expires_at": expires_at.isoformat(),
50	"user_id": token.user_id,
51	"client_id": token.client_id,
52	}
53
54	with open(cache_path, "w") as f:	1✔
55	json.dump(cache_data, f, indent=2)	1✔
56
57	# Set restrictive permissions (readable only by owner)
58	try:	1✔
59	cache_path.chmod(0o600)	1✔
NEW 60	except OSError:	×
61	# On Windows, chmod might not work as expected
62	# The file is still protected by the user's directory permissions
NEW 63	pass	×
64
65	console.print(f"💾 Token cached to [green]{cache_path}[/green]")	1✔
66
67	except Exception as e:	1✔
68	console.print(f"⚠️ [yellow]Warning: Could not cache token: {e}[/yellow]")	1✔
69
70
71	def load_token_from_cache(include_expired: bool = False) -> dict[str, Any] \| None:	1✔
72	"""Load token from cache file.
73
74	Args:
75	include_expired: If True, return expired tokens (useful for refresh)
76	"""
77	try:	1✔
78	cache_path = get_token_cache_path()	1✔
79
80	if not cache_path.exists():	1✔
81	return None	1✔
82
83	with open(cache_path) as f:	1✔
84	cache_data: dict[str, Any] = json.load(f)	1✔
85
86	# Check if token has expired
87	if not include_expired:	1✔
88	expires_at = datetime.fromisoformat(cache_data["expires_at"])	1✔
89	if datetime.now() >= expires_at - timedelta(minutes=5): # 5 min buffer	1✔
90	return None	1✔
91
92	return cache_data	1✔
93
NEW 94	except (OSError, ValueError, TypeError, KeyError, FileNotFoundError):	×
NEW 95	return None	×
96
97
98	def clear_token_cache() -> None:	1✔
99	"""Clear the token cache."""
100	try:	1✔
101	cache_path = get_token_cache_path()	1✔
102	if cache_path.exists():	1✔
103	cache_path.unlink()	1✔
NEW 104	except (OSError, ValueError, TypeError, KeyError, FileNotFoundError):	×
NEW 105	pass	×
106
107
108	def try_refresh_token(	1✔
109	cached_token: dict[str, Any], oauth: MonzoOAuth, console: Console
110	) -> str \| None:
111	"""Try to refresh an expired token."""
112	if not cached_token.get("refresh_token"):	1✔
113	return None	1✔
114
115	try:	1✔
116	console.print("🔄 Refreshing expired access token...")	1✔
117
118	with oauth:	1✔
119	new_token = oauth.refresh_token(cached_token["refresh_token"])	1✔
120
121	save_token_to_cache(new_token, console)	1✔
122	console.print("✅ [green]Token refreshed successfully![/green]")	1✔
123	return new_token.access_token	1✔
124
125	except Exception as e:	1✔
126	console.print(f"⚠️ [yellow]Token refresh failed: {e}[/yellow]")	1✔
127	clear_token_cache()	1✔
128	return None	1✔

samdobson / monzoh / 17249640673

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous