diff --git a/README.md b/README.md index 47c26b9..1d9cb5c 100644 --- a/README.md +++ b/README.md @@ -61,6 +61,32 @@ The command prints a Google OAuth URL. Open that URL in a browser on any machine access. The browser will then fail to load a `127.0.0.1` page; this is expected. Copy the full URL from the browser address bar, paste it back into the terminal, and ytstudio will finish the login. +## Managing multiple channels + +If you run more than one YouTube channel, ytstudio stores each login under its own named profile. +Every command (`videos`, `analytics`, `comments`, ...) operates on the active channel. + +```bash +ytstudio profile add work # authenticate a new channel and make it active +ytstudio profile add personal # add another +ytstudio profile list # show profiles; the active one is marked with * +ytstudio profile use work # switch the active profile +ytstudio profile status work # auth status for one profile (defaults to active) +ytstudio profile remove personal +``` + +For scripting you can override the active channel per command without switching: + +```bash +YTSTUDIO_PROFILE=work ytstudio videos list +``` + +Credentials live in `~/.config/ytstudio-cli/profiles//` with owner-only (`600`) permissions. +The shared OAuth client secrets stay at the top level, so `ytstudio init` is only needed once. +A plain `ytstudio login` (no `profile add`) authenticates the active profile, which is `default` +on a fresh setup. Existing single-channel installs are migrated to the `default` profile +automatically on first run. + ## API quota The YouTube Data API enforces a default quota of 10_000 units per project per day. Most read diff --git a/src/ytstudio/api.py b/src/ytstudio/api.py index 876e23d..48f11f6 100644 --- a/src/ytstudio/api.py +++ b/src/ytstudio/api.py @@ -12,8 +12,10 @@ from ytstudio.config import ( CLIENT_SECRETS_FILE, clear_credentials, + get_active_profile, load_credentials, save_credentials, + save_profile_meta, ) from ytstudio.ui import console, success_message @@ -73,7 +75,7 @@ def _create_flow() -> InstalledAppFlow: ) -def _save_credentials(credentials: Credentials) -> None: +def _save_credentials(credentials: Credentials, profile: str | None = None) -> None: creds_data = { "token": credentials.token, "refresh_token": credentials.refresh_token, @@ -82,17 +84,36 @@ def _save_credentials(credentials: Credentials) -> None: "client_secret": credentials.client_secret, "scopes": credentials.scopes, } - save_credentials(creds_data) + save_credentials(creds_data, profile) -def _show_login_success(credentials: Credentials) -> None: - service = build("youtube", "v3", credentials=credentials) - response = service.channels().list(part="snippet", mine=True).execute() +def _fetch_channel_info(credentials: Credentials) -> dict | None: + # Best-effort: a quota/network error here must not fail a successful login. + try: + service = build("youtube", "v3", credentials=credentials) + response = service.channels().list(part="snippet", mine=True).execute() + except Exception: + return None + + items = response.get("items") + if not items: + return None + + snippet = items[0]["snippet"] + return { + "id": items[0].get("id", ""), + "title": snippet.get("title", ""), + "custom_url": snippet.get("customUrl", ""), + } + + +def _show_login_success(credentials: Credentials, profile: str) -> None: + info = _fetch_channel_info(credentials) console.print() - if response.get("items"): - channel = response["items"][0]["snippet"] - success_message(f"Logged in as: {channel['title']}") + if info: + save_profile_meta(profile, info) + success_message(f"Logged in as: {info['title']}") else: success_message("Authentication successful") @@ -143,11 +164,15 @@ def _authenticate_headless() -> Credentials: return flow.credentials -def authenticate(headless: bool = False) -> None: +def authenticate(headless: bool = False, profile: str | None = None) -> None: if not CLIENT_SECRETS_FILE.exists(): console.print("[red]No client secrets found. Run 'ytstudio init' first.[/red]") raise SystemExit(1) from None + # Resolve the target profile up-front: a profile switch during a long OAuth + # flow must not redirect the freshly minted credentials to a different one. + profile = profile or get_active_profile() + console.print("[bold]Authenticating with YouTube...[/bold]\n") if headless: @@ -160,12 +185,13 @@ def authenticate(headless: bool = False) -> None: open_browser=True, ) - _save_credentials(credentials) - _show_login_success(credentials) + _save_credentials(credentials, profile) + _show_login_success(credentials, profile) -def get_credentials() -> Credentials | None: - creds_data = load_credentials() +def get_credentials(profile: str | None = None) -> Credentials | None: + profile = profile or get_active_profile() + creds_data = load_credentials(profile) if not creds_data: return None @@ -188,27 +214,31 @@ def get_credentials() -> Credentials | None: raise SystemExit(1) from None # Save refreshed credentials creds_data["token"] = credentials.token - save_credentials(creds_data) + save_credentials(creds_data, profile) return credentials -def get_authenticated_service(api_name: str = "youtube", version: str = "v3"): - credentials = get_credentials() +def get_authenticated_service( + api_name: str = "youtube", + version: str = "v3", + profile: str | None = None, +): + credentials = get_credentials(profile) if not credentials: console.print("[red]Not authenticated. Run 'ytstudio login' first.[/red]") raise typer.Exit(1) return build(api_name, version, credentials=credentials) -def get_status() -> None: - creds_data = load_credentials() +def get_status(profile: str | None = None) -> None: + creds_data = load_credentials(profile) if not creds_data: console.print("[yellow]Not authenticated. Run 'ytstudio login' to authenticate.[/yellow]") return - credentials = get_credentials() + credentials = get_credentials(profile) if not credentials or not credentials.valid: console.print( "[yellow]Credentials expired. Run 'ytstudio login' to re-authenticate.[/yellow]" diff --git a/src/ytstudio/commands/profile.py b/src/ytstudio/commands/profile.py new file mode 100644 index 0000000..11727e9 --- /dev/null +++ b/src/ytstudio/commands/profile.py @@ -0,0 +1,96 @@ +import typer + +from ytstudio import config +from ytstudio.api import authenticate, get_status +from ytstudio.ui import console, create_table, dim, success_message + +app = typer.Typer(help="Manage credential profiles (one per YouTube channel)") + + +@app.command() +def add( + name: str = typer.Argument(help="Name for the new profile"), + headless: bool = typer.Option( + False, + "--headless", + help="Authenticate by pasting a redirect URL from another browser", + ), +): + """Authenticate a new profile and switch to it""" + if not config.is_valid_profile_name(name): + console.print( + "[red]Invalid name. Use letters, digits, '-' and '_' " + "(must start with a letter or digit).[/red]" + ) + raise typer.Exit(1) + if config.profile_exists(name): + console.print( + f"[red]Profile '{name}' already exists. " + f"Switch with 'ytstudio profile use {name}'.[/red]" + ) + raise typer.Exit(1) + + authenticate(headless=headless, profile=name) + config.set_active_profile(name) + success_message(f"Profile '{name}' is now active.") + + +@app.command("list") +def list_profiles(): + """List configured profiles""" + profiles = config.list_profiles() + if not profiles: + console.print("No profiles yet. Run 'ytstudio login' or 'ytstudio profile add '.") + return + + active = config.get_active_profile() + table = create_table() + table.add_column("") + table.add_column("Name") + table.add_column("YouTube channel") + table.add_column("Handle") + + for name in profiles: + meta = config.load_profile_meta(name) + marker = "[green]*[/green]" if name == active else " " + title = meta.get("title") or dim("unknown (run login for this profile)") + handle = meta.get("custom_url") or "" + table.add_row(marker, name, title, handle) + + console.print(table) + + +@app.command() +def use(name: str = typer.Argument(help="Profile to make active")): + """Switch the active profile""" + if not config.profile_exists(name): + console.print(f"[red]No profile named '{name}'. See 'ytstudio profile list'.[/red]") + raise typer.Exit(1) + config.set_active_profile(name) + success_message(f"Active profile: {name}") + + +@app.command() +def status(name: str = typer.Argument(None, help="Profile name (default: active)")): + """Show authentication status for a profile""" + target = name or config.get_active_profile() + if not config.profile_exists(target): + console.print(f"[red]No profile named '{target}'. See 'ytstudio profile list'.[/red]") + raise typer.Exit(1) + console.print(f"[bold]Profile:[/bold] {target}") + get_status(target) + + +@app.command() +def remove( + name: str = typer.Argument(help="Profile to remove"), + force: bool = typer.Option(False, "--force", "-f", help="Skip confirmation"), +): + """Remove a profile and its stored credentials""" + if not config.profile_exists(name): + console.print(f"[red]No profile named '{name}'.[/red]") + raise typer.Exit(1) + if not force and not typer.confirm(f"Remove profile '{name}' and its stored credentials?"): + raise typer.Exit() + config.remove_profile(name) + success_message(f"Profile '{name}' removed.") diff --git a/src/ytstudio/config.py b/src/ytstudio/config.py index 05bfa69..ca177d0 100644 --- a/src/ytstudio/config.py +++ b/src/ytstudio/config.py @@ -1,4 +1,9 @@ +import fcntl import json +import os +import re +import shutil +from contextlib import contextmanager from pathlib import Path from rich.prompt import Prompt @@ -7,11 +12,76 @@ CONFIG_DIR = Path.home() / ".config" / "ytstudio-cli" CLIENT_SECRETS_FILE = CONFIG_DIR / "client_secrets.json" -CREDENTIALS_FILE = CONFIG_DIR / "credentials.json" +PROFILES_DIR = CONFIG_DIR / "profiles" +STATE_FILE = CONFIG_DIR / "state.json" + +# Pre-profiles single credentials file, kept only for one-shot migration. +LEGACY_CREDENTIALS_FILE = CONFIG_DIR / "credentials.json" + +DEFAULT_PROFILE = "default" +PROFILE_ENV_VAR = "YTSTUDIO_PROFILE" + +_VALID_PROFILE_NAME = re.compile(r"[A-Za-z0-9][A-Za-z0-9_-]*") + + +def is_valid_profile_name(name: str) -> bool: + return bool(_VALID_PROFILE_NAME.fullmatch(name)) + + +def _ensure_private_dir(path: Path) -> None: + path.mkdir(parents=True, exist_ok=True) + path.chmod(0o700) def ensure_config_dir() -> None: - CONFIG_DIR.mkdir(parents=True, exist_ok=True) + _ensure_private_dir(CONFIG_DIR) + + +def _ensure_profile_dir(name: str) -> Path: + ensure_config_dir() + _ensure_private_dir(PROFILES_DIR) + target = profile_dir(name) + _ensure_private_dir(target) + return target + + +def _write_private(path: Path, text: str) -> None: + """Atomically write a secret with owner-only permissions, with no readable window.""" + tmp = path.with_name(path.name + ".tmp") + fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + try: + os.write(fd, text.encode()) + os.fsync(fd) + finally: + os.close(fd) + tmp.replace(path) + + +def _atomic_write_text(path: Path, text: str, mode: int = 0o644) -> None: + tmp = path.with_name(path.name + ".tmp") + fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, mode) + try: + os.write(fd, text.encode()) + os.fsync(fd) + finally: + os.close(fd) + tmp.replace(path) + + +@contextmanager +def _config_lock(): + """Serialize state mutations and the legacy migration across CLI invocations.""" + ensure_config_dir() + lock_path = CONFIG_DIR / ".lock" + with lock_path.open("w") as fh: + try: + fcntl.flock(fh.fileno(), fcntl.LOCK_EX) + yield + finally: + fcntl.flock(fh.fileno(), fcntl.LOCK_UN) + + +# --- client secrets (shared OAuth app, identical for every profile) --- def setup_credentials(client_secrets_file: str | None = None) -> None: @@ -24,7 +94,7 @@ def setup_credentials(client_secrets_file: str | None = None) -> None: console.print(f"[red]File not found: {client_secrets_file}[/red]") raise SystemExit(1) - CLIENT_SECRETS_FILE.write_text(source.read_text()) + _write_private(CLIENT_SECRETS_FILE, source.read_text()) success_message(f"Client secrets saved to {CLIENT_SECRETS_FILE}") else: # Interactive setup @@ -45,7 +115,7 @@ def setup_credentials(client_secrets_file: str | None = None) -> None: } } - CLIENT_SECRETS_FILE.write_text(json.dumps(secrets, indent=2)) + _write_private(CLIENT_SECRETS_FILE, json.dumps(secrets, indent=2)) console.print() success_message(f"Client secrets saved to {CLIENT_SECRETS_FILE}") @@ -58,17 +128,135 @@ def get_client_secrets() -> dict | None: return json.loads(CLIENT_SECRETS_FILE.read_text()) -def save_credentials(credentials: dict) -> None: +# --- profiles (one named credential set per YouTube channel) --- + + +def _load_state() -> dict: + if not STATE_FILE.exists(): + return {} + try: + return json.loads(STATE_FILE.read_text()) + except json.JSONDecodeError: + return {} + + +def _save_state(state: dict) -> None: ensure_config_dir() - CREDENTIALS_FILE.write_text(json.dumps(credentials, indent=2)) + _atomic_write_text(STATE_FILE, json.dumps(state, indent=2)) + + +def get_active_profile() -> str: + env = os.environ.get(PROFILE_ENV_VAR) + if env: + if is_valid_profile_name(env): + return env + console.print( + f"[yellow]Ignoring invalid {PROFILE_ENV_VAR}='{env}'; " + f"using the configured active channel.[/yellow]" + ) + + name = _load_state().get("active_profile", DEFAULT_PROFILE) + return name if is_valid_profile_name(name) else DEFAULT_PROFILE + +def set_active_profile(name: str) -> None: + with _config_lock(): + state = _load_state() + state["active_profile"] = name + _save_state(state) -def load_credentials() -> dict | None: - if not CREDENTIALS_FILE.exists(): + +def profile_dir(name: str) -> Path: + if not is_valid_profile_name(name): + raise ValueError(f"Invalid profile name: {name!r}") + return PROFILES_DIR / name + + +def credentials_path(name: str | None = None) -> Path: + return profile_dir(name or get_active_profile()) / "credentials.json" + + +def _meta_path(name: str) -> Path: + return profile_dir(name) / "meta.json" + + +def list_profiles() -> list[str]: + if not PROFILES_DIR.exists(): + return [] + return sorted( + p.name for p in PROFILES_DIR.iterdir() if p.is_dir() and is_valid_profile_name(p.name) + ) + + +def profile_exists(name: str) -> bool: + return is_valid_profile_name(name) and profile_dir(name).is_dir() + + +def remove_profile(name: str) -> None: + target = profile_dir(name) + if target.exists(): + shutil.rmtree(target) + + with _config_lock(): + state = _load_state() + if state.get("active_profile") == name: + state.pop("active_profile", None) + _save_state(state) + + +def save_credentials(credentials: dict, name: str | None = None) -> None: + target = _ensure_profile_dir(name or get_active_profile()) + _write_private(target / "credentials.json", json.dumps(credentials, indent=2)) + + +def load_credentials(name: str | None = None) -> dict | None: + path = credentials_path(name) + if not path.exists(): + return None + try: + return json.loads(path.read_text()) + except json.JSONDecodeError: return None - return json.loads(CREDENTIALS_FILE.read_text()) -def clear_credentials() -> None: - if CREDENTIALS_FILE.exists(): - CREDENTIALS_FILE.unlink() +def clear_credentials(name: str | None = None) -> None: + path = credentials_path(name) + if path.exists(): + path.unlink() + + +def save_profile_meta(name: str, meta: dict) -> None: + _ensure_profile_dir(name) + _meta_path(name).write_text(json.dumps(meta, indent=2)) + + +def load_profile_meta(name: str) -> dict: + path = _meta_path(name) + if not path.exists(): + return {} + try: + return json.loads(path.read_text()) + except json.JSONDecodeError: + return {} + + +def migrate_legacy_credentials() -> bool: + """Move a pre-profiles credentials.json into the default profile. One-shot. + + Held under the config lock so concurrent first-run invocations cannot both + pass the existence check and race on read/unlink. + """ + if not LEGACY_CREDENTIALS_FILE.exists() or list_profiles(): + return False + + with _config_lock(): + if not LEGACY_CREDENTIALS_FILE.exists() or list_profiles(): + return False + + dest = _ensure_profile_dir(DEFAULT_PROFILE) + _write_private(dest / "credentials.json", LEGACY_CREDENTIALS_FILE.read_text()) + LEGACY_CREDENTIALS_FILE.unlink() + state = _load_state() + state["active_profile"] = DEFAULT_PROFILE + _save_state(state) + return True diff --git a/src/ytstudio/main.py b/src/ytstudio/main.py index 835562c..54650c2 100644 --- a/src/ytstudio/main.py +++ b/src/ytstudio/main.py @@ -4,8 +4,8 @@ from rich.console import Console from ytstudio.api import authenticate, get_status -from ytstudio.commands import analytics, comments, videos -from ytstudio.config import setup_credentials +from ytstudio.commands import analytics, comments, profile, videos +from ytstudio.config import migrate_legacy_credentials, setup_credentials from ytstudio.version import get_current_version, is_update_available app = typer.Typer( @@ -20,6 +20,7 @@ app.add_typer(videos.app, name="videos") app.add_typer(analytics.app, name="analytics") app.add_typer(comments.app, name="comments") +app.add_typer(profile.app, name="profile") @app.command() @@ -77,6 +78,8 @@ def main( console.print(f"ytstudio v{get_current_version()}") raise typer.Exit() + migrate_legacy_credentials() + if not _update_state["registered"]: atexit.register(_show_update_notification) _update_state["registered"] = True diff --git a/src/ytstudio/services.py b/src/ytstudio/services.py index 1756fb9..2873af9 100644 --- a/src/ytstudio/services.py +++ b/src/ytstudio/services.py @@ -2,13 +2,13 @@ from ytstudio.demo import DemoAnalyticsService, DemoDataService, is_demo_mode -def get_data_service(): +def get_data_service(profile: str | None = None): if is_demo_mode(): return DemoDataService() - return get_authenticated_service("youtube", "v3") + return get_authenticated_service("youtube", "v3", profile=profile) -def get_analytics_service(): +def get_analytics_service(profile: str | None = None): if is_demo_mode(): return DemoAnalyticsService() - return get_authenticated_service("youtubeAnalytics", "v2") + return get_authenticated_service("youtubeAnalytics", "v2", profile=profile) diff --git a/tests/test_api.py b/tests/test_api.py index 397a4ed..b3de5bc 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -47,6 +47,17 @@ def test_exits_when_no_credentials(self): with patch("ytstudio.api.get_credentials", return_value=None), pytest.raises(Exit): get_authenticated_service() + def test_passes_profile_to_get_credentials(self): + credentials = MagicMock() + with ( + patch("ytstudio.api.get_credentials", return_value=credentials) as get_creds, + patch("ytstudio.api.build") as build, + ): + get_authenticated_service("youtube", "v3", profile="work") + + get_creds.assert_called_once_with("work") + build.assert_called_once_with("youtube", "v3", credentials=credentials) + class TestCommands: def test_login_requires_client_secrets(self): @@ -77,6 +88,7 @@ def test_normal_login_uses_local_server(self): with ( patch("ytstudio.api.CLIENT_SECRETS_FILE") as mock_file, patch("ytstudio.api._create_flow", return_value=flow), + patch("ytstudio.api.get_active_profile", return_value="default"), patch("ytstudio.api._save_credentials") as save_credentials, patch("ytstudio.api._show_login_success") as show_login_success, ): @@ -89,8 +101,29 @@ def test_normal_login_uses_local_server(self): prompt="consent", open_browser=True, ) - save_credentials.assert_called_once_with(credentials) - show_login_success.assert_called_once_with(credentials) + save_credentials.assert_called_once_with(credentials, "default") + show_login_success.assert_called_once_with(credentials, "default") + + def test_login_resolves_active_profile_once(self): + """A profile switch mid-OAuth must not redirect freshly minted credentials.""" + credentials = MagicMock() + flow = MagicMock() + flow.run_local_server.return_value = credentials + + with ( + patch("ytstudio.api.CLIENT_SECRETS_FILE") as mock_file, + patch("ytstudio.api._create_flow", return_value=flow), + patch("ytstudio.api.get_active_profile") as get_active, + patch("ytstudio.api._save_credentials") as save_credentials, + patch("ytstudio.api._show_login_success") as show_login_success, + ): + mock_file.exists.return_value = True + get_active.side_effect = ["work", "personal", "personal"] + + api_module.authenticate() + + save_credentials.assert_called_once_with(credentials, "work") + show_login_success.assert_called_once_with(credentials, "work") def test_headless_login_exchanges_pasted_redirect_url(self): credentials = MagicMock() @@ -103,6 +136,7 @@ def test_headless_login_exchanges_pasted_redirect_url(self): with ( patch("ytstudio.api.CLIENT_SECRETS_FILE") as mock_file, patch("ytstudio.api._create_flow", return_value=flow), + patch("ytstudio.api.get_active_profile", return_value="default"), patch("ytstudio.api.Prompt.ask", return_value=redirect_url), patch("ytstudio.api._save_credentials") as save_credentials, patch("ytstudio.api._show_login_success") as show_login_success, @@ -114,8 +148,8 @@ def test_headless_login_exchanges_pasted_redirect_url(self): assert flow.redirect_uri == api_module.HEADLESS_REDIRECT_URI flow.authorization_url.assert_called_once_with(prompt="consent") flow.fetch_token.assert_called_once_with(code="test-code") - save_credentials.assert_called_once_with(credentials) - show_login_success.assert_called_once_with(credentials) + save_credentials.assert_called_once_with(credentials, "default") + show_login_success.assert_called_once_with(credentials, "default") def test_headless_login_rejects_missing_code(self): with pytest.raises(SystemExit): diff --git a/tests/test_config.py b/tests/test_config.py index 111f84b..1c8838a 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1,3 +1,5 @@ +import json +import stat from unittest.mock import patch import pytest @@ -6,14 +8,18 @@ @pytest.fixture -def temp_config(tmp_path): +def temp_config(tmp_path, monkeypatch): config_dir = tmp_path / ".config" / "ytstudio-cli" config_dir.mkdir(parents=True) + monkeypatch.delenv(config.PROFILE_ENV_VAR, raising=False) + with ( patch.object(config, "CONFIG_DIR", config_dir), patch.object(config, "CLIENT_SECRETS_FILE", config_dir / "client_secrets.json"), - patch.object(config, "CREDENTIALS_FILE", config_dir / "credentials.json"), + patch.object(config, "PROFILES_DIR", config_dir / "profiles"), + patch.object(config, "STATE_FILE", config_dir / "state.json"), + patch.object(config, "LEGACY_CREDENTIALS_FILE", config_dir / "credentials.json"), ): yield config_dir @@ -32,6 +38,21 @@ def test_clear(self, temp_config): config.clear_credentials() assert config.load_credentials() is None + def test_credentials_file_is_private(self, temp_config): + config.save_credentials({"token": "secret"}) + mode = stat.S_IMODE(config.credentials_path().stat().st_mode) + assert mode == 0o600 + + def test_profile_dir_is_private(self, temp_config): + config.save_credentials({"token": "secret"}, name="work") + mode = stat.S_IMODE(config.profile_dir("work").stat().st_mode) + assert mode == 0o700 + + def test_load_returns_none_on_corrupt_json(self, temp_config): + config.save_credentials({"token": "ok"}, name="work") + config.credentials_path("work").write_text("{not json") + assert config.load_credentials("work") is None + class TestSetupCredentials: def test_setup_with_file(self, temp_config, tmp_path): @@ -45,3 +66,143 @@ def test_setup_with_file(self, temp_config, tmp_path): def test_setup_with_missing_file(self, temp_config): with pytest.raises(SystemExit): config.setup_credentials("/nonexistent/file.json") + + +class TestActiveProfile: + def test_default_when_unset(self, temp_config): + assert config.get_active_profile() == config.DEFAULT_PROFILE + + def test_set_and_get(self, temp_config): + config.set_active_profile("work") + assert config.get_active_profile() == "work" + + def test_env_override_wins(self, temp_config, monkeypatch): + config.set_active_profile("work") + monkeypatch.setenv(config.PROFILE_ENV_VAR, "personal") + assert config.get_active_profile() == "personal" + + def test_invalid_env_is_ignored(self, temp_config, monkeypatch): + config.set_active_profile("work") + monkeypatch.setenv(config.PROFILE_ENV_VAR, "../../etc") + assert config.get_active_profile() == "work" + + +class TestProfiles: + def test_save_under_named_profile(self, temp_config): + config.save_credentials({"token": "a"}, name="work") + config.save_credentials({"token": "b"}, name="home") + assert config.load_credentials("work") == {"token": "a"} + assert config.load_credentials("home") == {"token": "b"} + + def test_active_profile_routes_credentials(self, temp_config): + config.set_active_profile("work") + config.save_credentials({"token": "w"}) + assert config.load_credentials("work") == {"token": "w"} + assert config.load_credentials("home") is None + + def test_list_profiles_sorted(self, temp_config): + assert config.list_profiles() == [] + config.save_credentials({"token": "a"}, name="work") + config.save_credentials({"token": "b"}, name="home") + assert config.list_profiles() == ["home", "work"] + + def test_profile_exists(self, temp_config): + assert not config.profile_exists("work") + config.save_credentials({"token": "a"}, name="work") + assert config.profile_exists("work") + + def test_remove_profile(self, temp_config): + config.save_credentials({"token": "a"}, name="work") + config.remove_profile("work") + assert not config.profile_exists("work") + + def test_remove_active_profile_resets_active(self, temp_config): + config.save_credentials({"token": "a"}, name="work") + config.set_active_profile("work") + config.remove_profile("work") + assert config.get_active_profile() == config.DEFAULT_PROFILE + + +class TestProfileNames: + @pytest.mark.parametrize("name", ["work", "work-2", "home_tv", "a", "Channel1"]) + def test_valid(self, name): + assert config.is_valid_profile_name(name) + + @pytest.mark.parametrize( + "name", ["", "..", "a/b", "a b", ".hidden", "work.tv", "wo:rk", "work\n", "/abs"] + ) + def test_invalid(self, name): + assert not config.is_valid_profile_name(name) + + def test_profile_dir_rejects_traversal(self, temp_config): + with pytest.raises(ValueError): + config.profile_dir("../../etc") + + def test_profile_exists_false_for_invalid_name(self, temp_config): + assert config.profile_exists("../../etc") is False + + +class TestProfileMeta: + def test_save_and_load(self, temp_config): + config.save_profile_meta("work", {"title": "My Channel"}) + assert config.load_profile_meta("work")["title"] == "My Channel" + + def test_missing_returns_empty(self, temp_config): + assert config.load_profile_meta("work") == {} + + +class TestMigration: + def test_migrates_legacy_credentials(self, temp_config): + config.ensure_config_dir() + config.LEGACY_CREDENTIALS_FILE.write_text('{"token": "legacy"}') + + migrated = config.migrate_legacy_credentials() + + assert migrated is True + assert config.load_credentials("default") == {"token": "legacy"} + assert config.get_active_profile() == "default" + assert not config.LEGACY_CREDENTIALS_FILE.exists() + + def test_noop_without_legacy_file(self, temp_config): + assert config.migrate_legacy_credentials() is False + + def test_noop_when_profiles_already_exist(self, temp_config): + config.save_credentials({"token": "new"}, name="work") + config.ensure_config_dir() + config.LEGACY_CREDENTIALS_FILE.write_text('{"token": "legacy"}') + + assert config.migrate_legacy_credentials() is False + assert config.LEGACY_CREDENTIALS_FILE.exists() + + +class TestAtomicState: + def test_state_file_write_is_atomic(self, temp_config): + """A truncated/partial state.json must not slip past _save_state.""" + config.set_active_profile("work") + # No leftover temp file from the atomic rename. + assert not (temp_config / "state.json.tmp").exists() + assert json.loads(config.STATE_FILE.read_text())["active_profile"] == "work" + + +class TestListProfilesFiltering: + def test_list_profiles_skips_invalid_directory_names(self, temp_config): + config.save_credentials({"token": "a"}, name="work") + # Simulate a stray directory with an invalid profile name (e.g. left over + # from a manual edit). list_profiles must not surface it. + config.PROFILES_DIR.mkdir(parents=True, exist_ok=True) + (config.PROFILES_DIR / "not.a.profile").mkdir() + (config.PROFILES_DIR / ".hidden").mkdir() + + assert config.list_profiles() == ["work"] + + def test_migration_skips_when_only_invalid_dirs_exist(self, temp_config): + """Stray invalid directories must not block legacy migration.""" + config.ensure_config_dir() + config.PROFILES_DIR.mkdir(parents=True) + (config.PROFILES_DIR / "not.a.profile").mkdir() + config.LEGACY_CREDENTIALS_FILE.write_text('{"token": "legacy"}') + + migrated = config.migrate_legacy_credentials() + + assert migrated is True + assert config.load_credentials("default") == {"token": "legacy"} diff --git a/tests/test_profile.py b/tests/test_profile.py new file mode 100644 index 0000000..04e224e --- /dev/null +++ b/tests/test_profile.py @@ -0,0 +1,126 @@ +from unittest.mock import patch + +import pytest +from typer.testing import CliRunner + +from ytstudio import config +from ytstudio.main import app + +runner = CliRunner() + + +@pytest.fixture +def temp_config(tmp_path, monkeypatch): + config_dir = tmp_path / ".config" / "ytstudio-cli" + config_dir.mkdir(parents=True) + + monkeypatch.delenv(config.PROFILE_ENV_VAR, raising=False) + + with ( + patch.object(config, "CONFIG_DIR", config_dir), + patch.object(config, "CLIENT_SECRETS_FILE", config_dir / "client_secrets.json"), + patch.object(config, "PROFILES_DIR", config_dir / "profiles"), + patch.object(config, "STATE_FILE", config_dir / "state.json"), + patch.object(config, "LEGACY_CREDENTIALS_FILE", config_dir / "credentials.json"), + ): + yield config_dir + + +def _make_profile(name, title=None): + config.save_credentials({"token": "x"}, name=name) + config.save_profile_meta(name, {"title": title or f"{name} channel"}) + + +class TestProfileAdd: + def test_add_authenticates_and_activates(self, temp_config): + def fake_auth(headless=False, profile=None): + _make_profile(profile) + + with patch("ytstudio.commands.profile.authenticate", side_effect=fake_auth) as auth: + result = runner.invoke(app, ["profile", "add", "work"]) + + assert result.exit_code == 0 + auth.assert_called_once_with(headless=False, profile="work") + assert config.profile_exists("work") + assert config.get_active_profile() == "work" + + def test_add_rejects_invalid_name(self, temp_config): + result = runner.invoke(app, ["profile", "add", "bad/name"]) + assert result.exit_code == 1 + assert not config.profile_exists("bad/name") + + def test_add_rejects_duplicate(self, temp_config): + _make_profile("work") + result = runner.invoke(app, ["profile", "add", "work"]) + assert result.exit_code == 1 + + +class TestProfileList: + def test_list_empty(self, temp_config): + result = runner.invoke(app, ["profile", "list"]) + assert result.exit_code == 0 + assert "No profiles yet" in result.stdout + + def test_list_shows_profiles_and_active(self, temp_config): + _make_profile("work", "Work Channel") + _make_profile("home", "Home Channel") + config.set_active_profile("home") + + result = runner.invoke(app, ["profile", "list"]) + + assert result.exit_code == 0 + assert "Work Channel" in result.stdout + assert "Home Channel" in result.stdout + assert "home" in result.stdout + + +class TestProfileUse: + def test_use_switches_active(self, temp_config): + _make_profile("work") + result = runner.invoke(app, ["profile", "use", "work"]) + assert result.exit_code == 0 + assert config.get_active_profile() == "work" + + def test_use_missing_exits(self, temp_config): + result = runner.invoke(app, ["profile", "use", "ghost"]) + assert result.exit_code == 1 + + +class TestProfileRemove: + def test_remove_force(self, temp_config): + _make_profile("work") + result = runner.invoke(app, ["profile", "remove", "work", "--force"]) + assert result.exit_code == 0 + assert not config.profile_exists("work") + + def test_remove_missing_exits(self, temp_config): + result = runner.invoke(app, ["profile", "remove", "ghost"]) + assert result.exit_code == 1 + + def test_remove_aborts_without_confirmation(self, temp_config): + _make_profile("work") + runner.invoke(app, ["profile", "remove", "work"], input="n\n") + assert config.profile_exists("work") + + +class TestProfileStatus: + def test_status_delegates_to_target(self, temp_config): + _make_profile("work") + with patch("ytstudio.commands.profile.get_status") as get_status: + result = runner.invoke(app, ["profile", "status", "work"]) + + assert result.exit_code == 0 + get_status.assert_called_once_with("work") + + def test_status_defaults_to_active(self, temp_config): + _make_profile("work") + config.set_active_profile("work") + with patch("ytstudio.commands.profile.get_status") as get_status: + result = runner.invoke(app, ["profile", "status"]) + + assert result.exit_code == 0 + get_status.assert_called_once_with("work") + + def test_status_missing_profile_exits(self, temp_config): + result = runner.invoke(app, ["profile", "status", "ghost"]) + assert result.exit_code == 1 diff --git a/uv.lock b/uv.lock index 974c349..f4bd719 100644 --- a/uv.lock +++ b/uv.lock @@ -817,7 +817,7 @@ wheels = [ [[package]] name = "ytstudio-cli" -version = "0.1.0" +version = "0.1.1" source = { editable = "." } dependencies = [ { name = "google-api-python-client" },