"""Update + uninstall paths from :mod:`lib.extensions`.

``subprocess.run`` is mocked; a small fake manifest is written by

the side effect so the real ``Manifest`` machinery exercises the

validation step.

"""

from __future__ import annotations

import json

import os

import subprocess

import sys

import tempfile

import textwrap

import unittest

from pathlib import Path

from unittest import mock

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from lib import config as cfg # noqa: E402

from lib import extensions # noqa: E402

_OLD_HEAD = "1" * 40

def _manifest(version: str, *, min_core: str = "0.7.1",

state_paths: list[str] | None = None) -> str:

payload = {

"name": "agent",

"version": version,

"module": "agent",

"min_tmux_browse": min_core,

"routes_entry": "server.routes:register",

}

if state_paths is not None:

payload["state_paths"] = state_paths

return json.dumps(payload)

class _IsolatedExt:

def setUp(self):

self._tmp = tempfile.TemporaryDirectory()

self._state = tempfile.TemporaryDirectory()

root = Path(self._tmp.name)

self._project = root

self._ext_root = root / "extensions"

self._ext_root.mkdir()

self._enabled = Path(self._state.name) / "extensions.json"

self._patches = [

mock.patch.object(cfg, "PROJECT_DIR", root),

mock.patch.object(cfg, "STATE_DIR", Path(self._state.name)),

mock.patch.object(extensions, "EXTENSIONS_ROOT", self._ext_root),

mock.patch.object(extensions, "ENABLED_FILE", self._enabled),

]

for p in self._patches:

p.start()

def tearDown(self):

for p in self._patches:

p.stop()

self._tmp.cleanup()

self._state.cleanup()

def _install_fake(self, version: str, *, submodule: bool = False,

state_paths: list[str] | None = None) -> Path:

target = self._ext_root / "agent"

target.mkdir(parents=True, exist_ok=True)

(target / "manifest.json").write_text(

_manifest(version, state_paths=state_paths))

if submodule:

(self._project / ".gitmodules").write_text(textwrap.dedent("""\

[submodule "extensions/agent"]

\tpath = extensions/agent

\turl = https://example.com/agent.git

\tbranch = main

"""))

return target

class UpdateTests(_IsolatedExt, unittest.TestCase):

def test_missing_install_raises_missing_stage(self):

with self.assertRaises(extensions.UpdateError) as ctx:

extensions.update("agent", core_version="0.7.1")

self.assertEqual(ctx.exception.stage, "missing")

def test_unknown_name_raises_unknown_stage(self):

with self.assertRaises(extensions.UpdateError) as ctx:

extensions.update("nope", core_version="0.7.1")

self.assertEqual(ctx.exception.stage, "unknown")

def test_clone_path_fetches_then_checks_out_ref(self):

path = self._install_fake("0.7.1")

calls: list[list[str]] = []

def run(args, **_kwargs):

calls.append(list(args))

if args[:2] == ["git", "rev-parse"]:

return subprocess.CompletedProcess(args, 0, _OLD_HEAD + "\n", "")

if args[:2] == ["git", "fetch"]:

return subprocess.CompletedProcess(args, 0, "", "")

if args[:2] == ["git", "checkout"]:

# Simulate the checkout by bumping the installed manifest.

(path / "manifest.json").write_text(_manifest("0.7.2"))

return subprocess.CompletedProcess(args, 0, "", "")

raise AssertionError(f"unexpected call: {args!r}")

with mock.patch("subprocess.run", side_effect=run):

result = extensions.update("agent", core_version="0.7.2")

self.assertTrue(result.changed)

self.assertEqual(result.from_version, "0.7.1")

self.assertEqual(result.to_version, "0.7.2")

self.assertEqual(result.via, "clone")

self.assertEqual(calls[0][:2], ["git", "rev-parse"])

self.assertEqual(

calls[1],

["git", "fetch", "--depth", "50", "origin", "v0.7.3-agent"],

)

self.assertEqual(

calls[2], ["git", "checkout", "--detach", "FETCH_HEAD"])

def test_fetch_failure_raises_fetch_stage(self):

self._install_fake("0.7.1")

def run(args, **_kwargs):

if args[:2] == ["git", "rev-parse"]:

return subprocess.CompletedProcess(args, 0, _OLD_HEAD + "\n", "")

