integration_github_token_resilience.py

integration_github_token_resilience.py

#!/usr/local/autopkg/python
"""
Integration test for GitHub token resilience.

Exercises four new behaviors in autopkglib.github:
  1. Malformed token (interior whitespace) is rejected, unauthenticated fallback
  2. Token without a known prefix is accepted without warnings
  3. Unauthenticated session can make a real API call successfully
  4. Bad token triggers 401 -> anonymous retry -> result returned, warning shown

Paths 3 and 4 make real network requests to api.github.com.
"""

import os
import sys
import tempfile
from io import StringIO
from unittest.mock import patch

TESTS_DIR = os.path.dirname(os.path.abspath(__file__))
CODE_DIR = os.path.dirname(TESTS_DIR)
sys.path.insert(0, CODE_DIR)

from autopkglib.github import GitHubSession, get_github_token  # noqa: E402

TEST_ENDPOINT = "/repos/autopkg/autopkg"
BAD_TOKEN = "ghp_thisisnotarealtoken000000000000"

passed = 0
failed = 0


def check(label, condition, detail=None):
    global passed, failed
    if condition:
        passed += 1
        print(f"  [PASS] {label}")
    else:
        failed += 1
        print(f"  [FAIL] {label}")
    if detail:
        print(f"         {detail}")


def make_session(token=None):
    with patch.object(GitHubSession, "_get_token", return_value=token):
        return GitHubSession()


# ---------------------------------------------------------------------------
print("\nPath 1: Malformed token (interior whitespace) is rejected")
# ---------------------------------------------------------------------------

with tempfile.TemporaryDirectory() as tmpdir:
    token_path = os.path.join(tmpdir, ".autopkg_gh_token")
    with open(token_path, "w") as f:
        f.write("ghp_bad token\n")

    buf = StringIO()
    with (
        patch("autopkglib.github.get_pref", return_value=None),
        patch("sys.stderr", new=buf),
    ):
        token = get_github_token(token_path)

    stderr_out = buf.getvalue().strip()
    check("Returns None", token is None)
    check("Logs 'Ignoring malformed'", "Ignoring malformed" in stderr_out,
          f"stderr: {stderr_out}")
    check("Names the source file", token_path in stderr_out,
          f"stderr: {stderr_out}")


# ---------------------------------------------------------------------------
print("\nPath 2: Token without known prefix is accepted without warnings")
# ---------------------------------------------------------------------------

with tempfile.TemporaryDirectory() as tmpdir:
    token_path = os.path.join(tmpdir, ".autopkg_gh_token")
    with open(token_path, "w") as f:
        f.write("custom_token_abc123\n")

    buf = StringIO()
    with (
        patch("autopkglib.github.get_pref", return_value=None),
        patch("sys.stderr", new=buf),
    ):
        token = get_github_token(token_path)

    stderr_out = buf.getvalue().strip()
    check("Token is accepted", token == "custom_token_abc123")
    check("No warnings logged", not stderr_out,
          f"stderr: {stderr_out}" if stderr_out else None)


# ---------------------------------------------------------------------------
print("\nPath 3: Unauthenticated session makes a successful API call")
print("  (1 real network request to api.github.com)")
# ---------------------------------------------------------------------------

session = make_session(token=None)
resp, status = session.call_api(TEST_ENDPOINT)

check("HTTP 200 returned", status == 200, f"status: {status}")
check("Response contains expected repo name",
      isinstance(resp, dict) and resp.get("name") == "autopkg",
      f"name field: {resp.get('name') if isinstance(resp, dict) else type(resp)}")
check("No token was sent (session.token is None)", session.token is None)


# ---------------------------------------------------------------------------
print("\nPath 4: Bad token triggers 401, retries anonymously, succeeds")
print("  (2 real network requests to api.github.com)")
# ---------------------------------------------------------------------------

session = make_session(token=BAD_TOKEN)
buf = StringIO()
with patch("sys.stderr", new=buf):
    resp, status = session.call_api(TEST_ENDPOINT)

stderr_out = buf.getvalue().strip()
print(f"  stderr: {stderr_out}")
check("HTTP 200 returned after anonymous retry", status == 200, f"status: {status}")
check("Response contains expected repo name",
      isinstance(resp, dict) and resp.get("name") == "autopkg",
      f"name field: {resp.get('name') if isinstance(resp, dict) else type(resp)}")
check("Warning: token invalid or expired", "invalid or expired" in stderr_out)
check("Warning: continuing without it", "Continuing without it" in stderr_out)


# ---------------------------------------------------------------------------
total = passed + failed
print(f"\n{passed} of {total} checks passed")
if failed:
    sys.exit(1)