Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -58,11 +58,12 @@ exclude_lines = [
"pragma: no cover",
"if TYPE_CHECKING:",
"raise NotImplementedError",
"if __name__ == .__main__.:",
]

[project.scripts]
ytstudio = "ytstudio.main:app"
yts = "ytstudio.main:app"
ytstudio = "ytstudio.main:cli"
yts = "ytstudio.main:cli"

[project.urls]
Homepage = "https://github.com/jdwit/ytstudio-cli"
Expand Down
29 changes: 19 additions & 10 deletions src/ytstudio/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from oauthlib.oauth2 import AccessDeniedError, OAuth2Error
from rich.prompt import Prompt

from ytstudio.config import (
Expand Down Expand Up @@ -37,7 +38,7 @@ def handle_api_error(error: HttpError) -> None:
console.print("[red]Access denied. You may not have permission for this action.[/red]")
raise SystemExit(1) from None

# Re-raise for other errors
# Re-raise: command handlers or the CLI boundary (main.cli) clean it up.
raise error


Expand Down Expand Up @@ -168,6 +169,22 @@ def _authenticate_headless() -> Credentials:
return flow.credentials


def _authenticate_local_server() -> Credentials:
flow = _create_flow()
try:
return flow.run_local_server(port=9876, prompt="consent", open_browser=True)
except AccessDeniedError:
console.print(
"[red]Authorization denied.[/red] "
"Approve the requested (read-only) scopes and try again."
)
raise SystemExit(1) from None
except OAuth2Error as error:
message = getattr(error, "description", "") or str(error)
console.print(f"[red]Authorization failed: {message}[/red]")
raise SystemExit(1) from 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]")
Expand All @@ -179,15 +196,7 @@ def authenticate(headless: bool = False, profile: str | None = None) -> None:

console.print("[bold]Authenticating with YouTube...[/bold]\n")

if headless:
credentials = _authenticate_headless()
else:
flow = _create_flow()
credentials = flow.run_local_server(
port=9876,
prompt="consent",
open_browser=True,
)
credentials = _authenticate_headless() if headless else _authenticate_local_server()

_save_credentials(credentials, profile)
_show_login_success(credentials, profile)
Expand Down
19 changes: 18 additions & 1 deletion src/ytstudio/main.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
import atexit

import typer
from googleapiclient.errors import HttpError
from oauthlib.oauth2 import OAuth2Error
from rich.console import Console

from ytstudio.api import authenticate, get_status
Expand Down Expand Up @@ -87,5 +89,20 @@ def main(
_update_state["registered"] = True


def cli():
# Error boundary: turn expected API/OAuth failures into one clean line.
# Unexpected exceptions keep their traceback.
try:
app()
except HttpError as error:
message = error.reason or f"YouTube API request failed (HTTP {error.resp.status})."
console.print(f"[red]{message}[/red]")
raise SystemExit(1) from None
except OAuth2Error as error:
message = getattr(error, "description", "") or str(error)
console.print(f"[red]Authorization failed: {message}[/red]")
raise SystemExit(1) from None


if __name__ == "__main__":
app()
cli()
53 changes: 52 additions & 1 deletion tests/test_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,12 +3,13 @@
import pytest
from google.auth.exceptions import RefreshError
from googleapiclient.errors import HttpError
from oauthlib.oauth2 import AccessDeniedError, OAuth2Error
from typer import Exit
from typer.testing import CliRunner

from ytstudio import api as api_module
from ytstudio.api import api, get_authenticated_service, handle_api_error
from ytstudio.main import app
from ytstudio.main import app, cli

runner = CliRunner()

Expand Down Expand Up @@ -42,6 +43,7 @@ def test_forbidden_exits(self):
handle_api_error(error)

def test_other_errors_reraise(self):
# Unknown errors re-raise so command handlers / the CLI boundary handle them.
error = make_http_error(404)

with pytest.raises(HttpError):
Expand Down Expand Up @@ -178,6 +180,33 @@ def test_status_not_authenticated(self):
assert "Not authenticated" in result.stdout


class TestAuthenticateLocalServer:
def test_access_denied_exits_cleanly(self, capsys):
flow = MagicMock()
flow.run_local_server.side_effect = AccessDeniedError()

with (
patch("ytstudio.api._create_flow", return_value=flow),
pytest.raises(SystemExit),
):
api_module._authenticate_local_server()

assert "Authorization denied" in capsys.readouterr().out

def test_other_oauth_error_exits_cleanly(self, capsys):
flow = MagicMock()
flow.run_local_server.side_effect = OAuth2Error(description="boom")

with (
patch("ytstudio.api._create_flow", return_value=flow),
pytest.raises(SystemExit),
):
api_module._authenticate_local_server()

out = capsys.readouterr().out
assert "Authorization failed" in out and "boom" in out


class TestHelpers:
def test_create_flow_uses_client_secrets_and_scopes(self):
with patch("ytstudio.api.InstalledAppFlow.from_client_secrets_file") as factory:
Expand Down Expand Up @@ -251,6 +280,28 @@ def test_logout_clears_credentials(self):
clear_credentials.assert_called_once()


class TestCliBoundary:
def test_unhandled_http_error_prints_clean_message(self, capsys):
resp = MagicMock()
resp.status = 400
error = HttpError(resp, b'{"error": {"message": "Date range does not align."}}')
with patch("ytstudio.main.app", side_effect=error), pytest.raises(SystemExit):
cli()

out = capsys.readouterr().out
assert "Date range does not align." in out
assert "Traceback" not in out

def test_unhandled_oauth_error_prints_clean_message(self, capsys):
with (
patch("ytstudio.main.app", side_effect=OAuth2Error(description="boom")),
pytest.raises(SystemExit),
):
cli()

assert "Authorization failed" in capsys.readouterr().out


class TestAuthenticate:
def test_normal_login_uses_local_server(self):
credentials = MagicMock()
Expand Down
Loading