diff --git a/pm/packages.py b/pm/packages.py index a9491e77b4..247308f323 100644 --- a/pm/packages.py +++ b/pm/packages.py @@ -429,7 +429,7 @@ class Venv(StatePackage): replay = recorded.parent seed = (Path(prior["resolved_lock"]) if members and prior.get("resolved_lock") else project / "uv.lock") - lock_and_sync(members, extras, venv_dir=candidate, root=generation / "workspace", + lock_and_sync(members, extras, root=generation / "workspace", seed_lock=seed, frozen=repair or not members, replay=replay, source=project, environment=environment) resolved_lock = generation / "workspace" / "uv.lock" diff --git a/pm/workspace.py b/pm/workspace.py index db37ff6938..5d73d143b4 100644 --- a/pm/workspace.py +++ b/pm/workspace.py @@ -19,7 +19,6 @@ if TYPE_CHECKING: from pm import paths from pm.package import InstallError -WORKSPACE_DIRNAME = ".pm-workspace" _MEMBER_EXCLUDE = frozenset({".git", ".venv", "venv", "node_modules", "__pycache__"}) @@ -58,17 +57,6 @@ def classify_uv_failure(stage: str, returncode: int, output: str) -> InstallErro return InstallError("venv", cause) -def workspace_root() -> Path: - """Default preparation root; callers can supply a fresh transaction root.""" - from hermes_cli.runtime_paths import install_state_dir - return install_state_dir(paths.repo_root()) / WORKSPACE_DIRNAME - - -def _member_rel(root: Path, plugin_dir: Path) -> str: - """Use portable separators for a member inside the generated workspace.""" - return os.path.relpath(plugin_dir.resolve(), root.resolve()).replace("\\", "/") - - def member_sources(plugin_dirs) -> dict[Path, Path]: """Map installed identities to build inputs, including staged plugin updates.""" rows = plugin_dirs.items() if isinstance(plugin_dirs, Mapping) else ((path, path) for path in plugin_dirs) @@ -100,8 +88,6 @@ def members_stamp(plugin_dirs) -> str: def _copy_core_inputs(source: Path, destination: Path) -> None: """Build from a writable snapshot, never from signed/read-only source.""" - import shutil - import fnmatch import tomllib @@ -130,8 +116,6 @@ def _copy_core_inputs(source: Path, destination: Path) -> None: and not entry.name.startswith(".") and entry.resolve() != destination.resolve() and any(fnmatch.fnmatchcase(entry.name, pattern) for pattern in package_roots)): target = destination / entry.name - if target.exists(): - shutil.rmtree(target) shutil.copytree(entry, target, ignore=ignore) for name in files: entry = source / name @@ -144,24 +128,17 @@ def _copy_core_inputs(source: Path, destination: Path) -> None: shutil.copy2(entry, target) -def _generate_pyproject(plugin_dirs: list[Path], root: Optional[Path] = None, *, - source: Optional[Path] = None) -> tuple[Path, bool]: - """(Re)generate the workspace root's pyproject.toml from core's - pyproject + the enabled plugin members. Idempotent — same inputs, - same bytes. Returns (root, changed): changed is True when the member - surface moved (member set or a member's pyproject content), which is - the signal to re-seed the resolution from the committed lock.""" - if root is None: - root = workspace_root() - source = (paths.repo_root() if source is None else source).resolve() +def _generate_pyproject(plugin_dirs: list[Path] | Mapping[Path, Path], root: Path, *, source: Path) -> None: + """Snapshot core and plugin build inputs into a fresh generation.""" + source = source.resolve() if root.resolve() == source or source.is_relative_to(root.resolve()): raise InstallError("venv", "workspace must not replace the core source") - root.mkdir(parents=True, exist_ok=True) + root.mkdir(parents=True) core_pyproject = source / "pyproject.toml" core_text = core_pyproject.read_text(encoding="utf-8-sig") - members = [_member_rel(root, _workspace_member(source, root, identity=identity)) + members = [_workspace_member(source, root, identity=identity).relative_to(root).as_posix() for identity, source in member_sources(plugin_dirs).items()] lines = [core_text.rstrip("\n")] @@ -172,40 +149,8 @@ def _generate_pyproject(plugin_dirs: list[Path], root: Optional[Path] = None, *, text = "\n".join(lines) + "\n" target = root / "pyproject.toml" - try: - changed = target.read_text(encoding="utf-8") != text - except OSError: - changed = True _copy_core_inputs(source, root) target.write_text(text, encoding="utf-8") - return root, changed - - -def _seed_lock(root: Path, seed_lock: Optional[Path] = None, *, source: Optional[Path] = None) -> None: - """Seed the generated root's uv.lock with the CURRENT resolution. - - Seed precedence: the parent-supplied ``seed_lock`` path first, then - the root's own existing uv.lock (the current EXTENDED resolution from - the previous sync), then the committed core lock — so a plugin-driven - extension keeps every compatible selection it already made, and a - fresh root extends the committed resolution. uv preserves compatible - selections from the seed (a plugin's range spec does not move core - pins); explicit exact requirements stay binding as declared - constraints. The lock is COPIED — shipped/extended source bytes are - never rewritten; only the staging root receives the copy. - - Called only when the member surface changed; an unchanged root keeps - its lock untouched, so repeated syncs are stable. Seed failures - SURFACE (they would silently degrade the resolution otherwise).""" - if seed_lock is None: - existing = root / "uv.lock" - if existing.is_file(): - seed_lock = existing - else: - seed_lock = (paths.repo_root() if source is None else source) / "uv.lock" - if not seed_lock.is_file(): - return # nothing committed to seed from; uv resolves from scratch - (root / "uv.lock").write_bytes(seed_lock.read_bytes()) def _is_member_candidate(plugin_dir: Path) -> bool: @@ -283,18 +228,15 @@ def _legacy_requirements(plugin_dir: Path) -> list[str]: return list(dict.fromkeys(specs)) -def _workspace_member(plugin_dir: Path, root: Path, *, identity: Path | None = None) -> Path: +def _workspace_member(plugin_dir: Path, root: Path, *, identity: Path) -> Path: """Keep workspace members with their generation, not a temporary install clone.""" import json - import shutil import tomllib - key = hashlib.sha256(str((identity or plugin_dir).resolve()).encode()).hexdigest()[:16] + key = hashlib.sha256(str(identity.resolve()).encode()).hexdigest()[:16] pyproject = plugin_dir / "pyproject.toml" if pyproject.is_file() and "GENERATED by pm" not in pyproject.read_text(encoding="utf-8-sig"): member = root / "plugin-sources" / key - if member.exists(): - shutil.rmtree(member) shutil.copytree(plugin_dir, member, symlinks=True, ignore=_member_ignored) document = tomllib.loads(pyproject.read_text(encoding="utf-8-sig")) @@ -309,7 +251,7 @@ def _workspace_member(plugin_dir: Path, root: Path, *, identity: Path | None = N resolved = (plugin_dir / relative).resolve() if resolved.is_relative_to(plugin_dir.resolve()): continue # The referenced tree was copied with this member. - spec["path"] = ((identity or plugin_dir) / relative).resolve().as_posix() + spec["path"] = (identity / relative).resolve().as_posix() changed = True if changed: import tomli_w @@ -318,7 +260,7 @@ def _workspace_member(plugin_dir: Path, root: Path, *, identity: Path | None = N return member specs = _legacy_requirements(plugin_dir) member = root / "plugin-deps" / key - member.mkdir(parents=True, exist_ok=True) + member.mkdir(parents=True) (member / "pyproject.toml").write_text( f'[project]\nname = "hermes-plugin-{key}"\nversion = "0.0.0"\n' 'requires-python = ">=3.11"\n' @@ -376,57 +318,34 @@ def install_node_sidecar( def lock_and_sync( - plugin_dirs: list[Path], - extras: Optional[list[str]] = None, + plugin_dirs: list[Path] | Mapping[Path, Path], + extras: list[str], *, - venv_dir: Path, - root: Optional[Path] = None, - env: Optional[dict] = None, - seed_lock: Optional[Path] = None, + root: Path, + source: Path, + seed_lock: Path | None, + environment: PythonEnvironment, frozen: bool = False, - replay: Optional[Path] = None, - source: Optional[Path] = None, - environment: PythonEnvironment | None = None, + replay: Path | None = None, ) -> None: - """Build the root, then `uv lock` + `uv sync --frozen --extra ...`. + """Prepare a fresh generation using explicit inputs and a prepared engine. - Everything resolves into a parent-supplied STAGING surface: ``root`` - pins the generated workspace dir, ``venv_dir`` pins - UV_PROJECT_ENVIRONMENT and ``seed_lock`` (optional) pins which - existing lock seeds the extension (default: the root's current - extended lock, else the committed core lock). ``env`` replaces the - ambient base environment when supplied; either way the subprocess - gets a COPY — the live process environment is never mutated. ``replay`` - copies a recorded sibling workspace and uses its lock without resolution - or plugin discovery; it is reserved for restoring an existing selection. - ``source`` and ``environment`` bypass live source/tool discovery when supplied; - the environment owns the child process policy, cache and interpreter. - - Raises a CLASSIFIED InstallError on failure: ResolutionConflict only - for a confirmed resolver conflict; network, build and tool failures - stay generic InstallError — they are not evidence of a dependency - conflict and must not disable plugins. + The caller selects the seed; uv retains its compatible versions. Repair + copies the recorded workspace verbatim and never reads current manifests. + Resolver conflicts remain distinct from download/build failures. """ - if environment is not None and environment.destination != venv_dir: - raise ValueError("workspace and environment destinations differ") - + if root.exists() or root.is_symlink(): + raise InstallError("venv", f"workspace must be fresh: {root}") if replay is None: - generated, changed = _generate_pyproject(plugin_dirs, root, source=source) - if changed: - _seed_lock(generated, seed_lock, source=source) + _generate_pyproject(plugin_dirs, root, source=source) + if seed_lock is not None: + (root / "uv.lock").write_bytes(seed_lock.read_bytes()) else: - import shutil - - if root is None or not (replay / "pyproject.toml").is_file() or not (replay / "uv.lock").is_file(): + if not (replay / "pyproject.toml").is_file() or not (replay / "uv.lock").is_file(): raise InstallError("venv", f"recorded workspace is missing: {replay}") - # Generation workspaces are siblings at the same depth. External - # member paths still resolve. Generated members move with the copy. + # Sibling generations keep external relative paths at the same depth; + # snapshotted members and their exact lock travel with the workspace. shutil.copytree(replay, root, ignore=shutil.ignore_patterns("__pycache__", ".venv", "build", "*.egg-info")) - generated = root frozen = True - if environment is None: - from pm.environment import managed_environment - - environment = managed_environment(venv_dir, env=env) - environment.sync(generated, extras=extras or (), frozen=frozen) + environment.sync(root, extras=extras, frozen=frozen) diff --git a/tests/pm/test_environment_build.py b/tests/pm/test_environment_build.py index 6a4f5c7e2e..db9a0dafa6 100644 --- a/tests/pm/test_environment_build.py +++ b/tests/pm/test_environment_build.py @@ -504,7 +504,7 @@ def test_explicit_workspace_preserves_seed_and_replays_copied_members(locked_pro first.create() workspace.lock_and_sync( [original_member], ["chosen"], source=source, root=tmp_path / "first" / "workspace", - venv_dir=first.destination, environment=first, seed_lock=source / "uv.lock", + environment=first, seed_lock=source / "uv.lock", ) first.check() assert _run([str(first.executable), "-I", "-c", "import base_dep, member_dep; print(base_dep.__version__)"], @@ -522,7 +522,7 @@ def test_explicit_workspace_preserves_seed_and_replays_copied_members(locked_pro second.create() workspace.lock_and_sync( [], ["chosen"], source=source, root=tmp_path / "second" / "workspace", - venv_dir=second.destination, environment=second, replay=recorded, + environment=second, seed_lock=None, replay=recorded, ) second.check() assert _run([str(second.executable), "-I", "-c", "import member_dep, chosen_dep; print(member_dep.__version__)"], diff --git a/tests/pm/test_recovery_validation.py b/tests/pm/test_recovery_validation.py index f7b7c14619..cf754a0b7f 100644 --- a/tests/pm/test_recovery_validation.py +++ b/tests/pm/test_recovery_validation.py @@ -42,8 +42,10 @@ def test_validation_rejects_a_missing_required_import(tmp_path, monkeypatch, dam candidate = tmp_path / "venv" monkeypatch.setattr("pm._uv._toolchain", lambda **kwargs: (Path(uv), Path(sys.executable))) from pm.workspace import lock_and_sync + from pm.environment import managed_environment - lock_and_sync([], [], root=workspace, venv_dir=candidate) + lock_and_sync([], [], root=workspace, source=core, seed_lock=None, + environment=managed_environment(candidate)) python = candidate / ("Scripts/python.exe" if os.name == "nt" else "bin/python") validate_environment(python, env=env, cwd=workspace) shutil.rmtree(site_packages(candidate) / "dotenv") diff --git a/tests/pm/test_union_installs_members.py b/tests/pm/test_union_installs_members.py index de1b017c9c..bbb0fbf97c 100644 --- a/tests/pm/test_union_installs_members.py +++ b/tests/pm/test_union_installs_members.py @@ -11,12 +11,14 @@ lock, empty site-packages) can never silently return. from __future__ import annotations +import json import sys from pathlib import Path import pytest import pm.workspace as ws +from pm.environment import managed_environment @pytest.fixture(autouse=True) def isolated_machine_home(tmp_path, monkeypatch): @@ -44,12 +46,17 @@ def mini_workspace(tmp_path, monkeypatch): whose dep (pyfiglet) is NOT a root dep. Store paths pointed at tmp.""" core = tmp_path / "core" core.mkdir() + from tests.pm.test_workspace_build_inputs import _wheel + wheels = tmp_path / "wheels" + wheels.mkdir() + _wheel(wheels, "pyfiglet", "1.0.2") (core / "pyproject.toml").write_text( "[project]\n" 'name = "fake-core"\n' 'version = "0.1.0"\n' 'requires-python = ">=3.11"\n' - 'dependencies = []\n', + 'dependencies = []\n[tool.uv]\npackage=false\nno-index=true\n' + f'find-links=[{json.dumps(wheels.as_posix())}]\n', encoding="utf-8", ) plug = tmp_path / "plugins" / "member-plug" @@ -74,7 +81,7 @@ def mini_workspace(tmp_path, monkeypatch): monkeypatch.setattr(pm.paths, "store_root", lambda: store) monkeypatch.setattr(ws.paths, "repo_root", lambda: core) monkeypatch.setattr(ws.paths, "store_root", lambda: store) - return tmp_path, plug, venv + return tmp_path, core, plug, venv @pytest.mark.skipif(_uv_available() is False, reason="uv not on PATH") @@ -82,8 +89,9 @@ def test_lock_and_sync_installs_member_deps(mini_workspace): """The probe scenario: after lock_and_sync, the member's dep MUST be importable from the synced venv. Under plain `uv sync --frozen` the lock contains the dep but site-packages does not — this fails.""" - _, plug, venv = mini_workspace - ws.lock_and_sync([plug], [], venv_dir=venv) + tmp, core, plug, venv = mini_workspace + ws.lock_and_sync([plug], [], root=tmp / "workspace", source=core, seed_lock=None, + environment=managed_environment(venv)) sp = _site_packages(venv) assert sp.is_dir(), "no site-packages in the synced venv" @@ -102,9 +110,12 @@ def test_union_survives_a_resync(mini_workspace): """The prune-proof contract: a second lock_and_sync (what an update rebuild does) must NOT remove the member's deps — they are in the union lock, not pip-guests.""" - _, plug, venv = mini_workspace - ws.lock_and_sync([plug], [], venv_dir=venv) - ws.lock_and_sync([plug], [], venv_dir=venv) + tmp, core, plug, venv = mini_workspace + ws.lock_and_sync([plug], [], root=tmp / "workspace", source=core, seed_lock=None, + environment=managed_environment(venv)) + venv = tmp / "next-venv" + ws.lock_and_sync([plug], [], root=tmp / "next-workspace", source=core, + seed_lock=tmp / "workspace" / "uv.lock", environment=managed_environment(venv)) sp = _site_packages(venv) assert any(p.name.startswith("pyfiglet") for p in sp.iterdir()), ( diff --git a/tests/pm/test_workspace.py b/tests/pm/test_workspace.py index 5db887213d..527ef87b1a 100644 --- a/tests/pm/test_workspace.py +++ b/tests/pm/test_workspace.py @@ -3,9 +3,8 @@ The workspace root is a pm-GENERATED project (never the committed pyproject.toml — sealed installs are read-only and member lists are machine-specific). Its pyproject = core's pyproject verbatim + -``[tool.uv.workspace] members`` pointing at each enabled plugin dir via -relative ``../``-escaping paths (proven to resolve). ``uv lock`` unions -core + plugin deps into ONE lock; conflict = loud refusal. +``[tool.uv.workspace] members`` pointing at each snapshotted plugin. +``uv lock`` unions core + plugin deps into ONE lock; conflict = loud refusal. """ from __future__ import annotations @@ -20,6 +19,7 @@ from pathlib import Path import pytest import pm.workspace as ws +from pm.environment import managed_environment @pytest.fixture(autouse=True) @@ -68,17 +68,11 @@ def layout(tmp_path, monkeypatch): return tmp_path, core, plug_a, store -def test_workspace_root_is_per_install_not_in_the_store(layout): - from hermes_cli.runtime_paths import install_state_dir - _, core, _, store = layout - assert ws.workspace_root() == install_state_dir(core) / ".pm-workspace" - assert not ws.workspace_root().is_relative_to(store) - - def test_build_writes_core_pyproject_verbatim(layout): - _, core, plug_a, _ = layout - root = ws.workspace_root() - ws.lock_and_sync([plug_a], root=root, venv_dir=root.parent / "env") + tmp, core, plug_a, _ = layout + root = tmp / "workspace" + ws.lock_and_sync([plug_a], [], root=root, source=core, seed_lock=None, + environment=managed_environment(tmp / "env")) text = (root / "pyproject.toml").read_text(encoding="utf-8") core_text = (core / "pyproject.toml").read_text(encoding="utf-8") # core's project table is carried verbatim (name, deps, requires-python) @@ -93,9 +87,10 @@ def test_build_writes_core_pyproject_verbatim(layout): def test_members_keep_their_source_with_the_generation(layout): import tomllib - _, _, plug_a, _ = layout - root = ws.workspace_root() - ws.lock_and_sync([plug_a], root=root, venv_dir=root.parent / "env") + tmp, core, plug_a, _ = layout + root = tmp / "workspace" + ws.lock_and_sync([plug_a], [], root=root, source=core, seed_lock=None, + environment=managed_environment(tmp / "env")) document = tomllib.loads((root / "pyproject.toml").read_text(encoding="utf-8")) [relative] = document["tool"]["uv"]["workspace"]["members"] copied = root / relative / "pyproject.toml" @@ -106,20 +101,35 @@ def test_members_keep_their_source_with_the_generation(layout): assert copied.read_bytes() == before -def test_build_is_idempotent(layout): - _, _, plug_a, _ = layout - root = ws.workspace_root() - ws.lock_and_sync([plug_a], root=root, venv_dir=root.parent / "env") - first = (ws.workspace_root() / "pyproject.toml").read_text(encoding="utf-8") - ws.lock_and_sync([plug_a], root=root, venv_dir=root.parent / "env") - second = (ws.workspace_root() / "pyproject.toml").read_text(encoding="utf-8") - assert first == second +def test_preparation_refuses_existing_workspace_without_mutating_it(layout): + from pm.package import InstallError + + tmp, core, plug_a, _ = layout + root = tmp / "workspace" + environment = managed_environment(tmp / "env") + kwargs = dict(root=root, source=core, seed_lock=None, + environment=environment) + ws.lock_and_sync([plug_a], [], **kwargs) + before = {p.relative_to(root): p.read_bytes() for p in root.rglob("*") if p.is_file()} + (plug_a / "pyproject.toml").write_text('changed after publication') + with pytest.raises(InstallError, match="fresh"): + ws.lock_and_sync([], [], **kwargs) + assert {p.relative_to(root): p.read_bytes() for p in root.rglob("*") if p.is_file()} == before + + +def test_missing_explicit_seed_cannot_silently_resolve_new_versions(layout): + tmp, core, plug_a, _ = layout + with pytest.raises(FileNotFoundError): + ws.lock_and_sync([plug_a], [], root=tmp / "workspace", source=core, + seed_lock=tmp / "missing.lock", environment=managed_environment(tmp / "env")) + assert not (tmp / "env").exists() def test_zero_plugins_still_builds_a_root_with_no_members(layout): - _, _, _, _ = layout - root = ws.workspace_root() - ws.lock_and_sync([], root=root, venv_dir=root.parent / "env") + tmp, core, _, _ = layout + root = tmp / "workspace" + ws.lock_and_sync([], [], root=root, source=core, seed_lock=None, + environment=managed_environment(tmp / "env")) text = (root / "pyproject.toml").read_text(encoding="utf-8") assert 'name = "hermes-agent"' in text assert "[tool.uv.workspace]" not in text or "members = []" in text @@ -261,115 +271,40 @@ def test_classify_network_failure_stays_generic(): assert not isinstance(err, ResolutionConflict) -def test_sync_failure_is_never_a_conflict(tmp_path, monkeypatch): - """--frozen: the lock already resolved, so a sync failure (download, - build, tooling) must stay generic — it must not disable plugins.""" +def test_sync_failure_is_never_a_conflict(layout, monkeypatch): from pm.package import InstallError - from pm.workspace import ResolutionConflict - class FakeProc: - returncode = 1 - stderr = "error: Failed to download wheel (connection reset)" - stdout = "" - - monkeypatch.setattr(ws, "_generate_pyproject", lambda *a, **k: (Path("/x/ws"), False)) - monkeypatch.setattr("pm._uv._toolchain", lambda **kwargs: (Path("uv"), Path("pm-python"))) - - captured = {} - - def fake_run(cmd, **kwargs): - captured["cmd"] = cmd - return FakeProc() - - monkeypatch.setattr(subprocess, "run", fake_run) + tmp, core, _, _ = layout + environment = managed_environment(tmp / "candidate") + monkeypatch.setattr(subprocess, "run", lambda cmd, **kwargs: + subprocess.CompletedProcess(cmd, 1, "", "Failed to download wheel")) with pytest.raises(InstallError) as excinfo: - ws.lock_and_sync([], [], venv_dir=tmp_path / "candidate") - assert not isinstance(excinfo.value, ResolutionConflict) + ws.lock_and_sync([], [], root=tmp / "workspace", source=core, + seed_lock=None, environment=environment, frozen=True) + assert not isinstance(excinfo.value, ws.ResolutionConflict) -def test_staging_root_and_env_are_honored_without_live_mutation(monkeypatch, tmp_path): - """lock_and_sync must resolve into the PARENT-SUPPLIED staging root + - venv, pass a COPY of the environment (never mutate the live one), and - leave the default generated root untouched.""" - staging = tmp_path / "staging-ws" - staging.mkdir() - - seen = {} - - class FakeProc: - returncode = 0 - stderr = "" - stdout = "" - - def fake_run(cmd, cwd=None, env=None, **kwargs): - seen["cwd"] = cwd - seen["env"] = env - return FakeProc() - - monkeypatch.setattr(ws, "_generate_pyproject", lambda *a, **k: (staging, False)) - monkeypatch.setattr("pm._uv._toolchain", lambda **kwargs: (Path("uv"), Path("pm-python"))) - monkeypatch.setattr("pm.packages.uv_cache_dir", lambda: tmp_path / "cache") - monkeypatch.setattr(subprocess, "run", fake_run) - - live_key = "PM_WORKSPACE_TEST_SENTINEL" - os.environ[live_key] = "live" - try: - ws.lock_and_sync( - [], [], venv_dir=tmp_path / "staging-venv", root=staging, - env={"PATH": "/staged/bin", live_key: "staged"}, - ) - - assert Path(seen["cwd"]) == staging - assert seen["env"][live_key] == "staged" # staged env wins - assert seen["env"]["UV_CACHE_DIR"] == str(tmp_path / "cache") - assert seen["env"]["UV_PROJECT_ENVIRONMENT"] == str(tmp_path / "staging-venv") - assert seen["env"]["UV_PYTHON"] == "pm-python" - assert os.environ[live_key] == "live" # live env untouched - finally: - del os.environ[live_key] - - -def test_changed_root_seeds_from_committed_lock_unchanged_keeps_extended( - layout, monkeypatch, tmp_path -): - """Seed the CURRENT resolution without writing shipped bytes: a fresh - root seeds from the committed lock; an existing root seeds from its - own extended lock; an explicit parent-supplied seed_lock wins.""" - _, core, plug_a, store = layout - (core / "uv.lock").write_bytes(b"committed-bytes\n") - - class FakeProc: - returncode = 0 - stderr = "" - stdout = "" - - monkeypatch.setattr("pm._uv._toolchain", lambda **kwargs: (Path("uv"), Path("pm-python"))) - monkeypatch.setattr(subprocess, "run", lambda cmd, **k: FakeProc()) - - root = ws.workspace_root() - venv = tmp_path / "venv" - - # First sync: fresh root -> seeded from the COMMITTED lock. - ws.lock_and_sync([plug_a], [], venv_dir=venv) - assert (root / "uv.lock").read_bytes() == b"committed-bytes\n" - - # Changed surface (new member), root's own EXTENDED lock present -> - # the current extended resolution is the seed, not the committed one. - (root / "uv.lock").write_bytes(b"extended-bytes\n") - plug_b = layout[0] / "home" / "plugins" / "plug-b" - plug_b.mkdir(parents=True) - (plug_b / "pyproject.toml").write_text("[project]\n", encoding="utf-8") - ws.lock_and_sync([plug_a, plug_b], [], venv_dir=venv) - assert (root / "uv.lock").read_bytes() == b"extended-bytes\n" - - # Parent-supplied seed_lock wins over both. - other = tmp_path / "parent-extended.lock" - other.write_bytes(b"parent-bytes\n") - plug_c = layout[0] / "home" / "plugins" / "plug-c" - plug_c.mkdir(parents=True) - (plug_c / "pyproject.toml").write_text("[project]\n", encoding="utf-8") - ws.lock_and_sync([plug_a, plug_b, plug_c], [], venv_dir=venv, seed_lock=other) - assert (root / "uv.lock").read_bytes() == b"parent-bytes\n" - - # Shipped core lock untouched throughout. - assert (core / "uv.lock").read_bytes() == b"committed-bytes\n" +def test_staging_root_and_env_are_honored_without_live_mutation(layout, monkeypatch): + tmp, core, _, _ = layout + staging = tmp / "staging-ws" + monkeypatch.setenv("PM_WORKSPACE_TEST_SENTINEL", "live") + environment = managed_environment(tmp / "staging-venv", env={ + "PATH": "/staged/bin", "PM_WORKSPACE_TEST_SENTINEL": "staged", + }) + # The prepared environment is authoritative; workspace never discovers tools. + monkeypatch.setattr(shutil, "which", lambda *args, **kwargs: pytest.fail("PATH discovery")) + seen = [] + def run(cmd, **kwargs): + seen.append((cmd, kwargs)) + return subprocess.CompletedProcess(cmd, 0, "", "") + monkeypatch.setattr(subprocess, "run", run) + ws.lock_and_sync([], [], root=staging, source=core, seed_lock=None, environment=environment) + assert [cmd[1] for cmd, _ in seen] == ["lock", "sync"] + for cmd, kwargs in seen: + assert Path(cmd[0]) == environment.uv + assert Path(kwargs["cwd"]) == staging + assert kwargs["env"]["PM_WORKSPACE_TEST_SENTINEL"] == "staged" + assert kwargs["env"]["UV_CACHE_DIR"] == str(environment.cache) + assert kwargs["env"]["UV_PROJECT_ENVIRONMENT"] == str(environment.destination) + assert kwargs["env"]["UV_PYTHON"] == str(environment.python) + assert os.environ["PM_WORKSPACE_TEST_SENTINEL"] == "live" diff --git a/tests/pm/test_workspace_build_inputs.py b/tests/pm/test_workspace_build_inputs.py index dd6a6b4b7d..99071a176c 100644 --- a/tests/pm/test_workspace_build_inputs.py +++ b/tests/pm/test_workspace_build_inputs.py @@ -8,6 +8,7 @@ import sys import pytest from pm import workspace +from pm.environment import managed_environment @pytest.fixture(autouse=True) @@ -37,7 +38,8 @@ def test_real_build_inputs_stay_in_generated_root(tmp_path, monkeypatch): uv = shutil.which("uv") assert uv is not None monkeypatch.setattr("pm._uv._toolchain", lambda **kwargs: (Path(uv), Path(sys.executable))) - workspace.lock_and_sync([], [], root=root, venv_dir=venv) + workspace.lock_and_sync([], [], root=root, source=core, seed_lock=None, + environment=managed_environment(venv)) python = venv / ("Scripts/python.exe" if sys.platform == "win32" else "bin/python") probe = subprocess.run([str(python), "-c", "import buildable_core; print(buildable_core.VALUE)"], cwd=tmp_path, text=True, capture_output=True, check=True, timeout=30) @@ -57,13 +59,19 @@ def test_source_refresh_does_not_need_metadata_change_and_refuses_live_root(tmp_ uv = shutil.which("uv") assert uv monkeypatch.setattr("pm._uv._toolchain", lambda **kwargs: (Path(uv), Path(sys.executable))) - workspace.lock_and_sync([], root=staged, venv_dir=tmp_path / "env") + workspace.lock_and_sync([], [], root=staged, source=core, seed_lock=None, + environment=managed_environment(tmp_path / "env")) (core / "code.py").write_text("VALUE = 2\n") - workspace.lock_and_sync([], root=staged, venv_dir=tmp_path / "env") + fresh = tmp_path / "fresh" + workspace.lock_and_sync([], [], root=fresh, source=core, seed_lock=staged / "uv.lock", + environment=managed_environment(tmp_path / "fresh-env")) + assert (staged / "code.py").read_text() == "VALUE = 1\n" + staged = fresh assert (staged / "code.py").read_text() == "VALUE = 2\n" before = (core / "code.py").read_bytes() - with pytest.raises(workspace.InstallError, match="source"): - workspace.lock_and_sync([], root=core, venv_dir=tmp_path / "env") + with pytest.raises(workspace.InstallError, match="fresh"): + workspace.lock_and_sync([], [], root=core, source=core, seed_lock=None, + environment=managed_environment(tmp_path / "env")) assert (core / "code.py").read_bytes() == before @@ -86,7 +94,8 @@ def test_legacy_member_is_generated_only_inside_workspace(tmp_path, monkeypatch) assert uv monkeypatch.setattr("pm._uv._toolchain", lambda **kwargs: (Path(uv), Path(sys.executable))) generated = tmp_path / "stage" - workspace.lock_and_sync([plugin], root=generated, venv_dir=tmp_path / "env") + workspace.lock_and_sync([plugin], [], root=generated, source=core, seed_lock=None, + environment=managed_environment(tmp_path / "env")) metadata = tomllib.loads((generated / "pyproject.toml").read_text()) member = (generated / metadata["tool"]["uv"]["workspace"]["members"][0]).resolve() assert member.is_relative_to(generated) @@ -134,7 +143,8 @@ def test_plugin_can_move_compatible_transitive_but_not_exact_requirement(tmp_pat monkeypatch.setattr(workspace.paths, "repo_root", lambda: core) monkeypatch.setattr("pm._uv._toolchain", lambda **kwargs: (Path(uv), Path(sys.executable))) baseline, first_env = tmp_path / "baseline", tmp_path / "first-env" - workspace.lock_and_sync([], [], root=baseline, venv_dir=first_env) + workspace.lock_and_sync([], [], root=baseline, source=core, seed_lock=None, + environment=managed_environment(first_env)) first_lock = (baseline / "uv.lock").read_bytes() assert next(p["version"] for p in tomllib.loads(first_lock.decode())["package"] if p["name"] == "pkgb") == "1.2" _wheel(wheels, "pkgb", "1.3") @@ -147,9 +157,11 @@ def test_plugin_can_move_compatible_transitive_but_not_exact_requirement(tmp_pat extended, candidate = tmp_path / "extended", tmp_path / "candidate" if exact: with pytest.raises(workspace.ResolutionConflict): - workspace.lock_and_sync([plugin], [], root=extended, venv_dir=candidate, seed_lock=baseline / "uv.lock") + workspace.lock_and_sync([plugin], [], root=extended, source=core, seed_lock=baseline / "uv.lock", + environment=managed_environment(candidate)) else: - workspace.lock_and_sync([plugin], [], root=extended, venv_dir=candidate, seed_lock=baseline / "uv.lock") + workspace.lock_and_sync([plugin], [], root=extended, source=core, seed_lock=baseline / "uv.lock", + environment=managed_environment(candidate)) installed = tomllib.loads((extended / "uv.lock").read_text(encoding="utf-8"))["package"] assert next(p["version"] for p in installed if p["name"] == "pkgb") == "1.3" python = candidate / ("Scripts/python.exe" if os.name == "nt" else "bin/python") diff --git a/tests/pm/test_workspace_output_encoding.py b/tests/pm/test_workspace_output_encoding.py index 5a21c2d5f3..0563f52f4a 100644 --- a/tests/pm/test_workspace_output_encoding.py +++ b/tests/pm/test_workspace_output_encoding.py @@ -42,7 +42,9 @@ def test_uv_failure_retains_utf8_build_diagnostic( diagnostic = "🔍 cryptography: OpenSSL headers not found" raw = diagnostic.encode("utf-8") + suffix + b"\n" expected = diagnostic + ("�" if suffix else "") - monkeypatch.setattr(ws, "_generate_pyproject", lambda *a, **k: (tmp_path, False)) + core = tmp_path / "core" + core.mkdir() + (core / "pyproject.toml").write_text('[project]\nname="test-core"\nversion="1"\n') from pm.environment import PythonEnvironment environment = PythonEnvironment( @@ -62,8 +64,8 @@ def test_uv_failure_retains_utf8_build_diagnostic( monkeypatch.setattr("pm.environment.subprocess.run", run_uv) with pytest.raises(InstallError) as excinfo: - ws.lock_and_sync([], venv_dir=environment.destination, root=tmp_path, - source=tmp_path, environment=environment) + ws.lock_and_sync([], [], root=tmp_path / "workspace", source=core, + seed_lock=None, environment=environment) assert type(excinfo.value) is InstallError # A build error is not a resolver conflict. assert excinfo.value.cause == f"uv {stage} exited 17: {expected}" diff --git a/tests/test_managed_runtime_resolution.py b/tests/test_managed_runtime_resolution.py deleted file mode 100644 index 597c0cbc3c..0000000000 --- a/tests/test_managed_runtime_resolution.py +++ /dev/null @@ -1,35 +0,0 @@ -"""Workspace commands preserve PM's toolchain instead of consulting PATH.""" -import importlib -import shutil -import subprocess - -import pytest - -from pm import workspace -from pm.package import InstallError - - -def test_workspace_uv_preserves_managed_tool_and_interpreter(monkeypatch, tmp_path): - ensure = importlib.import_module("pm.ensure") - monkeypatch.setattr(ensure, "uv", lambda **kwargs: ("managed/uv", {"UV_PYTHON": "managed/python", "UV_CACHE_DIR": str(tmp_path / "cache")})) - monkeypatch.setattr(workspace, "_generate_pyproject", lambda *args, **kwargs: (tmp_path, False)) - def reject_path(*args, **kwargs): - raise AssertionError("workspace commands must not resolve tools from PATH") - monkeypatch.setattr(shutil, "which", reject_path) - calls = [] - def run(cmd, **kwargs): - calls.append((cmd, kwargs["env"])) - return subprocess.CompletedProcess(cmd, 0, "", "") - monkeypatch.setattr(workspace.subprocess, "run", run) - workspace.lock_and_sync([], venv_dir=tmp_path / "venv", root=tmp_path) - assert [cmd[1] for cmd, _ in calls] == ["lock", "sync"] - assert all(cmd[0] == "managed/uv" and env["UV_PYTHON"] == "managed/python" for cmd, env in calls) - - -def test_workspace_uv_missing_toolchain_never_falls_back(monkeypatch, tmp_path): - ensure = importlib.import_module("pm.ensure") - monkeypatch.setattr(ensure, "uv", lambda **kwargs: (None, {})) - monkeypatch.setattr(workspace, "_generate_pyproject", lambda *args, **kwargs: (tmp_path, False)) - monkeypatch.setattr(shutil, "which", lambda name: "developer/uv") - with pytest.raises(InstallError, match="PM's uv and Python"): - workspace.lock_and_sync([], venv_dir=tmp_path / "venv", root=tmp_path)