From 6b11f11f4981e8cbfd8dce8543c2d4f1bb81e7b6 Mon Sep 17 00:00:00 2001 From: kicka5h Date: Tue, 10 Mar 2026 14:56:53 -0700 Subject: [PATCH 1/3] add github auth for private repos --- pyproject.toml | 2 +- snacks/auth.py | 115 ++++++++++++++++++++++++++++++ snacks/main.py | 44 +++++++++--- snacks/ops.py | 134 +++++++++++++++++++++++------------ tests/test_commands.py | 19 +++++ tests/test_stash_commands.py | 84 ++++++++++++++++++++++ 6 files changed, 342 insertions(+), 56 deletions(-) create mode 100644 snacks/auth.py diff --git a/pyproject.toml b/pyproject.toml index 42b29f4..cc6875b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "python-snacks" -version = "0.1.4" +version = "0.2.0" description = "A CLI tool for managing a personal stash of reusable Python code snippets." readme = "README.md" requires-python = ">=3.10" diff --git a/snacks/auth.py b/snacks/auth.py new file mode 100644 index 0000000..fe6730a --- /dev/null +++ b/snacks/auth.py @@ -0,0 +1,115 @@ +"""GitHub authentication for snack CLI. + +Resolution order: +1. GITHUB_TOKEN env var (useful for CI) +2. `gh auth token` (GitHub CLI, if installed and logged in) +3. GitHub OAuth device flow (prompts user in browser) +""" +from __future__ import annotations + +import json +import os +import subprocess +import time +import urllib.error +import urllib.parse +import urllib.request +from typing import Optional + +import typer + +# Register at https://github.com/settings/apps/new (Device Flow, no client secret needed) +_CLIENT_ID = os.environ.get("SNACK_GITHUB_CLIENT_ID", "") + + +def get_github_token() -> Optional[str]: + """Return a GitHub token, prompting via device flow if needed.""" + token = os.environ.get("GITHUB_TOKEN") + if token: + return token + + token = _token_from_gh_cli() + if token: + return token + + return _device_flow() + + +def _token_from_gh_cli() -> Optional[str]: + try: + result = subprocess.run( + ["gh", "auth", "token"], + capture_output=True, + text=True, + timeout=5, + ) + if result.returncode == 0: + return result.stdout.strip() + except (FileNotFoundError, subprocess.TimeoutExpired): + pass + return None + + +def _device_flow() -> Optional[str]: + if not _CLIENT_ID: + typer.echo( + "[error] No GitHub client ID configured. " + "Set SNACK_GITHUB_CLIENT_ID or GITHUB_TOKEN to authenticate.", + err=True, + ) + raise typer.Exit(1) + + # Step 1 — request device + user code + data = urllib.parse.urlencode({"client_id": _CLIENT_ID, "scope": "repo"}).encode() + req = urllib.request.Request( + "https://github.com/login/device/code", + data=data, + headers={"Accept": "application/json"}, + ) + try: + with urllib.request.urlopen(req) as resp: + payload = json.loads(resp.read()) + except urllib.error.URLError as e: + typer.echo(f"[error] Could not reach GitHub: {e.reason}", err=True) + raise typer.Exit(1) + + device_code = payload["device_code"] + user_code = payload["user_code"] + verification_uri = payload["verification_uri"] + interval = payload.get("interval", 5) + expires_in = payload.get("expires_in", 900) + + typer.echo(f"\nOpen: {verification_uri}") + typer.echo(f"Code: {user_code}\n") + + # Step 2 — poll until the user approves + deadline = time.time() + expires_in + while time.time() < deadline: + time.sleep(interval) + + data = urllib.parse.urlencode({ + "client_id": _CLIENT_ID, + "device_code": device_code, + "grant_type": "urn:ietf:params:oauth:grant-type:device_code", + }).encode() + req = urllib.request.Request( + "https://github.com/login/oauth/access_token", + data=data, + headers={"Accept": "application/json"}, + ) + with urllib.request.urlopen(req) as resp: + result = json.loads(resp.read()) + + if "access_token" in result: + typer.echo("Authenticated.") + return result["access_token"] + + error = result.get("error") + if error == "slow_down": + interval += 5 + elif error != "authorization_pending": + typer.echo(f"[error] Authentication failed: {error}", err=True) + raise typer.Exit(1) + + typer.echo("[error] Authentication timed out.", err=True) + raise typer.Exit(1) diff --git a/snacks/main.py b/snacks/main.py index 9392223..5858640 100644 --- a/snacks/main.py +++ b/snacks/main.py @@ -11,6 +11,7 @@ from snacks.config import SnackConfig, get_stash_path from snacks.ops import add_remote as do_add_remote from snacks.ops import pack as do_pack, unpack as do_unpack +from snacks.ops import read_index app = typer.Typer( name="snack", @@ -65,11 +66,7 @@ def list_snacks( ) -> None: """List all snippets in the stash.""" stash = get_stash_path() - snippets = sorted( - p.relative_to(stash).as_posix() - for p in stash.rglob("*.py") - if not any(part.startswith(("_", ".")) for part in p.relative_to(stash).parts) - ) + snippets = sorted(read_index(stash)) if category: snippets = [s for s in snippets if s.startswith(f"{category}/")] typer.echo("\n".join(snippets) if snippets else "No snippets found.") @@ -82,10 +79,8 @@ def search( """Search snippet filenames for a keyword.""" stash = get_stash_path() matches = sorted( - p.relative_to(stash).as_posix() - for p in stash.rglob("*.py") - if keyword.lower() in p.name.lower() - and not any(part.startswith(("_", ".")) for part in p.relative_to(stash).parts) + s for s in read_index(stash) + if keyword.lower() in Path(s).name.lower() ) typer.echo("\n".join(matches) if matches else f"No snippets matching '{keyword}'.") @@ -179,6 +174,37 @@ def stash_move( cfg.save() +@stash_app.command("delete") +def stash_delete( + name: str = typer.Argument(..., help="Name of the stash to remove from config."), +) -> None: + """Remove a stash from config (does not delete files on disk).""" + cfg = SnackConfig() + if not cfg.has_stash(name): + typer.echo( + f"[error] No stash named '{name}'. Run 'snack stash list' to see available stashes.", + err=True, + ) + raise typer.Exit(1) + + was_active = cfg.active_name() == name + cfg.remove_stash(name) + + if was_active: + remaining = cfg.stashes() + if remaining: + next_name = next(iter(sorted(remaining))) + cfg.set_active(next_name) + cfg.save() + typer.echo(f"Deleted stash '{name}'. Active stash → '{next_name}'.") + else: + cfg.save() + typer.echo(f"Deleted stash '{name}'. No stashes remaining.") + else: + cfg.save() + typer.echo(f"Deleted stash '{name}'.") + + @stash_app.command("add-remote") def stash_add_remote( repo: str = typer.Argument(..., help="GitHub repo as 'owner/repo' or a full GitHub URL."), diff --git a/snacks/ops.py b/snacks/ops.py index a09b12d..1fa32e4 100644 --- a/snacks/ops.py +++ b/snacks/ops.py @@ -8,6 +8,25 @@ import typer +_MANIFEST = ".snack_index" + + +def read_index(stash: Path) -> list[str]: + """Return all tracked snack paths (relative to stash root).""" + index = stash / _MANIFEST + if not index.exists(): + return [] + return [l.strip() for l in index.read_text().splitlines() if l.strip()] + + +def _track(stash: Path, rel_path: str) -> None: + """Add a path to the stash manifest if not already present.""" + index = stash / _MANIFEST + existing = set(read_index(stash)) + if rel_path not in existing: + with open(index, "a") as f: + f.write(rel_path + "\n") + def unpack(stash: Path, snippet_path: str, flat: bool, force: bool) -> None: """Copy a file from the stash into the current working directory.""" @@ -30,69 +49,92 @@ def pack(stash: Path, snippet_path: str, force: bool) -> None: dest = stash / snippet_path _copy(src, dest, force) + _track(stash, snippet_path) typer.echo(f"Packed {snippet_path} → {dest}") def add_remote(stash: Path, repo: str, subdir: Optional[str], force: bool) -> None: """Download .py files from a GitHub repo into the stash.""" + from snacks.auth import get_github_token + owner, repo_name = _parse_github_repo(repo) url = f"https://api.github.com/repos/{owner}/{repo_name}/tarball" + headers = {"User-Agent": "python-snacks", "Accept": "application/vnd.github+json"} typer.echo(f"Fetching {owner}/{repo_name}...") - req = urllib.request.Request( - url, - headers={"User-Agent": "snack-stash", "Accept": "application/vnd.github+json"}, - ) - try: - with tempfile.TemporaryDirectory() as tmpdir: - tmp = Path(tmpdir) - tarball = tmp / "repo.tar.gz" - - with urllib.request.urlopen(req) as response: - tarball.write_bytes(response.read()) - - extract_dir = tmp / "repo" - extract_dir.mkdir() - with tarfile.open(tarball) as tf: - try: - tf.extractall(extract_dir, filter="data") - except TypeError: - tf.extractall(extract_dir) # Python < 3.12 - - roots = list(extract_dir.iterdir()) - if not roots: - typer.echo("[error] Downloaded archive was empty.", err=True) - raise typer.Exit(1) - repo_root = roots[0] - - py_files = sorted( - p for p in repo_root.rglob("*.py") - if subdir is None or _is_under_subdir(p.relative_to(repo_root), subdir) - ) - - if not py_files: - msg = "No Python files found" - if subdir: - msg += f" under '{subdir}'" - typer.echo(msg + ".") - return - - for src in py_files: - rel = src.relative_to(repo_root) - _copy(src, stash / rel, force) - typer.echo(f" + {rel.as_posix()}") - - typer.echo(f"\nAdded {len(py_files)} file(s) from {owner}/{repo_name}.") + def _make_request() -> urllib.request.Request: + return urllib.request.Request(url, headers=headers) + try: + _download_and_install(stash, owner, repo_name, subdir, force, _make_request()) except urllib.error.HTTPError as e: - typer.echo(f"[error] HTTP {e.code}: {e.reason}", err=True) - raise typer.Exit(1) + if e.code != 404: + typer.echo(f"[error] HTTP {e.code}: {e.reason}", err=True) + raise typer.Exit(1) + # 404 may mean private repo — authenticate and retry once + token = get_github_token() + headers["Authorization"] = f"Bearer {token}" + try: + _download_and_install(stash, owner, repo_name, subdir, force, _make_request()) + except urllib.error.HTTPError as e2: + typer.echo(f"[error] HTTP {e2.code}: {e2.reason}", err=True) + raise typer.Exit(1) except urllib.error.URLError as e: typer.echo(f"[error] Network error: {e.reason}", err=True) raise typer.Exit(1) +def _download_and_install( + stash: Path, + owner: str, + repo_name: str, + subdir: Optional[str], + force: bool, + req: urllib.request.Request, +) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + tmp = Path(tmpdir) + tarball = tmp / "repo.tar.gz" + + with urllib.request.urlopen(req) as response: + tarball.write_bytes(response.read()) + + extract_dir = tmp / "repo" + extract_dir.mkdir() + with tarfile.open(tarball) as tf: + try: + tf.extractall(extract_dir, filter="data") + except TypeError: + tf.extractall(extract_dir) # Python < 3.12 + + roots = list(extract_dir.iterdir()) + if not roots: + typer.echo("[error] Downloaded archive was empty.", err=True) + raise typer.Exit(1) + repo_root = roots[0] + + py_files = sorted( + p for p in repo_root.rglob("*.py") + if subdir is None or _is_under_subdir(p.relative_to(repo_root), subdir) + ) + + if not py_files: + msg = "No Python files found" + if subdir: + msg += f" under '{subdir}'" + typer.echo(msg + ".") + return + + for src in py_files: + rel = src.relative_to(repo_root) + _copy(src, stash / rel, force) + _track(stash, rel.as_posix()) + typer.echo(f" + {rel.as_posix()}") + + typer.echo(f"\nAdded {len(py_files)} file(s) from {owner}/{repo_name}.") + + def _parse_github_repo(repo: str) -> tuple[str, str]: repo = repo.strip().rstrip("/") for prefix in ("https://github.com/", "http://github.com/", "github.com/"): diff --git a/tests/test_commands.py b/tests/test_commands.py index 72445cc..1bf3015 100644 --- a/tests/test_commands.py +++ b/tests/test_commands.py @@ -19,6 +19,9 @@ def stash(tmp_path_factory): (d / "auth" / "jwt_helpers.py").write_text("# jwt\n") (d / "forms").mkdir() (d / "forms" / "contact_form.py").write_text("# contact\n") + (d / ".snack_index").write_text( + "auth/google_oauth.py\nauth/jwt_helpers.py\nforms/contact_form.py\n" + ) return d @@ -76,6 +79,22 @@ def test_list_empty_category(stash_env): assert "No snippets found" in result.output +def test_list_excludes_non_snack_files(stash, stash_env): + """Files present in the stash directory but not tracked in .snack_index are excluded.""" + # Drop arbitrary .py files directly into the stash (not via pack) + (stash / "setup.py").write_text("# setup\n") + (stash / "some_project").mkdir() + (stash / "some_project" / "main.py").write_text("# project file\n") + + result = runner.invoke(app, ["list"], env=stash_env) + assert result.exit_code == 0 + assert "setup.py" not in result.output + assert "some_project/main.py" not in result.output + # Tracked snippets still show up + assert "auth/google_oauth.py" in result.output + assert "forms/contact_form.py" in result.output + + # --------------------------------------------------------------------------- # search # --------------------------------------------------------------------------- diff --git a/tests/test_stash_commands.py b/tests/test_stash_commands.py index 0a9cbfc..e7af9a4 100644 --- a/tests/test_stash_commands.py +++ b/tests/test_stash_commands.py @@ -140,6 +140,49 @@ def test_stash_move_target_exists_errors(cfg_file, tmp_path): assert "already exists" in result.output +# ── stash delete ───────────────────────────────────────────────────────────── + +def test_stash_delete_removes_from_config(cfg_file, tmp_path): + stash_dir = tmp_path / "my-stash" + runner.invoke(app, ["stash", "create", "default", str(stash_dir)], env={}) + result = runner.invoke(app, ["stash", "delete", "default"], env={}) + assert result.exit_code == 0, result.output + assert "Deleted stash 'default'" in result.output + assert "stash.default" not in cfg_file.read_text() + + +def test_stash_delete_does_not_remove_files(cfg_file, tmp_path): + stash_dir = tmp_path / "my-stash" + runner.invoke(app, ["stash", "create", "default", str(stash_dir)], env={}) + runner.invoke(app, ["stash", "delete", "default"], env={}) + assert stash_dir.exists() + + +def test_stash_delete_unknown_name_errors(cfg_file): + result = runner.invoke(app, ["stash", "delete", "nonexistent"], env={}) + assert result.exit_code != 0 + assert "No stash named" in result.output + + +def test_stash_delete_active_promotes_next(cfg_file, tmp_path): + a = tmp_path / "stash-a" + b = tmp_path / "stash-b" + runner.invoke(app, ["stash", "create", "alpha", str(a)], env={}) + runner.invoke(app, ["stash", "create", "beta", str(b), "--no-activate"], env={}) + result = runner.invoke(app, ["stash", "delete", "alpha"], env={}) + assert result.exit_code == 0 + assert "beta" in result.output + assert "active = beta" in cfg_file.read_text() + + +def test_stash_delete_last_stash_leaves_no_active(cfg_file, tmp_path): + stash_dir = tmp_path / "my-stash" + runner.invoke(app, ["stash", "create", "default", str(stash_dir)], env={}) + result = runner.invoke(app, ["stash", "delete", "default"], env={}) + assert result.exit_code == 0 + assert "No stashes remaining" in result.output + + # ── stash add-remote ───────────────────────────────────────────────────────── def _make_tarball(files: dict[str, str]) -> bytes: @@ -240,6 +283,47 @@ def test_add_remote_http_error(remote_stash_env): assert "404" in result.output +def test_add_remote_public_repo_needs_no_auth(remote_stash, remote_stash_env): + tarball = _make_tarball({"auth/oauth.py": "# oauth\n"}) + mock_response = _mock_urlopen(tarball) + with patch("urllib.request.urlopen", return_value=mock_response) as mock_open: + result = runner.invoke(app, ["stash", "add-remote", "owner/repo"], env=remote_stash_env) + assert result.exit_code == 0 + req = mock_open.call_args[0][0] + assert req.get_header("Authorization") is None + + +def test_add_remote_private_repo_retries_with_gh_token(remote_stash, remote_stash_env): + """On 404, auth via gh CLI and retry.""" + import urllib.error + tarball = _make_tarball({"auth/oauth.py": "# oauth\n"}) + not_found = urllib.error.HTTPError(url="", code=404, msg="Not Found", hdrs=None, fp=None) + auth_response = _mock_urlopen(tarball) + + with patch("urllib.request.urlopen", side_effect=[not_found, auth_response]), \ + patch("snacks.auth._token_from_gh_cli", return_value="ghp_cli_token"): + result = runner.invoke(app, ["stash", "add-remote", "owner/private-repo"], env=remote_stash_env) + + assert result.exit_code == 0 + assert (remote_stash / "auth" / "oauth.py").exists() + + +def test_add_remote_private_repo_uses_github_token_env(remote_stash, remote_stash_env): + """GITHUB_TOKEN env var is used without prompting.""" + import urllib.error + tarball = _make_tarball({"auth/oauth.py": "# oauth\n"}) + not_found = urllib.error.HTTPError(url="", code=404, msg="Not Found", hdrs=None, fp=None) + auth_response = _mock_urlopen(tarball) + + env = {**remote_stash_env, "GITHUB_TOKEN": "ghp_envtoken"} + with patch("urllib.request.urlopen", side_effect=[not_found, auth_response]) as mock_open: + result = runner.invoke(app, ["stash", "add-remote", "owner/private-repo"], env=env) + + assert result.exit_code == 0 + final_req = mock_open.call_args[0][0] + assert final_req.get_header("Authorization") == "Bearer ghp_envtoken" + + def test_add_remote_no_py_files(remote_stash, remote_stash_env): tarball = _make_tarball({"README.md": "# readme"}) mock_response = _mock_urlopen(tarball) From c0f0cb6847304cd8e2069adf2c07b30dbef5ea92 Mon Sep 17 00:00:00 2001 From: kicka5h Date: Tue, 10 Mar 2026 15:07:06 -0700 Subject: [PATCH 2/3] passthrough secret to py test --- .github/workflows/ci.yml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4d9db20..bb380e9 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -28,3 +28,5 @@ jobs: - name: Run tests run: pytest tests/ -v + env: + SNACK_GITHUB_CLIENT_ID: ${{ secrets.SNACK_GITHUB_CLIENT_ID }} From f83fe438ee5cabeb04885e8f4084684758129997 Mon Sep 17 00:00:00 2001 From: kicka5h Date: Tue, 10 Mar 2026 15:12:13 -0700 Subject: [PATCH 3/3] mark as local only --- tests/test_stash_commands.py | 20 +++++++++++++++++++- 1 file changed, 19 insertions(+), 1 deletion(-) diff --git a/tests/test_stash_commands.py b/tests/test_stash_commands.py index e7af9a4..db410c1 100644 --- a/tests/test_stash_commands.py +++ b/tests/test_stash_commands.py @@ -1,5 +1,6 @@ """Tests for `snack stash *` commands.""" import io +import os import tarfile import tempfile from pathlib import Path @@ -8,6 +9,8 @@ import pytest from typer.testing import CliRunner +local_only = pytest.mark.skipif(os.environ.get("CI") == "true", reason="local only") + import snacks.config as config_module from snacks.main import app @@ -274,8 +277,21 @@ def test_add_remote_invalid_repo_errors(remote_stash_env): def test_add_remote_http_error(remote_stash_env): import urllib.error with patch("urllib.request.urlopen", side_effect=urllib.error.HTTPError( - url="", code=404, msg="Not Found", hdrs=None, fp=None + url="", code=500, msg="Internal Server Error", hdrs=None, fp=None )): + result = runner.invoke( + app, ["stash", "add-remote", "owner/repo"], env=remote_stash_env + ) + assert result.exit_code != 0 + assert "500" in result.output + + +def test_add_remote_not_found_after_auth(remote_stash_env): + """Repo not found even after authentication — should report 404.""" + import urllib.error + not_found = urllib.error.HTTPError(url="", code=404, msg="Not Found", hdrs=None, fp=None) + with patch("urllib.request.urlopen", side_effect=not_found), \ + patch("snacks.auth._token_from_gh_cli", return_value="ghp_token"): result = runner.invoke( app, ["stash", "add-remote", "owner/missing-repo"], env=remote_stash_env ) @@ -293,6 +309,7 @@ def test_add_remote_public_repo_needs_no_auth(remote_stash, remote_stash_env): assert req.get_header("Authorization") is None +@local_only def test_add_remote_private_repo_retries_with_gh_token(remote_stash, remote_stash_env): """On 404, auth via gh CLI and retry.""" import urllib.error @@ -308,6 +325,7 @@ def test_add_remote_private_repo_retries_with_gh_token(remote_stash, remote_stas assert (remote_stash / "auth" / "oauth.py").exists() +@local_only def test_add_remote_private_repo_uses_github_token_env(remote_stash, remote_stash_env): """GITHUB_TOKEN env var is used without prompting.""" import urllib.error