if args[:2] == ["git", "fetch"]:

return subprocess.CompletedProcess(

args, 128, "", "fatal: unable to access")

raise AssertionError("fetch should have short-circuited")

with mock.patch("subprocess.run", side_effect=run):

with self.assertRaises(extensions.UpdateError) as ctx:

extensions.update("agent", core_version="0.7.1")

self.assertEqual(ctx.exception.stage, "fetch")

def test_submodule_path_calls_update_remote(self):

self._install_fake("0.7.1", submodule=True)

def result(args, **_kwargs):

stdout = _OLD_HEAD + "\n" if args[:2] == ["git", "rev-parse"] else ""

return subprocess.CompletedProcess(args, 0, stdout, "")

with mock.patch("subprocess.run", side_effect=result) as run:

result = extensions.update("agent", core_version="0.7.1")

self.assertEqual(result.via, "submodule")

args, _ = run.call_args_list[1]

self.assertEqual(args[0][:4],

["git", "submodule", "update", "--remote"])

def test_unchanged_version_reports_changed_false(self):

path = self._install_fake("0.7.1")

def run(args, **_kwargs):

# Leave the manifest at 0.7.1 — fetch + checkout succeed but

# no new version landed.

stdout = _OLD_HEAD + "\n" if args[:2] == ["git", "rev-parse"] else ""

return subprocess.CompletedProcess(args, 0, stdout, "")

with mock.patch("subprocess.run", side_effect=run):

result = extensions.update("agent", core_version="0.7.1")

self.assertFalse(result.changed)

self.assertEqual(result.from_version, "0.7.1")

self.assertEqual(result.to_version, "0.7.1")

def test_too_new_manifest_raises_validate_stage(self):

path = self._install_fake("0.7.1")

def run(args, **_kwargs):

if args[:2] == ["git", "rev-parse"]:

return subprocess.CompletedProcess(args, 0, _OLD_HEAD + "\n", "")

if args == ["git", "checkout", "--detach", "FETCH_HEAD"]:

(path / "manifest.json").write_text(

_manifest("99.0.0", min_core="99.0.0"))

elif args == ["git", "checkout", "--detach", _OLD_HEAD]:

(path / "manifest.json").write_text(_manifest("0.7.1"))

return subprocess.CompletedProcess(args, 0, "", "")

with mock.patch("subprocess.run", side_effect=run):

with self.assertRaises(extensions.UpdateError) as ctx:

extensions.update("agent", core_version="0.7.1")

self.assertEqual(ctx.exception.stage, "validate")

restored = json.loads((path / "manifest.json").read_text())

self.assertEqual(restored["version"], "0.7.1")

def test_checkout_failure_attempts_rollback(self):

self._install_fake("0.7.1")

calls: list[list[str]] = []

def run(args, **_kwargs):

calls.append(list(args))

if args[:2] == ["git", "rev-parse"]:

return subprocess.CompletedProcess(args, 0, _OLD_HEAD + "\n", "")

if args == ["git", "checkout", "--detach", "FETCH_HEAD"]:

return subprocess.CompletedProcess(args, 1, "", "checkout failed")

return subprocess.CompletedProcess(args, 0, "", "")

with mock.patch("subprocess.run", side_effect=run):

with self.assertRaises(extensions.UpdateError) as ctx:

extensions.update("agent", core_version="0.7.1")

self.assertEqual(ctx.exception.stage, "checkout")

self.assertIn(

["git", "checkout", "--detach", _OLD_HEAD], calls)

class UninstallTests(_IsolatedExt, unittest.TestCase):

def test_clone_path_removes_directory_keeps_state(self):

target = self._install_fake(

"0.7.1",

state_paths=["agents.json", "agent-logs/"])

# Stand up the state files that should NOT be touched.

state = Path(self._state.name)

(state / "agents.json").write_text("{}")

(state / "agent-logs").mkdir()

(state / "agent-logs" / "gpt.jsonl").write_text("{}\n")

# Pre-seed enabled=true so we can verify the flip.

extensions.enable("agent")

result = extensions.uninstall("agent")

self.assertFalse(target.exists())

self.assertEqual(result["via"], "clone")

self.assertEqual(result["state_removed"], [])

# extensions.json entry flipped off + uninstall timestamp recorded.

