diff --git a/README.md b/README.md index 7bc3ef9..ecba293 100644 --- a/README.md +++ b/README.md @@ -34,6 +34,7 @@ pip install -e . ```bash komi-learn doctor # check the install and what to fix +komi-learn update # upgrade to the latest version (--check to only look) komi-learn status # config + how much it has learned komi-learn config # change any setting (menu, or `config set `) komi-learn sync # pull the latest community learnings diff --git a/komi/__init__.py b/komi/__init__.py index 3fe53b1..bf8f346 100644 --- a/komi/__init__.py +++ b/komi/__init__.py @@ -1,3 +1,23 @@ """komi-learn — a continuous, zero-friction learning layer for AI agents.""" -__version__ = "0.1.0" +# THE single source of truth for the version. pyproject.toml reads this literal at +# build time via setuptools dynamic version ([tool.setuptools.dynamic] version = +# {attr = "komi.__version__"}), so there is exactly one place to bump on a release +# and the packaged metadata can never drift from this value by construction. +__version__ = "0.3.0" + +# When installed, prefer the distribution metadata — it's the ground truth of what +# pip actually has on disk, which is what `komi-learn update` compares against +# PyPI. For a bare source tree (no installed dist) the literal above stands in. +# Because of the build-time attr binding the two agree, so this only matters for +# odd editable-install states — and it can only correct toward reality, never +# introduce a second hand-maintained number. +try: # importlib.metadata is stdlib on py3.8+ + from importlib.metadata import PackageNotFoundError, version as _dist_version + + try: + __version__ = _dist_version("komi-learn") + except PackageNotFoundError: + pass # not installed (source tree) — keep the literal +except Exception: # pragma: no cover - metadata API should always be present + pass diff --git a/komi/cli.py b/komi/cli.py index 6364dd9..80b6d72 100644 --- a/komi/cli.py +++ b/komi/cli.py @@ -299,6 +299,73 @@ def cmd_login(args) -> int: return rc +def cmd_update(args) -> int: + """Self-update: check PyPI for a newer komi-learn and upgrade in place. + + Upgrades *this* interpreter's install (the one the hooks import), via pip or + pipx depending on how komi-learn was installed. Use --check to only report + whether an update is available. After upgrading, re-run `komi-learn install` + is NOT required for code — but if a release adds new hook events, run it to + refresh your settings.""" + import komi + from komi import updater + from komi import cli_prompt as PR + + current = getattr(komi, "__version__", "?") + _p(f"{PRODUCT}: installed {current}. Checking PyPI…") + latest = updater.check_latest() + if latest is None: + _p(f"{PRODUCT}: couldn't reach PyPI (offline?). Try again later, or upgrade manually:") + _p(f" {updater.plan_upgrade().display()}") + return 1 + if not updater.is_newer(latest, current): + _p(f"{PRODUCT}: you're on the latest version ({current}). Nothing to do.") + return 0 + + _p(f"{PRODUCT}: a newer version is available — {current} → {latest}.") + plan = updater.plan_upgrade() + + if getattr(args, "check", False): + _p(f" to upgrade: {plan.display()}") + return 0 + + if not plan.runnable: + # Couldn't identify a safe package manager — never guess, just instruct. + _p(f" couldn't auto-detect how {PRODUCT} was installed. Upgrade with:") + _p(f" {plan.display()}") + return 1 + + if not getattr(args, "yes", False): + if not PR.ask_yes_no(f" Upgrade now via {plan.manager}?", default=True, + summary=f"Runs: {plan.display()}"): + _p(f" skipped. Upgrade anytime with: {plan.display()}") + return 0 + + _p(f"\n upgrading via {plan.manager}…\n") + ok, detail = updater.run_upgrade(plan) + if not ok: + _p(f"\n{PRODUCT}: upgrade failed ({detail}). You can run it manually:") + _p(f" {plan.display()}") + return 1 + + # importlib.metadata is cached in this process; read the truth from a fresh one. + # Only claim a confirmed version when we actually read one — a non-zero pip + # exit isn't proof komi-learn reached `latest` (pip can no-op or install a + # pinned older version), so never substitute `latest` and call it confirmed. + new = updater.installed_version_via_subprocess() + if new is None: + _p(f"\n{PRODUCT}: upgrade command finished, but I couldn't confirm the " + "installed version.") + _p(" Check it with: komi-learn update --check") + else: + _p(f"\n{PRODUCT}: upgraded {current} → {new}.") + if updater.is_newer(latest, new): + _p(f" note: PyPI shows {latest} but the install reports {new} — " + "you may be in a different environment than expected.") + _p(" If this release added hook events, refresh them with: komi-learn install") + return 0 + + def cmd_queue(args) -> int: """Inspect + act on the global-contribution review queue (the human gate). @@ -471,6 +538,13 @@ def build_parser() -> argparse.ArgumentParser: pl = sub.add_parser("login", help="log in for free OAuth distillation (claude CLI)") pl.set_defaults(func=cmd_login) + pup = sub.add_parser("update", help="check PyPI and upgrade komi-learn to the latest version") + pup.add_argument("--check", action="store_true", + help="only report whether an update is available; don't upgrade") + pup.add_argument("--yes", "-y", action="store_true", + help="upgrade without the confirmation prompt") + pup.set_defaults(func=cmd_update) + pc = sub.add_parser("curate", help="consolidate the learning library now (normally ~weekly)") pc.add_argument("--dry-run", action="store_true", help="preview changes without applying") pc.add_argument("--no-llm", action="store_true", help="prune only; don't merge clusters") diff --git a/komi/updater.py b/komi/updater.py new file mode 100644 index 0000000..7e71a38 --- /dev/null +++ b/komi/updater.py @@ -0,0 +1,271 @@ +"""komi-learn — self-update: check PyPI for a newer release and upgrade in place. + +`komi-learn update` should Just Work regardless of how the user installed us. The +hard part isn't the PyPI check — it's upgrading *the right environment*. A blind +`pip install -U` can hit the wrong interpreter, fight a pipx-managed venv, or need +permissions it doesn't have. So we detect the install method first and run the +command that matches it (pip-into-this-interpreter, or `pipx upgrade`). If we +genuinely can't tell, we print the command instead of guessing and breaking the +user's environment. + +Everything here is best-effort and network-failure-safe: a flaky PyPI lookup never +raises out of `check_latest`, it just returns ``None``. +""" + +from __future__ import annotations + +import json +import os +import re +import shutil +import subprocess +import sys +import urllib.request +from dataclasses import dataclass +from typing import Optional + +DIST_NAME = "komi-learn" +PYPI_JSON_URL = f"https://pypi.org/pypi/{DIST_NAME}/json" +_PYPI_TIMEOUT = 8 # seconds — a stuck lookup must not hang the CLI +_MAX_PYPI_BYTES = 5 * 1024 * 1024 # cap the response read (one project's JSON << this) + + +class _NoRedirect(urllib.request.HTTPRedirectHandler): + """A redirect handler that refuses every redirect (returns None).""" + + def redirect_request(self, *args, **kwargs): # noqa: D401 + return None + + +# Module-level opener so we don't rebuild it per call. Refuses 3xx redirects. +_NO_REDIRECT_OPENER = urllib.request.build_opener(_NoRedirect) + + +# ── version comparison ─────────────────────────────────────────────────────── +# +# A self-contained PEP 440-lite comparator. We deliberately do NOT depend on +# `packaging`: the engine's ethos is zero required deps, and "is 0.4.0 newer than +# 0.3.0" doesn't justify a runtime dependency that may or may not be present +# (depending on whether the lib happened to compute the answer differently per +# environment). This handles the version shapes komi-learn actually ships — +# release tuples, pre-releases (a/b/rc), post/dev, and a leading epoch — with one +# code path, so the answer never varies by environment. + +# pre-release phase ranks: anything pre sorts BEFORE the plain release; post AFTER. +_PRE_RANK = {"a": 0, "alpha": 0, "b": 1, "beta": 1, "rc": 2, "c": 2, "pre": 2, + "preview": 2} +_FINAL = 3 # a plain release (no pre/post/dev) sits between pre and post +_POST = 4 +# .devN sorts before everything at the same release (earliest), so it gets a phase +# below the lowest pre-rank. +_DEV = -1 + +_RELEASE_RE = re.compile(r"^(?:(\d+)!)?(\d+(?:\.\d+)*)(.*)$") +_PRE_RE = re.compile(r"[._-]?(a|b|c|rc|alpha|beta|pre|preview)[._-]?(\d*)") +_POST_RE = re.compile(r"[._-]?(?:post|rev|r)[._-]?(\d*)|-(\d+)") +_DEV_RE = re.compile(r"[._-]?dev[._-]?(\d*)") + + +def _norm_release(rel: str) -> tuple: + """Numeric release tuple with trailing zeros stripped so 1.0 == 1.0.0.""" + parts = [int(p) for p in rel.split(".")] + while len(parts) > 1 and parts[-1] == 0: + parts.pop() + return tuple(parts) + + +def _version_key(v: str): + """Map a version string to a tuple that sorts per PEP 440 (for the shapes we + ship). Unparseable input sorts lowest so it never spuriously beats a real + release. The key shape is: + (epoch, release_tuple, phase, phase_num, dev_num) + where ``phase`` orders dev < pre(a tuple: + """The numeric release tuple (trailing zeros stripped), e.g. "0.4.0rc1"->(0,4). + For full ordering (incl. pre/post/epoch) use :func:`is_newer`.""" + return _version_key(v)[1] or (0,) + + +def is_newer(latest: str, current: str) -> bool: + """True if ``latest`` is a strictly newer version than ``current`` (PEP 440-lite).""" + return _version_key(latest) > _version_key(current) + + +# ── PyPI lookup ────────────────────────────────────────────────────────────── + +def check_latest(*, timeout: int = _PYPI_TIMEOUT) -> Optional[str]: + """Return the latest version string on PyPI, or ``None`` if it can't be + determined (offline, PyPI down, malformed payload). Never raises.""" + try: + req = urllib.request.Request( + PYPI_JSON_URL, headers={"Accept": "application/json", + "User-Agent": f"{DIST_NAME}-updater"}) + # Refuse redirects: the hardcoded https literal only guarantees the FIRST + # hop. urllib's default handler would follow a 3xx to any host/scheme, + # including http — a MITM could downgrade the lookup and feed us a fake + # "newer" version to coerce an upgrade. PyPI's JSON API doesn't redirect + # cross-scheme here, so refusing is safe. + with _NO_REDIRECT_OPENER.open(req, timeout=timeout) as resp: # nosec B310 - https, no-redirect + if not (resp.geturl() or "").lower().startswith("https://"): + return None + # Bound the read: a hostile endpoint must not be able to stream a huge + # body and OOM the CLI. One project's JSON is far under this cap. + raw = resp.read(_MAX_PYPI_BYTES + 1) + if len(raw) > _MAX_PYPI_BYTES: + return None + data = json.loads(raw.decode("utf-8")) + ver = (data.get("info") or {}).get("version") + return ver or None + except Exception: + return None + + +# ── install-method detection ───────────────────────────────────────────────── + +def _is_pipx() -> bool: + """Are we running from a pipx-managed venv? pipx installs each app into + ``.../pipx/venvs//``. + + We key off the *interpreter prefix* — the one signal an attacker can't set + just by exporting an env var. A bare ``PIPX_HOME`` is NOT sufficient on its + own: that would let a hostile environment flip a plain pip install into the + pipx branch (and then a planted ``pipx`` on PATH would run). The env var only + *corroborates* a prefix that already lives under it. + """ + prefix = (sys.prefix or "").replace("\\", "/") + low = prefix.lower() + if "/pipx/venvs/" in low or "/pipx/shared" in low: + return True + # If PIPX_HOME is set AND our interpreter actually lives under it, trust it. + pipx_home = (os.environ.get("PIPX_HOME") or "").replace("\\", "/") + if pipx_home and prefix.startswith(pipx_home): + return True + return False + + +def _pip_available() -> bool: + try: + import pip # noqa: F401 + return True + except Exception: + return False + + +@dataclass +class UpgradePlan: + """How we intend to upgrade. ``cmd`` is the argv to run; ``manager`` is for + display; ``runnable`` is False when we couldn't determine a safe command (then + ``cmd`` is the human-facing suggestion to print, not to execute).""" + manager: str + cmd: list + runnable: bool + + def display(self) -> str: + return " ".join(self.cmd) + + +def plan_upgrade() -> UpgradePlan: + """Decide the upgrade command for *this* environment.""" + if _is_pipx(): + # pipx manages its own venv; pip -U inside it is the wrong tool. Resolve + # pipx to an ABSOLUTE path (never an unqualified "pipx" off PATH — that + # would let a planted binary earlier in PATH run during `update`). If we + # can't find it, refuse and print the command rather than guess. + pipx_path = shutil.which("pipx") + if not pipx_path: + return UpgradePlan("pipx", ["pipx", "upgrade", DIST_NAME], runnable=False) + return UpgradePlan("pipx", [pipx_path, "upgrade", DIST_NAME], runnable=True) + if _pip_available(): + # Upgrade into the very interpreter that's running komi-learn — this is the + # one whose `import komi` the hooks use. Mirrors model_install.py. + return UpgradePlan( + "pip", + [sys.executable, "-m", "pip", "install", "--upgrade", DIST_NAME], + runnable=True, + ) + # Couldn't find a package manager we trust to drive — hand the user a command + # rather than risk corrupting a standalone/frozen install. + return UpgradePlan("pip", ["pip", "install", "--upgrade", DIST_NAME], runnable=False) + + +# ── upgrade execution + re-verify ──────────────────────────────────────────── + +def installed_version_via_subprocess() -> Optional[str]: + """Read komi-learn's version in a *fresh* interpreter. + + importlib.metadata caches distribution info for the life of a process, so after + an in-process pip upgrade the running interpreter still reports the OLD version. + Shelling out to a clean python gets the truth post-upgrade. + """ + # Pass the dist name as argv (sys.argv[1]) rather than interpolating it into + # the code string — keeps the snippet free of string-building even though + # DIST_NAME is a trusted literal. + code = ( + "import sys\n" + "try:\n" + " from importlib.metadata import version\n" + " sys.stdout.write(version(sys.argv[1]))\n" + "except Exception:\n" + " pass\n" + ) + try: + r = subprocess.run([sys.executable, "-c", code, DIST_NAME], + capture_output=True, text=True, timeout=30) + out = (r.stdout or "").strip() + return out or None + except Exception: + return None + + +def run_upgrade(plan: UpgradePlan, *, timeout: int = 1200) -> tuple[bool, str]: + """Execute the upgrade command. Returns (ok, detail). Output streams to the + user's terminal so they see pip's progress (and any resolver errors).""" + if not plan.runnable: + return False, "no runnable upgrade command for this environment" + try: + r = subprocess.run(plan.cmd, timeout=timeout) + except FileNotFoundError: + return False, f"`{plan.cmd[0]}` not found on PATH" + except subprocess.TimeoutExpired: + return False, "upgrade timed out" + except Exception as e: + return False, f"upgrade failed to launch: {e}" + if r.returncode != 0: + return False, f"upgrade exited {r.returncode}" + return True, "upgraded" + + +__all__ = [ + "DIST_NAME", "PYPI_JSON_URL", + "check_latest", "is_newer", "plan_upgrade", "UpgradePlan", + "run_upgrade", "installed_version_via_subprocess", +] diff --git a/pyproject.toml b/pyproject.toml index cbb08b9..86f7d4d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,9 @@ build-backend = "setuptools.build_meta" [project] name = "komi-learn" -version = "0.3.0" +# Version is read from komi/__init__.py (__version__) — the single source of +# truth. Bump it there; see [tool.setuptools.dynamic] below. +dynamic = ["version"] description = "A continuous, zero-friction learning layer for AI agents — learns about you and improves its own skills across every session, with an anonymized, provenance-verified global knowledge pool." readme = "README.md" requires-python = ">=3.10" @@ -52,6 +54,11 @@ dev = ["pytest>=8"] [project.scripts] komi-learn = "komi.cli:main" +[tool.setuptools.dynamic] +# Read the version from the package's __version__ at build time (single source +# of truth in komi/__init__.py). +version = { attr = "komi.__version__" } + [tool.setuptools.packages.find] include = ["komi*"] diff --git a/tests/test_update.py b/tests/test_update.py new file mode 100644 index 0000000..00d2d91 --- /dev/null +++ b/tests/test_update.py @@ -0,0 +1,362 @@ +"""`komi-learn update`: self-update against PyPI. + +Covers the two halves: the pure logic in komi.updater (version compare, PyPI +lookup, install-method detection, upgrade execution) and the CLI command's +routing (up-to-date / newer / --check / undetectable / failure). All network and +subprocess calls are mocked — these tests never touch PyPI or run pip. +""" + +import io +import json +from unittest import mock + +import pytest + +from komi import updater as U +from komi import cli + + +# ── version comparison ─────────────────────────────────────────────────────── + +@pytest.mark.parametrize("latest,current,expected", [ + # basic release ordering + ("0.4.0", "0.3.0", True), + ("0.3.1", "0.3.0", True), + ("0.3.0", "0.3.0", False), # equal is not newer + ("0.2.0", "0.3.0", False), # older + ("0.10.0", "0.9.0", True), # numeric, not lexicographic + ("1.0.0", "0.99.0", True), + # zero-padding equivalence: 1.0 == 1.0.0 (must not be a phantom upgrade) + ("1.0.0", "1.0", False), + ("1.0", "1.0.0", False), + # pre-releases sort BEFORE the final release + ("0.4.0", "0.4.0rc1", True), # final newer than its rc + ("0.4.0rc1", "0.4.0", False), # rc not newer than final + ("0.4.0rc2", "0.4.0rc1", True), + ("0.4.0b1", "0.4.0a1", True), # b > a + ("0.4.0a1", "0.4.0b1", False), + # dev sorts before pre; post sorts after final + ("0.4.0rc1", "0.4.0.dev1", True), + ("0.4.0.dev1", "0.4.0rc1", False), + ("1.0.0.post1", "1.0.0", True), + # epoch dominates the release number + ("1!2.0", "2.0", True), + ("2.0", "1!2.0", False), + # tolerate a leading v/V tag + ("v0.4.0", "0.3.0", True), + ("0.4.0", "v0.4.0", False), + # empty / junk never spuriously beats a real release + ("", "0.3.0", False), + ("0.3.0", "", True), +]) +def test_is_newer(latest, current, expected): + assert U.is_newer(latest, current) is expected + + +def test_parse_version_release_tuple(): + # _parse_version returns the numeric release tuple with trailing zeros stripped + assert U._parse_version("0.4.0rc1") == (0, 4) # pre-release suffix dropped + assert U._parse_version("0.4") == (0, 4) + assert U._parse_version("1.2.3") == (1, 2, 3) + assert U._parse_version("v2.0.0") == (2,) # leading v + trailing zeros + assert U._parse_version("") == (0,) + + +def test_version_ordering_is_total_and_self_consistent(): + # an ascending chain must be strictly increasing under is_newer, and each step + # must be antisymmetric (a>b implies not b>a) + chain = ["0.4.0.dev1", "0.4.0a1", "0.4.0b1", "0.4.0rc1", "0.4.0", + "0.4.0.post1", "0.4.1", "0.5.0", "1.0.0", "1!0.0.1"] + for lo, hi in zip(chain, chain[1:]): + assert U.is_newer(hi, lo) is True, f"{hi} should be newer than {lo}" + assert U.is_newer(lo, hi) is False, f"{lo} should NOT be newer than {hi}" + + +# ── PyPI lookup ────────────────────────────────────────────────────────────── + +class _FakeResp: + """Stands in for the urllib response: https final URL + a bounded read().""" + def __init__(self, payload): self._b = json.dumps(payload).encode() + def geturl(self): return U.PYPI_JSON_URL + def read(self, n=-1): return self._b[:n] if (n and n > 0) else self._b + def __enter__(self): return self + def __exit__(self, *a): return False + + +def test_check_latest_parses_version(): + with mock.patch.object(U._NO_REDIRECT_OPENER, "open", + return_value=_FakeResp({"info": {"version": "0.5.0"}})): + assert U.check_latest() == "0.5.0" + + +def test_check_latest_network_failure_returns_none(): + # offline / PyPI down must be non-fatal — returns None, never raises + with mock.patch.object(U._NO_REDIRECT_OPENER, "open", side_effect=OSError("no network")): + assert U.check_latest() is None + + +def test_check_latest_malformed_payload_returns_none(): + with mock.patch.object(U._NO_REDIRECT_OPENER, "open", + return_value=_FakeResp({"nope": 1})): + assert U.check_latest() is None + + +# ── install-method detection ───────────────────────────────────────────────── + +def test_plan_pipx_when_prefix_looks_like_pipx(monkeypatch): + monkeypatch.delenv("PIPX_HOME", raising=False) + monkeypatch.delenv("PIPX_BIN_DIR", raising=False) + monkeypatch.setattr(U.sys, "prefix", "/home/u/.local/pipx/venvs/komi-learn") + monkeypatch.setattr(U.shutil, "which", lambda name: "/usr/local/bin/pipx") + plan = U.plan_upgrade() + assert plan.manager == "pipx" + assert plan.runnable is True + assert plan.cmd == ["/usr/local/bin/pipx", "upgrade", "komi-learn"] + + +def test_plan_pip_uses_running_interpreter(monkeypatch): + monkeypatch.delenv("PIPX_HOME", raising=False) + monkeypatch.delenv("PIPX_BIN_DIR", raising=False) + monkeypatch.setattr(U.sys, "prefix", "/usr") + monkeypatch.setattr(U.sys, "executable", "/usr/bin/python3") + monkeypatch.setattr(U, "_pip_available", lambda: True) + plan = U.plan_upgrade() + assert plan.manager == "pip" + assert plan.runnable is True + # critical: upgrade THIS interpreter, the one the hooks import + assert plan.cmd[:4] == ["/usr/bin/python3", "-m", "pip", "install"] + assert "--upgrade" in plan.cmd and "komi-learn" in plan.cmd + + +def test_plan_falls_back_to_unrunnable_when_no_pip(monkeypatch): + monkeypatch.delenv("PIPX_HOME", raising=False) + monkeypatch.delenv("PIPX_BIN_DIR", raising=False) + monkeypatch.setattr(U.sys, "prefix", "/opt/frozen") + monkeypatch.setattr(U, "_pip_available", lambda: False) + plan = U.plan_upgrade() + assert plan.runnable is False # never guess a command we can't run safely + + +# ── upgrade execution ──────────────────────────────────────────────────────── + +def test_run_upgrade_success(): + plan = U.UpgradePlan("pip", ["python", "-m", "pip", "install", "-U", "komi-learn"], True) + with mock.patch("subprocess.run", return_value=mock.Mock(returncode=0)): + ok, detail = U.run_upgrade(plan) + assert ok is True + + +def test_run_upgrade_nonzero_exit(): + plan = U.UpgradePlan("pip", ["python", "-m", "pip", "install", "-U", "komi-learn"], True) + with mock.patch("subprocess.run", return_value=mock.Mock(returncode=1)): + ok, detail = U.run_upgrade(plan) + assert ok is False and "exited 1" in detail + + +def test_run_upgrade_unrunnable_plan_refuses(): + plan = U.UpgradePlan("pip", ["pip", "install", "-U", "komi-learn"], runnable=False) + ok, detail = U.run_upgrade(plan) + assert ok is False and "no runnable" in detail + + +def test_run_upgrade_missing_binary(): + plan = U.UpgradePlan("pipx", ["pipx", "upgrade", "komi-learn"], True) + with mock.patch("subprocess.run", side_effect=FileNotFoundError()): + ok, detail = U.run_upgrade(plan) + assert ok is False and "not found" in detail + + +# ── CLI command routing ────────────────────────────────────────────────────── + +def _args(**kw): + ns = mock.Mock() + ns.check = kw.get("check", False) + ns.yes = kw.get("yes", False) + return ns + + +def _capture(fn, *a): + out = io.StringIO() + with mock.patch("sys.stdout", out): + rc = fn(*a) + return rc, out.getvalue() + + +def test_cmd_update_already_latest(monkeypatch): + monkeypatch.setattr("komi.__version__", "0.3.0", raising=False) + with mock.patch.object(U, "check_latest", return_value="0.3.0"): + rc, out = _capture(cli.cmd_update, _args()) + assert rc == 0 + assert "latest version" in out + + +def test_cmd_update_offline(monkeypatch): + with mock.patch.object(U, "check_latest", return_value=None): + rc, out = _capture(cli.cmd_update, _args()) + assert rc == 1 + assert "couldn't reach PyPI" in out + + +def test_cmd_update_check_only_does_not_upgrade(monkeypatch): + monkeypatch.setattr("komi.__version__", "0.3.0", raising=False) + good_plan = U.UpgradePlan("pip", ["py", "-m", "pip", "install", "-U", "komi-learn"], True) + with mock.patch.object(U, "check_latest", return_value="0.9.0"), \ + mock.patch.object(U, "plan_upgrade", return_value=good_plan), \ + mock.patch.object(U, "run_upgrade") as run: + rc, out = _capture(cli.cmd_update, _args(check=True)) + assert rc == 0 + run.assert_not_called() # --check must never mutate the environment + # must show the CHECK-specific "to upgrade" line + the actual command (not just + # the generic "newer version available" line that every non-latest path prints) + assert "to upgrade:" in out + assert "py -m pip install -U komi-learn" in out + + +def test_cmd_update_undetectable_prints_command(monkeypatch): + monkeypatch.setattr("komi.__version__", "0.3.0", raising=False) + bad_plan = U.UpgradePlan("pip", ["pip", "install", "-U", "komi-learn"], runnable=False) + with mock.patch.object(U, "check_latest", return_value="0.9.0"), \ + mock.patch.object(U, "plan_upgrade", return_value=bad_plan), \ + mock.patch.object(U, "run_upgrade") as run: + rc, out = _capture(cli.cmd_update, _args(yes=True)) + assert rc == 1 + run.assert_not_called() # unrunnable → instruct, don't execute + assert "pip install -U komi-learn" in out + + +def test_cmd_update_runs_upgrade_and_reports_new_version(monkeypatch): + monkeypatch.setattr("komi.__version__", "0.3.0", raising=False) + good_plan = U.UpgradePlan("pip", ["py", "-m", "pip", "install", "-U", "komi-learn"], True) + with mock.patch.object(U, "check_latest", return_value="0.9.0"), \ + mock.patch.object(U, "plan_upgrade", return_value=good_plan), \ + mock.patch.object(U, "run_upgrade", return_value=(True, "upgraded")) as run, \ + mock.patch.object(U, "installed_version_via_subprocess", return_value="0.9.0"): + rc, out = _capture(cli.cmd_update, _args(yes=True)) + assert rc == 0 + run.assert_called_once() + assert "0.9.0" in out + + +def test_cmd_update_reports_upgrade_failure(monkeypatch): + monkeypatch.setattr("komi.__version__", "0.3.0", raising=False) + good_plan = U.UpgradePlan("pip", ["py", "-m", "pip", "install", "-U", "komi-learn"], True) + with mock.patch.object(U, "check_latest", return_value="0.9.0"), \ + mock.patch.object(U, "plan_upgrade", return_value=good_plan), \ + mock.patch.object(U, "run_upgrade", return_value=(False, "upgrade exited 1")): + rc, out = _capture(cli.cmd_update, _args(yes=True)) + assert rc == 1 + assert "upgrade failed" in out + + +def test_cmd_update_does_not_claim_unconfirmed_version(monkeypatch): + """If the post-upgrade re-check can't read a version (returns None), we must + NOT print 'upgraded 0.3.0 → 0.9.0' as if confirmed — pip exiting 0 isn't proof + the new version actually landed.""" + monkeypatch.setattr("komi.__version__", "0.3.0", raising=False) + good_plan = U.UpgradePlan("pip", ["py", "-m", "pip", "install", "-U", "komi-learn"], True) + with mock.patch.object(U, "check_latest", return_value="0.9.0"), \ + mock.patch.object(U, "plan_upgrade", return_value=good_plan), \ + mock.patch.object(U, "run_upgrade", return_value=(True, "upgraded")), \ + mock.patch.object(U, "installed_version_via_subprocess", return_value=None): + rc, out = _capture(cli.cmd_update, _args(yes=True)) + assert rc == 0 + assert "couldn't confirm" in out + # the generic "a newer version is available — 0.3.0 → 0.9.0" line is fine; what + # must NOT appear is the CONFIRMED claim "upgraded ... → 0.9.0" + assert "upgraded" not in out + + +# ── security: PyPI fetch hardening ──────────────────────────────────────────── + +def test_check_latest_refuses_redirect(): + """The lookup must not follow a 3xx — a redirect to http/another host could + feed a spoofed 'newer' version. The no-redirect opener raises, we return None.""" + import urllib.error + err = urllib.error.HTTPError(U.PYPI_JSON_URL, 302, "Found", {}, None) + with mock.patch.object(U._NO_REDIRECT_OPENER, "open", side_effect=err): + assert U.check_latest() is None + + +def test_check_latest_rejects_non_https_final_url(): + class _Resp: + def geturl(self): return "http://evil.example/komi-learn/json" + def read(self, *a): return b'{"info":{"version":"999.0.0"}}' + def __enter__(self): return self + def __exit__(self, *a): return False + with mock.patch.object(U._NO_REDIRECT_OPENER, "open", return_value=_Resp()): + assert U.check_latest() is None # downgraded scheme → refuse the answer + + +def test_check_latest_bounds_oversized_body(): + big = b"x" * (U._MAX_PYPI_BYTES + 1000) + class _Resp: + def geturl(self): return U.PYPI_JSON_URL + def read(self, n=-1): return big[:n] if n and n > 0 else big + def __enter__(self): return self + def __exit__(self, *a): return False + with mock.patch.object(U._NO_REDIRECT_OPENER, "open", return_value=_Resp()): + # reads at most _MAX_PYPI_BYTES+1, sees it's over the cap → None (no OOM, no parse) + assert U.check_latest() is None + + +# ── security: pipx upgrade path ─────────────────────────────────────────────── + +def test_pipx_plan_uses_absolute_path(monkeypatch): + monkeypatch.setattr(U, "_is_pipx", lambda: True) + monkeypatch.setattr(U.shutil, "which", lambda name: "/usr/local/bin/pipx") + plan = U.plan_upgrade() + assert plan.manager == "pipx" + assert plan.runnable is True + assert plan.cmd[0] == "/usr/local/bin/pipx" # absolute, never bare "pipx" off PATH + assert plan.cmd[0] != "pipx" + + +def test_pipx_plan_refuses_when_pipx_not_found(monkeypatch): + monkeypatch.setattr(U, "_is_pipx", lambda: True) + monkeypatch.setattr(U.shutil, "which", lambda name: None) + plan = U.plan_upgrade() + assert plan.runnable is False # can't resolve pipx → don't guess + + +def test_is_pipx_bare_env_var_is_not_enough(monkeypatch): + """A hostile environment setting PIPX_HOME alone must NOT flip a pip install + into the pipx branch (which could then run a planted `pipx`).""" + monkeypatch.setenv("PIPX_HOME", "/tmp/attacker") + monkeypatch.setattr(U.sys, "prefix", "/usr") # interpreter NOT under PIPX_HOME + assert U._is_pipx() is False + + +def test_is_pipx_true_when_prefix_under_pipx_home(monkeypatch): + monkeypatch.setenv("PIPX_HOME", "/home/u/.local/pipx") + monkeypatch.setattr(U.sys, "prefix", "/home/u/.local/pipx/venvs/komi-learn") + assert U._is_pipx() is True + + +def test_is_pipx_true_from_prefix_substring(monkeypatch): + monkeypatch.delenv("PIPX_HOME", raising=False) + monkeypatch.setattr(U.sys, "prefix", "/home/u/.local/pipx/venvs/komi-learn") + assert U._is_pipx() is True + + +# ── version: single source of truth ─────────────────────────────────────────── + +def test_package_version_is_single_source_of_truth(): + """Regression: __version__ was hardcoded 0.1.0 while pyproject was 0.3.0. + Now pyproject reads komi.__version__ dynamically, so they can't diverge.""" + import komi + assert komi.__version__ != "0.1.0" + # at least 0.3.0 (use the version comparator, not raw tuples — 0.3.0 -> (0,3)) + assert not U.is_newer("0.3.0", komi.__version__) # 0.3.0 is not newer than us + + +def test_pyproject_uses_dynamic_version(): + """Guard the dynamic-version wiring so nobody re-introduces a static literal + in pyproject that could drift from komi.__version__.""" + from pathlib import Path + root = Path(__file__).resolve().parents[1] + text = (root / "pyproject.toml").read_text(encoding="utf-8") + assert 'dynamic = ["version"]' in text + assert 'attr = "komi.__version__"' in text + # there must be no static `version = "x.y.z"` under [project] + import re as _re + assert not _re.search(r'(?m)^\s*version\s*=\s*"\d', text)