diff --git a/AGENTS.md b/AGENTS.md index a4f063e..d34085b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -4,7 +4,7 @@ Guidance for AI coding assistants working in this repository. ## TL;DR -- `src/onepin/_cli/` and `tests/` are **hand-rolled** — edit freely. +- `src/onepin/_cli/`, `src/onepin/_version_gate.py`, and `tests/` are **hand-rolled** — edit freely. - Everything else under `src/onepin/` is **generated by Fern** from the OpenAPI spec — do not edit; changes will be lost on the next regen. - Lint and tests must be scoped: never run `ruff check .` or `pytest src/onepin/`. @@ -13,10 +13,11 @@ Guidance for AI coding assistants working in this repository. | Path | Status | Editable? | |------|--------|-----------| | `src/onepin/_cli/` | Hand-rolled Typer CLI | YES | -| `src/onepin/` (excluding `_cli/`) | Fern-generated SDK | NO — overwritten on every regen | +| `src/onepin/` (excluding `_cli/` + `_version_gate.py`) | Fern-generated SDK | NO — overwritten on every regen | +| `src/onepin/_version_gate.py` | Hand-rolled SDK version gate (preserved via `.fernignore`; re-exported via `generators.yml`) | YES | | `scripts/post_fern.sh` | Restores `py.typed` markers after regen | YES | | `fern/` | Fern config (`fern.config.json`, `generators.yml`) for in-repo self-generation | YES | -| `src/onepin/.fernignore` | Paths Fern preserves on regen (`_cli/`, `py.typed`) | YES | +| `src/onepin/.fernignore` | Paths Fern preserves on regen (`_cli/`, `_version_gate.py`, `py.typed`) | YES | | `.github/workflows/regen.yml` | Self-generates the SDK from the OnePin API spec | YES | | `tests/` | Project tests (`unit/`, `cli/`, `build/`) | YES | | `src/onepin/tests/` | SDK-bundled tests | NOT COLLECTED — `testpaths = ["tests"]` | @@ -29,10 +30,16 @@ Files under `src/onepin/` (excluding `_cli/`) are generated by [Fern](https://bu **If you need to change generated behavior, change the upstream OpenAPI spec.** Generation runs *in this repo*: `.github/workflows/regen.yml` fetches the shaped spec and runs `fern generate --local` (config in `fern/`), then opens a regen PR. -`src/onepin/.fernignore` lists the hand-rolled paths Fern preserves across regeneration (`_cli/`, `py.typed`). After every regen, `scripts/post_fern.sh` restores the PEP 561 markers (CI runs it automatically; local regens require running it by hand). `tests/build/test_cli_preserved.py` guards both. +`src/onepin/.fernignore` lists the hand-rolled paths Fern preserves across regeneration (`_cli/`, `_version_gate.py`, `py.typed`). After every regen, `scripts/post_fern.sh` restores the PEP 561 markers (CI runs it automatically; local regens require running it by hand). `tests/build/test_cli_preserved.py` guards both. If a patch on a generated file becomes unavoidable, add it to `src/onepin/.fernignore` with a comment explaining why, and document the patch in this file under a new "Active patches" section. +## Active patches + +Patches applied to generated files (kept minimal; each must be reproducible on regen): + +- **`src/onepin/__init__.py`** — re-exports `make_client`, `make_async_client`, and `OnePinUpgradeRequiredError` from the hand-rolled `_version_gate` (lazy `_dynamic_imports` + `__all__` entries). Reproduced automatically by the `additional_init_exports` block in `fern/generators.yml`, so the hand edit is only a bridge until the next `fern generate`. + ## CLI subpackage (`src/onepin/_cli/`) | Module | Role | diff --git a/README.md b/README.md index 7bfcc60..2fc3477 100644 --- a/README.md +++ b/README.md @@ -47,6 +47,29 @@ your request is about OnePin. After the first install, restart your tool (or run in Claude Code) so it picks up the new skills directory. The skill drives the same `onepin` CLI, so run `onepin login` first. +## Version compatibility + +The OnePin API advertises the minimum SDK version it still accepts. When a response indicates your +installed `onepin` is **below that floor**, the SDK stops with a clear, copy-paste upgrade message: + +``` +onepin 0.4.1 is below the required minimum 0.5.0. Upgrade: pip install --upgrade 'onepin>=0.5.0' +``` + +The `onepin` CLI enforces this automatically. For **programmatic** use, build the client with +`onepin.make_client` (instead of `OnePinClient` directly) to get the same gate plus a corrected +`User-Agent`: + +```python +import onepin + +client = onepin.make_client(token="op_live_...") +client.workflows.list() # raises onepin.OnePinUpgradeRequiredError if the SDK is too old +``` + +The CLI also nudges you when a newer release is available on PyPI (surfaced through the OnePin agent +skill). Set `ONEPIN_NO_UPDATE_CHECK=1` to silence the recommended-upgrade check. + ## Command reference The CLI groups its commands by resource. Every group prints its own command list with diff --git a/fern/generators.yml b/fern/generators.yml index bd45dc1..88d8c0d 100644 --- a/fern/generators.yml +++ b/fern/generators.yml @@ -26,3 +26,12 @@ groups: client_class_name: OnePinClient package_name: onepin flat_layout: false + # Re-export the hand-rolled version gate (src/onepin/_version_gate.py, preserved by + # .fernignore) from the generated package __init__ on every regen, so programmatic + # users get `from onepin import make_client` (a version-gated OnePinClient). + additional_init_exports: + - from: _version_gate + imports: + - make_client + - make_async_client + - OnePinUpgradeRequiredError diff --git a/pyproject.toml b/pyproject.toml index 4045547..ebd68b9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -18,6 +18,7 @@ dependencies = [ "typer>=0.12,<1.0", "rich>=13", "pydantic>=2.7,<3.0", + "packaging>=23", "tomli; python_version<\"3.11\"", ] diff --git a/src/onepin/.fernignore b/src/onepin/.fernignore index 612a276..5b45c2a 100644 --- a/src/onepin/.fernignore +++ b/src/onepin/.fernignore @@ -2,4 +2,5 @@ # Fern auto-discovers this file in the generator's local-file-system output dir # (src/onepin/) and never overwrites the paths listed here. _cli/ +_version_gate.py py.typed diff --git a/src/onepin/__init__.py b/src/onepin/__init__.py index 4d6a066..3923001 100644 --- a/src/onepin/__init__.py +++ b/src/onepin/__init__.py @@ -517,6 +517,11 @@ "workspace_aggregates": ".workspace_aggregates", "workspace_members": ".workspace_members", "workspaces": ".workspaces", + # Hand-rolled version gate, re-exported here. Also declared in + # fern/generators.yml `additional_init_exports`, so `fern generate` reproduces these. + "OnePinUpgradeRequiredError": "._version_gate", + "make_async_client": "._version_gate", + "make_client": "._version_gate", } @@ -795,4 +800,7 @@ def __dir__(): "workspace_aggregates", "workspace_members", "workspaces", + "OnePinUpgradeRequiredError", + "make_async_client", + "make_client", ] diff --git a/src/onepin/_cli/_ctx.py b/src/onepin/_cli/_ctx.py index a1a6963..30a04a8 100644 --- a/src/onepin/_cli/_ctx.py +++ b/src/onepin/_cli/_ctx.py @@ -59,7 +59,9 @@ def build_client(creds: ResolvedCredentials) -> OnePinClient: (guards against sending the token to a ``file://``/``ftp://`` host). """ # Deferred import: importing _ctx must stay cheap (no SDK on `onepin --help`). - from onepin.client import OnePinClient + # make_client wraps OnePinClient with the version-gate response hook + a corrected + # User-Agent (the generated default is baked at codegen time and can be stale). + from onepin._version_gate import make_client if not creds.api_key: raise OnePinAuthError( @@ -71,7 +73,7 @@ def build_client(creds: ResolvedCredentials) -> OnePinClient: f"Invalid base URL {creds.base_url!r}: only http and https are supported.", error_code="INVALID_BASE_URL", ) - return OnePinClient(base_url=creds.base_url, token=creds.api_key) + return make_client(base_url=creds.base_url, token=creds.api_key) def _is_commandline_source(value: object) -> bool: @@ -164,6 +166,7 @@ def api_errors(json_out: bool) -> Iterator[None]: json_out: Whether to emit the structured-JSON error envelope. """ # Deferred: these SDK error classes live under the generated tree. + from onepin._version_gate import OnePinUpgradeRequiredError from onepin.core.api_error import ApiError from onepin.core.parse_error import ParsingError from onepin.errors import ConflictError, NotFoundError, UnprocessableEntityError @@ -192,7 +195,16 @@ def api_errors(json_out: bool) -> Iterator[None]: json_out, ) raise SystemExit(1) from exc + except OnePinUpgradeRequiredError as exc: + # Client-side gate (response hook) saw an install below the required floor. + _emit_upgrade_required(str(exc), json_out) + raise SystemExit(1) from exc except (NotFoundError, ConflictError, UnprocessableEntityError, ApiError) as exc: + # 426 Upgrade Required: the server hard-floors the SDK version. Surface it as the + # same yellow upgrade message instead of a generic API error. + if getattr(exc, "status_code", None) == 426: + _emit_upgrade_required(_format_upgrade_required(exc), json_out) + raise SystemExit(1) from exc code, message, request_id = _classify_api_error(exc) _emit_error(code, message, request_id, json_out) raise SystemExit(1) from exc @@ -295,3 +307,27 @@ def _emit_error(code: str, message: str, request_id: str | None, json_out: bool) else: rid = f" (request_id={request_id})" if request_id else "" print(f"[{code}] {message}{rid}", file=sys.stderr) + + +def _emit_upgrade_required(message: str, json_out: bool) -> None: + """Emit a version-floor failure: machine envelope under --json, else a yellow stderr note.""" + if json_out: + _emit_error("UPGRADE_REQUIRED", message, None, json_out) + return + from onepin._cli.render import echo_warning + + echo_warning(message) + + +def _format_upgrade_required(exc: Exception) -> str: + """Build an upgrade message from a 426 ApiError body (best-effort), else a generic one.""" + from onepin._version_gate import format_upgrade_message + + required = None + body = getattr(exc, "body", None) + if isinstance(body, dict): + detail = body.get("error") if isinstance(body.get("error"), dict) else body + if isinstance(detail, dict): + raw = detail.get("required_version") or detail.get("minimum_version") + required = raw.strip() if isinstance(raw, str) and raw.strip() else None + return format_upgrade_message(required) diff --git a/src/onepin/_cli/_http.py b/src/onepin/_cli/_http.py index 03f18b7..21d1301 100644 --- a/src/onepin/_cli/_http.py +++ b/src/onepin/_cli/_http.py @@ -164,6 +164,27 @@ def _call_whoami(key: str, base_url: str, timeout: float = 10.0, *, verbose: boo error_code = parsed.get("code", error_code) error_message = parsed.get("message", error_message) + if response.status_code == 426: + # Server-enforced SDK floor. The raw auth path bypasses the SDK's api_errors() + # mapper, so surface the same UPGRADE_REQUIRED message + copy-paste command here. + from onepin._version_gate import format_upgrade_message, required_version_from + + required = required_version_from(dict(response.headers)) + if not required: + try: + err = response.json().get("error") + raw = err.get("required_version") or err.get("minimum_version") if isinstance(err, dict) else None + required = raw.strip() if isinstance(raw, str) and raw.strip() else None + except Exception: # noqa: BLE001 # pragma: no cover - defensive 426 body-parse guard + required = None + raise OnePinHTTPError( + format_upgrade_message(required), + status_code=426, + error_code="UPGRADE_REQUIRED", + request_id=request_id, + response_body=body_text, + ) + if response.status_code in (401, 403): raise OnePinAuthError( error_message, diff --git a/src/onepin/_cli/_skill/onepin/SKILL.md b/src/onepin/_cli/_skill/onepin/SKILL.md index aadba18..5ad47cd 100644 --- a/src/onepin/_cli/_skill/onepin/SKILL.md +++ b/src/onepin/_cli/_skill/onepin/SKILL.md @@ -38,6 +38,16 @@ integration surface — this skill teaches its contract, not a frozen command li workspace and the wrong one is active. 3. If `onepin schema` errors or the command is missing, the CLI is old — suggest `pip install -U onepin`, or fall back to `onepin --help`. +4. **Offer upgrades (advisory, never block the task).** Run `onepin upgrade-check` (it prints + nothing when up to date, snoozed, offline, or disabled via `ONEPIN_NO_UPDATE_CHECK`). + - On `UPGRADE_AVAILABLE `: ask with AskUserQuestion — **Upgrade now** + (run `onepin upgrade-check --mark-upgrading && pip install --upgrade onepin` — the marker lets + the next run confirm the new version), **Not now** (run `onepin upgrade-check --snooze` — an + escalating quiet period: 24h, then 48h, then 7d), or **Don't ask again** (run + `onepin upgrade-check --disable`). Continue with the task either way. + - On `JUST_UPGRADED `: tell the user they're now on v`` and continue. + Separately, if *any* command fails with `UPGRADE_REQUIRED` (or an HTTP 426), the SDK is too old to + talk to the API — surface the message and its `pip install --upgrade` command, and stop. ## Discover, don't guess diff --git a/src/onepin/_cli/_spec.py b/src/onepin/_cli/_spec.py index 34dc59e..326ccf2 100644 --- a/src/onepin/_cli/_spec.py +++ b/src/onepin/_cli/_spec.py @@ -867,21 +867,6 @@ def _list_opts(*extra: Opt) -> list[Opt]: options=[_JSON], unwrap="data", ), - # --- health ------------------------------------------------------------------------- - Cmd( - "health", - "live", - "health.liveness", - "Liveness probe.", - options=[_JSON], - unwrap="data", - ), - Cmd( - "health", - "ready", - "health.readiness", - "Readiness probe.", - options=[_JSON], - unwrap="data", - ), + # health (live/ready) is hand-written in commands/health.py -- it blends local SDK version, + # the API's reported version, and version-gate headers, which the table model can't express. ] diff --git a/src/onepin/_cli/_update_check.py b/src/onepin/_cli/_update_check.py new file mode 100644 index 0000000..24f2021 --- /dev/null +++ b/src/onepin/_cli/_update_check.py @@ -0,0 +1,252 @@ +"""``onepin upgrade-check`` (hidden) -- gstack-style soft upgrade notifier. + +The CLI itself never nags on stderr; instead this command prints a single machine-readable +line that the ``/onepin`` agent skill reads and turns into an ``AskUserQuestion`` prompt: + + UPGRADE_AVAILABLE -- a newer release is on PyPI + JUST_UPGRADED -- a just-upgraded marker was found + (no output) -- up to date / snoozed / disabled / offline + +State lives under ``~/.onepin/`` (reusing the credentials home helper): + - ``update-check`` dual-TTL cache (UP_TO_DATE 60 min; UPGRADE_AVAILABLE replays ~12 h) + - ``update-snoozed`` `` `` escalating snooze (24 h / 48 h / 7 d) + - ``just-upgraded-from`` optional marker (````) cleared on read + +All network and filesystem failures are swallowed -- a version check must never break tooling. +""" + +from __future__ import annotations + +import os +import re +import time +from pathlib import Path +from typing import Optional + +import typer + +from onepin._cli import __version__ +from onepin._cli.auth.credentials import _home_path +from onepin._version_gate import is_older + +_PYPI_URL = "https://pypi.org/pypi/onepin/json" +_FETCH_TIMEOUT = 3.0 + +# Cache TTLs (minutes): re-check hourly when current; replay the nag ~12 h when behind. +_TTL_UP_TO_DATE = 60 +_TTL_UPGRADE_AVAILABLE = 720 + +# Snooze durations (seconds) by escalation level; level 3+ caps at a week. +_SNOOZE_DURATIONS = {1: 86_400, 2: 172_800} +_SNOOZE_DEFAULT = 604_800 + +_VERSION_RE = re.compile(r"^\d+\.\d+") + + +def _state_dir() -> Path: + return _home_path() / ".onepin" + + +def _cache_path() -> Path: + return _state_dir() / "update-check" + + +def _snooze_path() -> Path: + return _state_dir() / "update-snoozed" + + +def _marker_path() -> Path: + return _state_dir() / "just-upgraded-from" + + +def _disabled_path() -> Path: + return _state_dir() / "update-check-disabled" + + +def _read_text(path: Path) -> str: + try: + return path.read_text(encoding="utf-8").strip() + except OSError: + return "" + + +def _atomic_write(path: Path, content: str) -> None: + """Write ``content`` to ``path`` atomically (tmp + os.replace); swallow OS errors. + + Avoids a torn read when two ``onepin`` processes touch the same state file at once. + """ + import tempfile + + try: + path.parent.mkdir(parents=True, exist_ok=True) + fd, tmp = tempfile.mkstemp(dir=str(path.parent), suffix=".tmp") + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + handle.write(content) + os.replace(tmp, str(path)) + except OSError: # pragma: no cover - best-effort temp cleanup + try: + os.unlink(tmp) + except OSError: + pass + except OSError: # pragma: no cover - a cache write must never break the CLI + pass + + +def _write_cache(line: str) -> None: + _atomic_write(_cache_path(), line + "\n") + + +def _age_minutes(path: Path) -> float: + try: + return (time.time() - path.stat().st_mtime) / 60.0 + except OSError: # pragma: no cover - missing/unreadable cache file + return float("inf") + + +def _fetch_latest() -> Optional[str]: + """Best-effort latest stable version from PyPI; ``None`` on any failure/offline.""" + try: + import httpx + + response = httpx.get(_PYPI_URL, timeout=_FETCH_TIMEOUT) + if response.status_code != 200: + return None + version = str(response.json().get("info", {}).get("version", "")).strip() + return version if _VERSION_RE.match(version) else None + except Exception: # noqa: BLE001 - network/parse failures degrade to "no info" + return None + + +def _parse_cache(line: str) -> Optional[tuple[str, str]]: + """Parse a cache line into ``(kind, latest)``; ``None`` if unrecognized.""" + parts = line.split() + if not parts: + return None + kind = parts[0] + if kind == "UP_TO_DATE" and len(parts) >= 2: + return ("UP_TO_DATE", parts[1]) + if kind == "UPGRADE_AVAILABLE" and len(parts) >= 3: + return ("UPGRADE_AVAILABLE", parts[2]) + return None + + +def cached_latest() -> Optional[str]: + """Latest version recorded in the cache (any freshness); ``None`` if unknown. + + Used by ``onepin health`` to show the recommended version without a network call. + """ + parsed = _parse_cache(_read_text(_cache_path())) + return parsed[1] if parsed else None + + +def _read_fresh_latest() -> Optional[str]: + """Latest version from the cache only if still within its per-kind TTL, else ``None``.""" + path = _cache_path() + parsed = _parse_cache(_read_text(path)) + if parsed is None: + return None + kind, latest = parsed + ttl = _TTL_UP_TO_DATE if kind == "UP_TO_DATE" else _TTL_UPGRADE_AVAILABLE + if _age_minutes(path) > ttl: + return None + return latest + + +def _resolve_latest(current: str) -> Optional[str]: + """Return the latest version (cache when fresh, else fetch + cache); ``None`` if unknown.""" + fresh = _read_fresh_latest() + if fresh is not None: + return fresh + latest = _fetch_latest() + if latest is None: + # Offline / fetch error: do not cache. A failure must not masquerade as "up to date" + # (which would also mislead `onepin health`), and the next run should retry rather than + # stay silent for the full TTL. + return None + if is_older(current, latest): + _write_cache(f"UPGRADE_AVAILABLE {current} {latest}") + else: + _write_cache(f"UP_TO_DATE {current}") + return latest + + +def _is_snoozed(latest: str) -> bool: + parts = _read_text(_snooze_path()).split() + if len(parts) != 3: + return False + version, level, epoch = parts + if version != latest or not level.isdigit() or not epoch.isdigit(): + return False # a new release resets the snooze + duration = _SNOOZE_DURATIONS.get(int(level), _SNOOZE_DEFAULT) + return time.time() < int(epoch) + duration + + +def _bump_snooze() -> None: + """Escalate the snooze for the currently-cached upgrade (level +1, or 1 if new).""" + latest = cached_latest() + parsed = _parse_cache(_read_text(_cache_path())) + if not latest or parsed is None or parsed[0] != "UPGRADE_AVAILABLE": + return + existing = _read_text(_snooze_path()).split() + level = 1 + if len(existing) == 3 and existing[0] == latest and existing[1].isdigit(): + level = int(existing[1]) + 1 + _atomic_write(_snooze_path(), f"{latest} {level} {int(time.time())}\n") + + +def _consume_marker(current: str) -> bool: + """If a just-upgraded marker exists, emit ``JUST_UPGRADED`` and return True.""" + marker = _marker_path() + old = _read_text(marker) + if not old: + return False + try: + marker.unlink() + except OSError: # pragma: no cover - best-effort marker cleanup + pass + try: + _snooze_path().unlink(missing_ok=True) + except OSError: # pragma: no cover - best-effort snooze cleanup + pass + _write_cache(f"UP_TO_DATE {current}") + if old != current: + typer.echo(f"JUST_UPGRADED {old} {current}") + return True + + +def upgrade_check( + force: bool = typer.Option(False, "--force", help="Bypass the cache and re-check now."), + snooze: bool = typer.Option(False, "--snooze", help="Snooze the current upgrade prompt (escalating)."), + disable: bool = typer.Option( + False, "--disable", help="Stop all future upgrade checks (rm ~/.onepin/update-check-disabled to re-enable)." + ), + mark_upgrading: bool = typer.Option( + False, "--mark-upgrading", help="Record the current version before upgrading so the next run can confirm it." + ), +) -> None: + """Print a one-line upgrade signal for the agent skill (hidden; not for direct use).""" + if mark_upgrading: + _atomic_write(_marker_path(), f"{__version__}\n") + return + if disable: + _atomic_write(_disabled_path(), "1\n") + return + if os.environ.get("ONEPIN_NO_UPDATE_CHECK") or _disabled_path().exists(): + return + + current = __version__ + + if snooze: + _bump_snooze() + return + + if force: + _cache_path().unlink(missing_ok=True) + + if _consume_marker(current): + return + + latest = _resolve_latest(current) + if latest and is_older(current, latest) and not _is_snoozed(latest): + typer.echo(f"UPGRADE_AVAILABLE {current} {latest}") diff --git a/src/onepin/_cli/commands/_registry.py b/src/onepin/_cli/commands/_registry.py index d5b216a..3f69b8f 100644 --- a/src/onepin/_cli/commands/_registry.py +++ b/src/onepin/_cli/commands/_registry.py @@ -15,7 +15,8 @@ from onepin._cli import _dispatch, _manifest from onepin._cli._spec import TABLE, Cmd -from onepin._cli.commands import auth, composites, skill +from onepin._cli._update_check import upgrade_check +from onepin._cli.commands import auth, composites, health, skill # Per-group help text. Groups not listed fall back to a generic header. _GROUP_HELP = { @@ -60,6 +61,15 @@ def register(app: typer.Typer) -> None: skill_app.command(name="uninstall", help="Remove the installed OnePin agent skill.")(skill.uninstall) app.add_typer(skill_app, name="skill", help="Manage the OnePin agent skill for AI coding tools.") + # health (live/ready): hand-written so it can surface SDK/API/recommended/required versions. + health_app = typer.Typer(help=_GROUP_HELP["health"], no_args_is_help=True) + health_app.command(name="live", help="Liveness probe.")(health.live) + health_app.command(name="ready", help="Readiness probe.")(health.ready) + app.add_typer(health_app, name="health", help=_GROUP_HELP["health"]) + + # Hidden: the /onepin agent skill runs this to drive the upgrade prompt (not user-facing). + app.command(name="upgrade-check", hidden=True)(upgrade_check) + def _build_groups() -> dict[str, typer.Typer]: """Build one sub-Typer per group, with nested sub-Typers for subgroups, from the TABLE.""" diff --git a/src/onepin/_cli/commands/health.py b/src/onepin/_cli/commands/health.py new file mode 100644 index 0000000..5de948f --- /dev/null +++ b/src/onepin/_cli/commands/health.py @@ -0,0 +1,77 @@ +"""``onepin health`` -- liveness/readiness probes plus the version/status surface. + +Beyond the probe status, this reports the installed SDK version, the API's reported version, +the recommended version (latest on PyPI, from the upgrade-check cache -- no network here), and +the required floor (the ``X-OnePin-Required-Version`` response header). Hand-written (rather than +table-driven) so it can blend local, cached, and header-sourced facts into one view. +""" + +from __future__ import annotations + +from typing import Any, Optional + +import typer + +from onepin._cli import __version__ +from onepin._cli._ctx import api_errors, get_client, output_json +from onepin._cli.render import render_json +from onepin._version_gate import required_version_from + +# (info key, human label) in display order. +_FIELDS = [ + ("status", "status"), + ("sdk_version", "SDK version"), + ("api_version", "API version"), + ("recommended_version", "Recommended SDK version"), + ("required_version", "Required SDK version"), +] + + +def _recommended() -> Optional[str]: + """Latest version from the upgrade-check cache (no network); None if never checked.""" + from onepin._cli._update_check import cached_latest + + return cached_latest() + + +def _build_info(data: Any, headers: Any) -> dict[str, Optional[str]]: + body = data if isinstance(data, dict) else {} + status = body.get("status") or "ok" + return { + "status": status, + "sdk_version": __version__, + "api_version": body.get("version"), + "recommended_version": _recommended(), + "required_version": required_version_from(headers), + } + + +def _emit(info: dict[str, Optional[str]], json_on: bool) -> None: + if json_on: + render_json({key: value for key, value in info.items() if value is not None}) + return + for key, label in _FIELDS: + value = info.get(key) + typer.echo(f"{label}: {value if value is not None else 'unknown'}") + + +def live( + json_output_local: bool = typer.Option(False, "--json", help="Emit JSON instead of a table."), +) -> None: + """Liveness probe.""" + json_on = output_json(json_output_local) + with api_errors(json_on): + client = get_client() + response = client.health.with_raw_response.liveness() + _emit(_build_info(response.data, response.headers), json_on) + + +def ready( + json_output_local: bool = typer.Option(False, "--json", help="Emit JSON instead of a table."), +) -> None: + """Readiness probe.""" + json_on = output_json(json_output_local) + with api_errors(json_on): + client = get_client() + response = client.health.with_raw_response.readiness() + _emit(_build_info(response.data, response.headers), json_on) diff --git a/src/onepin/_cli/render.py b/src/onepin/_cli/render.py index 6e15c13..d65fabc 100644 --- a/src/onepin/_cli/render.py +++ b/src/onepin/_cli/render.py @@ -12,11 +12,11 @@ from typing import Any, Dict, List, Optional -def _use_color() -> bool: - """Return True if ANSI color should be used. +def _use_color(stream: Any = None) -> bool: + """Return True if ANSI color should be used for ``stream`` (defaults to stdout). Honors the ``--no-color`` flag (captured in root state), then the W3C - ``NO_COLOR`` env var, then TTY detection. + ``NO_COLOR`` env var, then TTY detection on the target stream. """ from onepin._cli import _state @@ -24,11 +24,19 @@ def _use_color() -> bool: return False if os.environ.get("NO_COLOR"): return False - if not sys.stdout.isatty(): + target = sys.stdout if stream is None else stream + isatty = getattr(target, "isatty", None) + if not (callable(isatty) and isatty()): return False return True +def echo_warning(message: str) -> None: + """Write a warning to stderr, yellow when color is enabled (NO_COLOR / non-TTY safe).""" + text = f"\033[33m{message}\033[0m" if _use_color(sys.stderr) else message + print(text, file=sys.stderr) + + def render_json(data: Any) -> None: """Emit JSON to stdout. No envelope -- just the raw payload (.data).""" print(json.dumps(data, indent=2, default=str)) diff --git a/src/onepin/_version_gate.py b/src/onepin/_version_gate.py new file mode 100644 index 0000000..0d2001d --- /dev/null +++ b/src/onepin/_version_gate.py @@ -0,0 +1,210 @@ +"""Client-side SDK version gate (hand-written; preserved across ``fern generate`` via ``.fernignore``). + +The OnePin API advertises the minimum SDK version it still accepts via the +``X-OnePin-Required-Version`` response header (and enforces it with HTTP 426). This module reads +that header off every response and stops the caller when the installed ``onepin`` package is +older than the floor. + +It is intentionally free of any CLI dependency so the generated SDK can re-export +:func:`make_client` (see ``fern/generators.yml`` ``additional_init_exports``) and programmatic +users get the same gate: + + import onepin + + client = onepin.make_client(token="...") # version-gated + client.workflows.list() # raises OnePinUpgradeRequiredError if too old + +A bare ``OnePinClient(token=...)`` is not hooked, but still hits the server-side 426 (which the +generated client surfaces as ``ApiError``). The "recommended" (soft) upgrade nudge lives in the +CLI/agent skill, not here -- this module is only the hard ``required`` floor. +""" + +from __future__ import annotations + +import typing +from importlib.metadata import PackageNotFoundError +from importlib.metadata import version as _pkg_version + +if typing.TYPE_CHECKING: # pragma: no cover - type-only imports + import httpx + + from onepin.client import AsyncOnePinClient, OnePinClient + +#: Response header the API uses to advertise the minimum acceptable SDK version. +REQUIRED_VERSION_HEADER = "X-OnePin-Required-Version" + +_BASE_UPGRADE_COMMAND = "pip install --upgrade onepin" + + +def installed_version() -> str: + """The installed ``onepin`` package version (``0.0.0+local`` for an unbuilt editable tree).""" + try: + return _pkg_version("onepin") + except PackageNotFoundError: + return "0.0.0+local" + + +def _is_valid_version(value: typing.Optional[str]) -> bool: + """True if ``value`` is a parseable PEP 440 version. + + Server-supplied required-version values are interpolated into a copy-paste ``pip`` command, + so they MUST be validated first -- a value like ``0.5.0' --extra-index-url https://evil #`` + (e.g. from a rogue ``--base-url``) must never become extra shell arguments. + """ + if not value: + return False + try: + from packaging.version import InvalidVersion, Version + except ModuleNotFoundError: # tuple-fallback world: accept only digits + dots + import re + + return bool(re.match(r"^\d+(\.\d+)*$", value)) + try: + Version(value) + except InvalidVersion: + return False + return True + + +def upgrade_command(required: typing.Optional[str] = None) -> str: + """Copy-paste command that upgrades the SDK -- pinned to the floor only if it's a valid version.""" + if _is_valid_version(required): + return f"pip install --upgrade 'onepin>={required}'" + return _BASE_UPGRADE_COMMAND + + +def format_upgrade_message(required: typing.Optional[str], current: typing.Optional[str] = None) -> str: + """The single, shared 'you must upgrade' sentence used by every surface (CLI, SDK, auth path). + + An unparseable/absent ``required`` degrades to the generic, unpinned message so a malformed or + hostile server value is never echoed into the suggested command. + """ + cur = current or installed_version() + if _is_valid_version(required): + return f"onepin {cur} is below the required minimum {required}. Upgrade: {upgrade_command(required)}" + return f"onepin {cur} is no longer supported by the API. Upgrade: {upgrade_command()}" + + +class OnePinUpgradeRequiredError(Exception): + """Raised when the installed SDK is older than the API's required floor. + + Carries the structured facts (``required``, ``current``, ``upgrade_command``) so a caller can + render its own message; ``str(exc)`` is a complete, copy-paste-ready sentence. + """ + + def __init__(self, *, required: str, current: typing.Optional[str] = None) -> None: + self.required = required + self.current = current or installed_version() + self.upgrade_command = upgrade_command(required) + super().__init__(format_upgrade_message(required, self.current)) + + +def required_version_from(headers: typing.Optional[typing.Mapping[str, str]]) -> typing.Optional[str]: + """Extract the required-version header value, case-insensitively (httpx lowercases keys).""" + if not headers: + return None + target = REQUIRED_VERSION_HEADER.lower() + for key, value in headers.items(): + if key.lower() == target: + return value.strip() or None + return None + + +def is_older(current: str, required: str) -> bool: + """True if ``current`` < ``required``. Unparseable versions are treated as not-older (no-op).""" + try: + from packaging.version import InvalidVersion, Version + except ModuleNotFoundError: # packaging not installed -> tuple fallback + return _tuple_older(current, required) + try: + return Version(current) < Version(required) + except InvalidVersion: + return False + + +def _tuple_older(current: str, required: str) -> bool: + """Best-effort numeric-dotted compare used only when ``packaging`` is unavailable.""" + + def parse(value: str) -> tuple[int, ...]: + head = value.split("+", 1)[0].split("-", 1)[0] + parts: list[int] = [] + for piece in head.split("."): + if not piece.isdigit(): + break + parts.append(int(piece)) + return tuple(parts) + + try: + return parse(current) < parse(required) + except Exception: # noqa: BLE001 - never let a compare crash a request + return False + + +def check_required( + headers: typing.Optional[typing.Mapping[str, str]], + *, + current: typing.Optional[str] = None, +) -> None: + """Raise :class:`OnePinUpgradeRequiredError` if ``headers`` advertise a floor above the install. + + Missing/blank/unparseable header -> no-op, so the gate stays inert until the API emits it. + """ + required = required_version_from(headers) + if not required: + return + installed = current or installed_version() + if is_older(installed, required): + raise OnePinUpgradeRequiredError(required=required, current=installed) + + +def _response_hook(response: "httpx.Response") -> None: + """httpx response event hook -- reads only headers, so it never consumes the body.""" + check_required(response.headers) + + +async def _async_response_hook(response: "httpx.Response") -> None: + check_required(response.headers) + + +def _user_agent() -> str: + return f"onepin/{installed_version()}" + + +def make_client(**kwargs: typing.Any) -> OnePinClient: + """Build a version-gated :class:`~onepin.client.OnePinClient`. + + Drop-in for ``OnePinClient(...)``: injects an httpx client whose response hook enforces the + server's required-version floor and corrects the ``User-Agent`` to the true installed version + (the generated default is baked at codegen time and can drift). A caller-supplied + ``httpx_client`` is respected (assumed already configured); ``headers`` are merged. + """ + import httpx + + from onepin.client import OnePinClient + + headers = dict(kwargs.pop("headers", None) or {}) + headers.setdefault("User-Agent", _user_agent()) + if "httpx_client" not in kwargs: + kwargs["httpx_client"] = httpx.Client( + timeout=kwargs.pop("timeout", 60), + follow_redirects=kwargs.pop("follow_redirects", True), + event_hooks={"response": [_response_hook]}, + ) + return OnePinClient(headers=headers, **kwargs) + + +def make_async_client(**kwargs: typing.Any) -> AsyncOnePinClient: + """Async counterpart of :func:`make_client`.""" + import httpx + + from onepin.client import AsyncOnePinClient + + headers = dict(kwargs.pop("headers", None) or {}) + headers.setdefault("User-Agent", _user_agent()) + if "httpx_client" not in kwargs: + kwargs["httpx_client"] = httpx.AsyncClient( + timeout=kwargs.pop("timeout", 60), + follow_redirects=kwargs.pop("follow_redirects", True), + event_hooks={"response": [_async_response_hook]}, + ) + return AsyncOnePinClient(headers=headers, **kwargs) diff --git a/tests/cli/test_cli_health.py b/tests/cli/test_cli_health.py new file mode 100644 index 0000000..cc8d5a6 --- /dev/null +++ b/tests/cli/test_cli_health.py @@ -0,0 +1,147 @@ +"""CLI tests for `onepin health` (version/status surface) and the required-version gate. + +Uses respx so the real Fern request/response path and the version-gate response hook both run. +""" + +from __future__ import annotations + +import json + +import httpx +import pytest +import respx +from typer.testing import CliRunner + +from onepin._cli import _update_check as uc +from onepin._cli.main import app + +runner = CliRunner() +_BASE = "https://api.onepin.ai" + + +def _invoke(argv: list[str]): + return runner.invoke(app, ["--api-key", "op_live_x", "--base-url", _BASE, *argv]) + + +@pytest.fixture(autouse=True) +def _fixed_sdk_version(monkeypatch: pytest.MonkeyPatch) -> None: + # Pin the version shown as "SDK version" so assertions are stable. + monkeypatch.setattr("onepin._cli.commands.health.__version__", "0.6.0") + + +def _seed_recommended(version: str) -> None: + path = uc._cache_path() + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(f"UPGRADE_AVAILABLE 0.6.0 {version}\n", encoding="utf-8") + + +class TestHealthLive: + @respx.mock + def test_human_full_surface(self, tmp_home) -> None: + _seed_recommended("0.9.0") + respx.get(f"{_BASE}/health").mock( + return_value=httpx.Response( + 200, + json={"status": "ok", "version": "0.34.3"}, + headers={"X-OnePin-Required-Version": "0.1.0"}, + ) + ) + result = _invoke(["health", "live"]) + assert result.exit_code == 0, result.output + assert "status: ok" in result.output + assert "SDK version: 0.6.0" in result.output + assert "API version: 0.34.3" in result.output + assert "Recommended SDK version: 0.9.0" in result.output + assert "Required SDK version: 0.1.0" in result.output + + @respx.mock + def test_json(self, tmp_home) -> None: + _seed_recommended("0.9.0") + respx.get(f"{_BASE}/health").mock( + return_value=httpx.Response( + 200, + json={"status": "ok", "version": "0.34.3"}, + headers={"X-OnePin-Required-Version": "0.1.0"}, + ) + ) + result = _invoke(["health", "live", "--json"]) + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + assert payload["status"] == "ok" + assert payload["sdk_version"] == "0.6.0" + assert payload["api_version"] == "0.34.3" + assert payload["recommended_version"] == "0.9.0" + assert payload["required_version"] == "0.1.0" + + @respx.mock + def test_unknown_sources(self, tmp_home) -> None: + # No version field in the body, no required header, no cached recommended. + respx.get(f"{_BASE}/health").mock(return_value=httpx.Response(200, json={})) + result = _invoke(["health", "live"]) + assert result.exit_code == 0, result.output + assert "status: ok" in result.output # synthesized on a 200 + assert "API version: unknown" in result.output + assert "Recommended SDK version: unknown" in result.output + assert "Required SDK version: unknown" in result.output + + +class TestRequiredGate: + @respx.mock + def test_stop_via_header_hook(self, tmp_home) -> None: + # A floor above any real installed version trips the client-side response hook. + respx.get(f"{_BASE}/health").mock( + return_value=httpx.Response(200, json={"status": "ok"}, headers={"X-OnePin-Required-Version": "999.0.0"}) + ) + result = _invoke(["health", "live"]) + assert result.exit_code == 1 + assert "999.0.0" in result.output + assert "pip install --upgrade" in result.output + + @respx.mock + def test_stop_json_envelope(self, tmp_home) -> None: + # --json upgrade failures emit the structured error envelope (UPGRADE_REQUIRED). + respx.get(f"{_BASE}/health").mock( + return_value=httpx.Response(200, json={}, headers={"X-OnePin-Required-Version": "999.0.0"}) + ) + result = runner.invoke(app, ["--api-key", "op_live_x", "--base-url", _BASE, "--json", "health", "live"]) + assert result.exit_code == 1 + assert '"code": "UPGRADE_REQUIRED"' in result.output + + @respx.mock + def test_stop_via_server_426(self, tmp_home) -> None: + respx.get(f"{_BASE}/health").mock( + return_value=httpx.Response( + 426, + json={"error": {"code": "sdk_upgrade_required", "required_version": "9.9.9"}}, + ) + ) + result = _invoke(["health", "live"]) + assert result.exit_code == 1 + assert "9.9.9" in result.output + assert "pip install --upgrade" in result.output + + +class TestHealthReady: + @respx.mock + def test_human(self, tmp_home) -> None: + respx.get(f"{_BASE}/ready").mock(return_value=httpx.Response(200, json={"status": "ok"})) + result = _invoke(["health", "ready"]) + assert result.exit_code == 0, result.output + assert "status: ok" in result.output + assert "SDK version: 0.6.0" in result.output + + +class TestAuthPath426: + @respx.mock + def test_whoami_surfaces_upgrade(self, tmp_home) -> None: + # The raw-httpx auth path must also surface a 426 floor as an upgrade stop. + respx.get(f"{_BASE}/api/v1/auth/whoami").mock( + return_value=httpx.Response( + 426, json={"error": {"code": "sdk_upgrade_required", "required_version": "9.9.9"}} + ) + ) + result = runner.invoke(app, ["--api-key", "op_live_x", "--base-url", _BASE, "whoami"]) + assert result.exit_code == 1 + assert "UPGRADE_REQUIRED" in result.output + assert "9.9.9" in result.output + assert "pip install --upgrade" in result.output diff --git a/tests/unit/test_update_check.py b/tests/unit/test_update_check.py new file mode 100644 index 0000000..ba98417 --- /dev/null +++ b/tests/unit/test_update_check.py @@ -0,0 +1,171 @@ +"""Unit tests for the gstack-style soft upgrade notifier (`onepin upgrade-check`).""" + +from __future__ import annotations + +import httpx +import pytest +import respx +from typer.testing import CliRunner + +from onepin._cli import _update_check as uc +from onepin._cli.main import app + +runner = CliRunner() + + +@pytest.fixture(autouse=True) +def _fixed_version(monkeypatch: pytest.MonkeyPatch) -> None: + """Pin the "installed" version so comparisons are deterministic (real one is VCS-derived).""" + monkeypatch.setattr(uc, "__version__", "0.6.0") + + +def _seed_cache(line: str) -> None: + path = uc._cache_path() + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(line + "\n", encoding="utf-8") + + +class TestUpgradeAvailable: + def test_fetch_prints_and_caches(self, tmp_home, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(uc, "_fetch_latest", lambda: "0.9.0") + result = runner.invoke(app, ["upgrade-check"]) + assert result.exit_code == 0 + assert result.output.strip() == "UPGRADE_AVAILABLE 0.6.0 0.9.0" + assert uc._read_text(uc._cache_path()) == "UPGRADE_AVAILABLE 0.6.0 0.9.0" + + def test_fresh_cache_replays_without_fetching(self, tmp_home, monkeypatch: pytest.MonkeyPatch) -> None: + _seed_cache("UPGRADE_AVAILABLE 0.6.0 0.9.0") + + def _boom() -> str: + raise AssertionError("fetch must not happen on a fresh cache") + + monkeypatch.setattr(uc, "_fetch_latest", _boom) + result = runner.invoke(app, ["upgrade-check"]) + assert result.output.strip() == "UPGRADE_AVAILABLE 0.6.0 0.9.0" + + +class TestQuietPaths: + def test_up_to_date_is_silent(self, tmp_home, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(uc, "_fetch_latest", lambda: "0.6.0") + result = runner.invoke(app, ["upgrade-check"]) + assert result.output.strip() == "" + + def test_offline_is_silent(self, tmp_home, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(uc, "_fetch_latest", lambda: None) + result = runner.invoke(app, ["upgrade-check"]) + assert result.output.strip() == "" + + def test_opt_out_env_silences(self, tmp_home, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(uc, "_fetch_latest", lambda: "0.9.0") + monkeypatch.setenv("ONEPIN_NO_UPDATE_CHECK", "1") + result = runner.invoke(app, ["upgrade-check"]) + assert result.output.strip() == "" + + +class TestSnooze: + def test_escalates_and_silences(self, tmp_home, monkeypatch: pytest.MonkeyPatch) -> None: + _seed_cache("UPGRADE_AVAILABLE 0.6.0 0.9.0") + # First decline -> level 1. + runner.invoke(app, ["upgrade-check", "--snooze"]) + snooze = uc._read_text(uc._snooze_path()).split() + assert snooze[0] == "0.9.0" and snooze[1] == "1" + # While snoozed for this version, the prompt stays quiet. + monkeypatch.setattr(uc, "_fetch_latest", lambda: "0.9.0") + assert runner.invoke(app, ["upgrade-check"]).output.strip() == "" + # Second decline -> level 2 (escalation). + runner.invoke(app, ["upgrade-check", "--snooze"]) + assert uc._read_text(uc._snooze_path()).split()[1] == "2" + + def test_new_version_resets_snooze(self, tmp_home, monkeypatch: pytest.MonkeyPatch) -> None: + _seed_cache("UPGRADE_AVAILABLE 0.6.0 0.9.0") + runner.invoke(app, ["upgrade-check", "--snooze"]) # snooze 0.9.0 + # A newer release appears; the snooze for 0.9.0 must not suppress 0.9.5. + _seed_cache("UPGRADE_AVAILABLE 0.6.0 0.9.5") + monkeypatch.setattr(uc, "_fetch_latest", lambda: "0.9.5") + result = runner.invoke(app, ["upgrade-check"]) + assert result.output.strip() == "UPGRADE_AVAILABLE 0.6.0 0.9.5" + + +class TestMarkUpgrading: + def test_mark_then_just_upgraded(self, tmp_home, monkeypatch: pytest.MonkeyPatch) -> None: + # "Upgrade now" writes the marker at the current version... + runner.invoke(app, ["upgrade-check", "--mark-upgrading"]) + assert uc._marker_path().read_text().strip() == "0.6.0" + # ...then after pip bumps the version, the next run confirms the upgrade. + monkeypatch.setattr(uc, "__version__", "0.9.0") + result = runner.invoke(app, ["upgrade-check"]) + assert result.output.strip() == "JUST_UPGRADED 0.6.0 0.9.0" + + +class TestMarker: + def test_just_upgraded(self, tmp_home) -> None: + marker = uc._marker_path() + marker.parent.mkdir(parents=True, exist_ok=True) + marker.write_text("0.5.0\n", encoding="utf-8") + result = runner.invoke(app, ["upgrade-check"]) + assert result.output.strip() == "JUST_UPGRADED 0.5.0 0.6.0" + assert not marker.exists() + + +class TestCachedLatest: + def test_reads_latest_regardless_of_freshness(self, tmp_home) -> None: + assert uc.cached_latest() is None + _seed_cache("UPGRADE_AVAILABLE 0.6.0 0.9.0") + assert uc.cached_latest() == "0.9.0" + _seed_cache("UP_TO_DATE 0.6.0") + assert uc.cached_latest() == "0.6.0" + + +class TestDisable: + def test_disable_silences_future_checks(self, tmp_home, monkeypatch: pytest.MonkeyPatch) -> None: + runner.invoke(app, ["upgrade-check", "--disable"]) + assert uc._disabled_path().exists() + # Even with an upgrade available, a disabled check stays silent and skips the fetch. + monkeypatch.setattr(uc, "_fetch_latest", lambda: "0.9.0") + result = runner.invoke(app, ["upgrade-check"]) + assert result.output.strip() == "" + + +class TestFetchLatest: + @respx.mock + def test_returns_pypi_version(self, tmp_home) -> None: + respx.get(uc._PYPI_URL).mock(return_value=httpx.Response(200, json={"info": {"version": "1.2.3"}})) + assert uc._fetch_latest() == "1.2.3" + + @respx.mock + def test_non_200_is_none(self, tmp_home) -> None: + respx.get(uc._PYPI_URL).mock(return_value=httpx.Response(503)) + assert uc._fetch_latest() is None + + @respx.mock + def test_unparseable_version_is_none(self, tmp_home) -> None: + respx.get(uc._PYPI_URL).mock(return_value=httpx.Response(200, json={"info": {"version": "nightly"}})) + assert uc._fetch_latest() is None + + +class TestEdgeBranches: + def test_force_refetches(self, tmp_home, monkeypatch: pytest.MonkeyPatch) -> None: + _seed_cache("UPGRADE_AVAILABLE 0.6.0 9.9.9") + monkeypatch.setattr(uc, "_fetch_latest", lambda: "0.6.0") # now up to date + result = runner.invoke(app, ["upgrade-check", "--force"]) + assert result.output.strip() == "" + assert uc._read_text(uc._cache_path()) == "UP_TO_DATE 0.6.0" + + def test_snooze_without_cached_upgrade_is_noop(self, tmp_home) -> None: + runner.invoke(app, ["upgrade-check", "--snooze"]) + assert not uc._snooze_path().exists() + + def test_stale_cache_triggers_refetch(self, tmp_home, monkeypatch: pytest.MonkeyPatch) -> None: + import os + import time + + _seed_cache("UP_TO_DATE 0.6.0") + old = time.time() - (uc._TTL_UP_TO_DATE + 5) * 60 + os.utime(uc._cache_path(), (old, old)) + monkeypatch.setattr(uc, "_fetch_latest", lambda: "0.9.0") + result = runner.invoke(app, ["upgrade-check"]) + assert result.output.strip() == "UPGRADE_AVAILABLE 0.6.0 0.9.0" # stale -> refetched + + def test_garbage_cache_is_none(self, tmp_home) -> None: + _seed_cache("garbage line here") + assert uc.cached_latest() is None diff --git a/tests/unit/test_version_gate.py b/tests/unit/test_version_gate.py new file mode 100644 index 0000000..4fea49a --- /dev/null +++ b/tests/unit/test_version_gate.py @@ -0,0 +1,103 @@ +"""Unit tests for the SDK version gate (src/onepin/_version_gate.py).""" + +from __future__ import annotations + +import httpx +import pytest + +from onepin import _version_gate as vg + + +class TestIsOlder: + def test_strictly_older(self) -> None: + assert vg.is_older("0.5.0", "0.6.0") is True + + def test_equal_not_older(self) -> None: + assert vg.is_older("0.6.0", "0.6.0") is False + + def test_newer_not_older(self) -> None: + assert vg.is_older("1.0.0", "0.6.0") is False + + def test_prerelease_below_release(self) -> None: + # A dev build of the same line is older than the final release. + assert vg.is_older("0.6.0.dev1", "0.6.0") is True + + def test_unparseable_is_not_older(self) -> None: + assert vg.is_older("not-a-version", "0.6.0") is False + + +class TestRequiredVersionFrom: + def test_case_insensitive(self) -> None: + assert vg.required_version_from({"X-OnePin-Required-Version": "0.7.0"}) == "0.7.0" + assert vg.required_version_from({"x-onepin-required-version": "0.7.0"}) == "0.7.0" + + def test_missing_and_blank(self) -> None: + assert vg.required_version_from({}) is None + assert vg.required_version_from(None) is None + assert vg.required_version_from({"x-onepin-required-version": " "}) is None + + +class TestCheckRequired: + def test_raises_when_below_floor(self) -> None: + with pytest.raises(vg.OnePinUpgradeRequiredError) as exc: + vg.check_required({"x-onepin-required-version": "9.9.9"}, current="0.6.0") + message = str(exc.value) + assert "9.9.9" in message + assert "pip install --upgrade 'onepin>=9.9.9'" in message + assert exc.value.required == "9.9.9" + assert exc.value.current == "0.6.0" + + def test_noop_at_or_above_floor(self) -> None: + vg.check_required({"x-onepin-required-version": "0.6.0"}, current="0.6.0") + vg.check_required({"x-onepin-required-version": "0.1.0"}, current="0.6.0") + + def test_noop_when_header_absent(self) -> None: + vg.check_required({}, current="0.6.0") + + +class TestUpgradeCommand: + def test_pins_to_floor_when_known(self) -> None: + assert vg.upgrade_command("0.7.0") == "pip install --upgrade 'onepin>=0.7.0'" + + def test_bare_when_unknown(self) -> None: + assert vg.upgrade_command() == "pip install --upgrade onepin" + + def test_malicious_version_falls_back_to_unpinned(self) -> None: + # A hostile/malformed server value must never be interpolated into the shell command. + evil = "0.5.0' --extra-index-url https://evil.example/simple #" + assert vg.upgrade_command(evil) == "pip install --upgrade onepin" + msg = vg.format_upgrade_message(evil) + assert "evil.example" not in msg + assert msg.endswith("pip install --upgrade onepin") + + +class TestMakeClient: + def test_injects_response_hook_and_user_agent(self) -> None: + client = vg.make_client(token="op_live_x", base_url="https://dev-api.onepin.ai") + raw = client._client_wrapper.httpx_client.httpx_client + assert raw.event_hooks.get("response"), "response hook not installed" + # User-Agent is corrected to the true installed version (not the codegen-baked default). + assert client._client_wrapper.get_headers()["User-Agent"] == vg._user_agent() + + def test_respects_caller_httpx_client(self) -> None: + custom = httpx.Client() + client = vg.make_client(token="op_live_x", httpx_client=custom) + assert client._client_wrapper.httpx_client.httpx_client is custom + + +class TestMakeAsyncClient: + def test_injects_response_hook_and_user_agent(self) -> None: + client = vg.make_async_client(token="op_live_x", base_url="https://dev-api.onepin.ai") + raw = client._client_wrapper.httpx_client.httpx_client + assert raw.event_hooks.get("response"), "async response hook not installed" + assert client._client_wrapper.get_headers()["User-Agent"] == vg._user_agent() + + +class TestPublicReexport: + def test_top_level_symbols(self) -> None: + # The gate's public API is re-exported from the package root. + from onepin import OnePinUpgradeRequiredError, make_async_client, make_client + + assert callable(make_client) + assert callable(make_async_client) + assert issubclass(OnePinUpgradeRequiredError, Exception) diff --git a/uv.lock b/uv.lock index fe855fc..0aab3a8 100644 --- a/uv.lock +++ b/uv.lock @@ -1016,6 +1016,7 @@ source = { editable = "." } dependencies = [ { name = "click" }, { name = "httpx" }, + { name = "packaging" }, { name = "pydantic" }, { name = "rich" }, { name = "tomli", marker = "python_full_version < '3.11'" }, @@ -1041,6 +1042,7 @@ dev = [ requires-dist = [ { name = "click", specifier = ">=8.1" }, { name = "httpx", specifier = ">=0.27,<1.0" }, + { name = "packaging", specifier = ">=23" }, { name = "pydantic", specifier = ">=2.7,<3.0" }, { name = "rich", specifier = ">=13" }, { name = "tomli", marker = "python_full_version < '3.11'" },