entry = extensions._read_enabled()["agent"]

self.assertFalse(entry["enabled"])

self.assertIn("uninstalled_ts", entry)

# State files untouched.

self.assertTrue((state / "agents.json").is_file())

self.assertTrue((state / "agent-logs" / "gpt.jsonl").is_file())

def test_clone_path_remove_state_deletes_declared_paths(self):

self._install_fake(

"0.7.1",

state_paths=["agents.json", "agent-logs/"])

state = Path(self._state.name)

(state / "agents.json").write_text("{}")

(state / "agent-logs").mkdir()

(state / "agent-logs" / "gpt.jsonl").write_text("{}\n")

result = extensions.uninstall("agent", remove_state=True)

self.assertCountEqual(

result["state_removed"], ["agents.json", "agent-logs/"])

self.assertFalse((state / "agents.json").exists())

self.assertFalse((state / "agent-logs").exists())

def test_remove_state_reports_missing_paths(self):

self._install_fake(

"0.7.1",

state_paths=["agents.json", "absent.json"])

(Path(self._state.name) / "agents.json").write_text("{}")

result = extensions.uninstall("agent", remove_state=True)

self.assertEqual(result["state_removed"], ["agents.json"])

self.assertEqual(result["state_missing"], ["absent.json"])

def test_remove_state_refuses_to_delete_state_root(self):

# A manifest listing "." (or any path resolving back to STATE_DIR)

# must NOT wipe the whole state dir — other extensions' state, the

# port registry, dashboard config all live there.

self._install_fake("0.7.1", state_paths=["."])

state = Path(self._state.name)

(state / "ports.json").write_text("{}")

(state / "agent-logs").mkdir()

result = extensions.uninstall("agent", remove_state=True)

# "." is rejected by the safety rail, so nothing under the root dies.

self.assertEqual(result["state_removed"], [])

self.assertTrue(state.is_dir())

self.assertTrue((state / "ports.json").is_file())

self.assertTrue((state / "agent-logs").is_dir())

def test_remove_state_refuses_path_traversal(self):

# An escaping relative path is dropped, not followed.

self._install_fake("0.7.1", state_paths=["../escape.json"])

outside = Path(self._state.name).parent / "escape.json"

outside.write_text("secret")

try:

result = extensions.uninstall("agent", remove_state=True)

self.assertEqual(result["state_removed"], [])

self.assertTrue(outside.exists())

finally:

outside.unlink(missing_ok=True)

def test_uninstall_rejects_extension_name_path_traversal(self):

outside = self._project / "outside"

outside.mkdir()

(outside / "keep.txt").write_text("secret")

with self.assertRaises(extensions.InvalidExtensionName):

extensions.uninstall("../outside")

self.assertTrue((outside / "keep.txt").exists())

def test_submodule_path_calls_deinit_not_rmtree(self):

self._install_fake("0.7.1", submodule=True)

target = self._ext_root / "agent"

with mock.patch(

"subprocess.run",

return_value=subprocess.CompletedProcess(

["git"], 0, "", ""),

) as run:

result = extensions.uninstall("agent")

self.assertEqual(result["via"], "submodule")

args, _ = run.call_args

self.assertEqual(args[0][:5],

["git", "submodule", "deinit", "-f", "--"])

# The submodule tree is not our responsibility to rm — the

# operator may want `git submodule update --init` to bring it

# back later.

self.assertTrue(target.exists())

def test_uninstall_is_idempotent_on_missing_tree(self):

# Never installed at all — uninstall should still mark the

# entry disabled without raising.

result = extensions.uninstall("agent")

self.assertFalse(result["state_removed"])

entry = extensions._read_enabled()["agent"]

self.assertFalse(entry["enabled"])

class RecordErrorResilienceTests(_IsolatedExt, unittest.TestCase):

"""``record_error`` is called during startup load and per-request

post-processor handling with filesystem-/manifest-derived names, so it

must never raise on an odd name — otherwise one badly-named extension

dir takes down ``load_enabled`` (server startup) or 500s a request."""

def test_record_error_tolerates_non_basename(self):

# A name with a dot (e.g. a leftover ``agent.bak`` copy) is not a

# clean basename, but record_error only writes a JSON key.

extensions.record_error("agent.bak", "boom")

