17826977812

Committed 18 Sep 2025 11:18AM UTC coverage: 89.877%. Remained the same

Build # 17826977812

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #28 from samdobson/fix/no-comments

Remove unnecessary comments

Run Details

8 of 9 new or added lines in 6 files covered. (88.89%)

16 existing lines in 7 files now uncovered.

1465 of 1630 relevant lines covered (89.88%)

0.9 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":
        cache_dir = (
            Path(os.getenv("LOCALAPPDATA", Path.home() / "AppData" / "Local"))
            / "monzoh"
        )
    elif system == "Darwin":
        cache_dir = Path.home() / "Library" / "Caches" / "monzoh"
    else:
        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()

        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)

        try:
            cache_path.chmod(0o600)
        except OSError:
            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)

    Returns:
        Cached token data if available and valid, None otherwise
    """
    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)

        if not include_expired:
            expires_at = datetime.fromisoformat(cache_data["expires_at"])
            if datetime.now() >= expires_at - timedelta(minutes=5):
                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✔
UNCOV 22	cache_dir = (	×
23	Path(os.getenv("LOCALAPPDATA", Path.home() / "AppData" / "Local"))
24	/ "monzoh"
25	)
26	elif system == "Darwin":	1✔
27	cache_dir = Path.home() / "Library" / "Caches" / "monzoh"	1✔
28	else:
29	cache_dir = Path(os.getenv("XDG_CACHE_HOME", Path.home() / ".cache")) / "monzoh"	1✔
30
31	cache_dir.mkdir(parents=True, exist_ok=True)	1✔
32	return cache_dir / "tokens.json"	1✔
33
34
35	def save_token_to_cache(token: OAuthToken, console: Console) -> None:	1✔
36	"""Save token to cache file."""
37	try:	1✔
38	cache_path = get_token_cache_path()	1✔
39
40	expires_at = datetime.now() + timedelta(seconds=token.expires_in)	1✔
41
42	cache_data = {	1✔
43	"access_token": token.access_token,
44	"refresh_token": token.refresh_token,
45	"expires_at": expires_at.isoformat(),
46	"user_id": token.user_id,
47	"client_id": token.client_id,
48	}
49
50	with open(cache_path, "w") as f:	1✔
51	json.dump(cache_data, f, indent=2)	1✔
52
53	try:	1✔
54	cache_path.chmod(0o600)	1✔
55	except OSError:	×
UNCOV 56	pass	×
57
58	console.print(f"💾 Token cached to [green]{cache_path}[/green]")	1✔
59
60	except Exception as e:	1✔
61	console.print(f"⚠️ [yellow]Warning: Could not cache token: {e}[/yellow]")	1✔
62
63
64	def load_token_from_cache(include_expired: bool = False) -> dict[str, Any] \| None:	1✔
65	"""Load token from cache file.
66
67	Args:
68	include_expired: If True, return expired tokens (useful for refresh)
69
70	Returns:
71	Cached token data if available and valid, None otherwise
72	"""
73	try:	1✔
74	cache_path = get_token_cache_path()	1✔
75
76	if not cache_path.exists():	1✔
77	return None	1✔
78
79	with open(cache_path) as f:	1✔
80	cache_data: dict[str, Any] = json.load(f)	1✔
81
82	if not include_expired:	1✔
83	expires_at = datetime.fromisoformat(cache_data["expires_at"])	1✔
84	if datetime.now() >= expires_at - timedelta(minutes=5):	1✔
85	return None	1✔
86
87	return cache_data	1✔
88
89	except (OSError, ValueError, TypeError, KeyError, FileNotFoundError):	×
90	return None	×
91
92
93	def clear_token_cache() -> None:	1✔
94	"""Clear the token cache."""
95	try:	1✔
96	cache_path = get_token_cache_path()	1✔
97	if cache_path.exists():	1✔
98	cache_path.unlink()	1✔
99	except (OSError, ValueError, TypeError, KeyError, FileNotFoundError):	×
100	pass	×
101
102
103	def try_refresh_token(	1✔
104	cached_token: dict[str, Any], oauth: MonzoOAuth, console: Console
105	) -> str \| None:
106	"""Try to refresh an expired token."""
107	if not cached_token.get("refresh_token"):	1✔
108	return None	1✔
109
110	try:	1✔
111	console.print("🔄 Refreshing expired access token...")	1✔
112
113	with oauth:	1✔
114	new_token = oauth.refresh_token(cached_token["refresh_token"])	1✔
115
116	save_token_to_cache(new_token, console)	1✔
117	console.print("✅ [green]Token refreshed successfully![/green]")	1✔
118	return new_token.access_token	1✔
119
120	except Exception as e:	1✔
121	console.print(f"⚠️ [yellow]Token refresh failed: {e}[/yellow]")	1✔
122	clear_token_cache()	1✔
123	return None	1✔

samdobson / monzoh / 17826977812

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