self.assertEqual(

extensions._read_enabled()["agent.bak"]["last_error"], "boom")

def test_record_error_ignores_empty_name(self):

extensions.record_error(" ", "boom")

self.assertEqual(extensions._read_enabled(), {})

def test_load_enabled_survives_oddly_named_extension(self):

# An *enabled* extension dir whose name isn't a clean basename and

# that fails to load must be recorded-and-skipped, not crash

# load_enabled via record_error.

ext = self._ext_root / "agent.bak"

ext.mkdir()

(ext / "manifest.json").write_text(json.dumps({

"name": "agent.bak",

"version": "0.0.1",

"module": "nope_missing_module",

"min_tmux_browse": "0.7.1",

# Import target doesn't exist → ExtensionLoadError at load time.

"routes_entry": "nope_missing_module:register",

}))

extensions._write_enabled({"agent.bak": {"enabled": True}})

# Must not raise.

extensions.load_enabled(core_version_override="0.7.1")

# The load failure was recorded under the raw name.

entry = extensions._read_enabled()["agent.bak"]

self.assertIsNotNone(entry.get("last_error"))

class ExtensionStateDurabilityTests(_IsolatedExt, unittest.TestCase):

def test_state_is_private_and_atomic_replace_failure_keeps_old_data(self):

extensions._write_enabled({"first": {"enabled": True}})

self.assertEqual(self._enabled.stat().st_mode & 0o777, 0o600)

with mock.patch.object(

extensions.os, "replace", side_effect=OSError("simulated crash"),

):

with self.assertRaises(OSError):

extensions._write_enabled({"second": {"enabled": True}})

self.assertEqual(

json.loads(self._enabled.read_text()),

{"first": {"enabled": True}},

)

leftovers = list(self._enabled.parent.glob(".extensions.json.*.tmp"))

self.assertEqual(leftovers, [])

def test_read_modify_write_keeps_parallel_extension_changes(self):

# Exercise separate interpreters, not just threads: dashboard HTTP,

# CLI and startup error recording can all mutate this file at once.

root = Path(__file__).resolve().parent.parent

programs = []

environment = dict(os.environ, HOME=self._state.name)

for index in range(12):

programs.append(subprocess.Popen(

[

sys.executable,

"-c",

"import sys; from pathlib import Path; "

"from lib import config, extensions; "

"path = Path(sys.argv[1]); "

"config.STATE_DIR = path.parent; "

"extensions.ENABLED_FILE = path; "

f"extensions.enable('parallel-{index}')",

str(self._enabled),

],

cwd=root,

env=environment,

stdout=subprocess.PIPE,

stderr=subprocess.PIPE,

text=True,

))

failures = []

for process in programs:

stdout, stderr = process.communicate(timeout=20)

if process.returncode:

failures.append((process.returncode, stdout, stderr))

self.assertEqual(failures, [])

state = json.loads(self._enabled.read_text())

self.assertEqual(

set(state), {f"parallel-{index}" for index in range(12)},

)

class CLIDriverTests(_IsolatedExt, unittest.TestCase):

"""``python3 -m lib.extensions ...`` entrypoint goes through the

same functions as the HTTP handlers."""

def test_list_prints_rows(self):

from lib.extensions import __main__ as driver

self._install_fake("0.7.1")

with mock.patch("sys.stdout") as out:

rc = driver.main(["list"])

self.assertEqual(rc, 0)

# At least one row should have been written.

self.assertTrue(out.write.called)

def test_install_failure_returns_nonzero(self):

from lib.extensions import __main__ as driver

def run(args, **_kwargs):

return subprocess.CompletedProcess(

args, 128, "", "fatal: could not resolve host")

with mock.patch("subprocess.run", side_effect=run):

rc = driver.main(["install", "agent"])

self.assertEqual(rc, 1)

def test_uninstall_with_remove_state_flag(self):

from lib.extensions import __main__ as driver

self._install_fake("0.7.1", state_paths=["agents.json"])

(Path(self._state.name) / "agents.json").write_text("{}")

rc = driver.main(["uninstall", "agent", "--remove-state"])

self.assertEqual(rc, 0)

self.assertFalse((Path(self._state.name) / "agents.json").exists())

if __name__ == "__main__":

unittest.main()