wehub-resource-sync
345 lines
12 KiB
Python
345 lines
12 KiB
Python
"""Tests for the cross-platform process/platform primitives and Windows backend.
|
|
|
|
Covers omnigent._platform, omnigent.inner._proc, the harness IPC endpoint
|
|
abstraction, and the windows_jobobject sandbox backend. The platform-specific
|
|
assertions are gated with ``posix_only`` / ``windows_only`` markers so the
|
|
file runs on both Linux CI and a Windows box.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import subprocess
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from omnigent import _platform
|
|
from omnigent.inner import _proc
|
|
|
|
|
|
def _spin_cmd() -> list[str]:
|
|
"""A short-lived child process that does nothing but sleep."""
|
|
if os.name == "nt":
|
|
return ["cmd", "/c", "ping -n 30 127.0.0.1 >NUL"]
|
|
return ["sleep", "30"]
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# _platform
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
def test_platform_flags_are_mutually_consistent() -> None:
|
|
assert (os.name == "nt") == _platform.IS_WINDOWS
|
|
assert (os.name == "posix") == _platform.IS_POSIX
|
|
# Exactly one OS family is true.
|
|
assert _platform.IS_WINDOWS != _platform.IS_POSIX
|
|
|
|
|
|
def test_default_shell_argv_runs_an_echo() -> None:
|
|
argv = _platform.default_shell_argv("echo omnigent-shell-ok")
|
|
out = subprocess.run(argv, capture_output=True, text=True, check=True)
|
|
assert "omnigent-shell-ok" in out.stdout
|
|
|
|
|
|
@pytest.mark.posix_only
|
|
def test_default_interactive_shell_honors_known_shell_on_path(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""A ``$SHELL`` naming a known shell that resolves on PATH is kept.
|
|
|
|
Returns the basename (not the ``$SHELL`` path) so it stays resolvable
|
|
when the terminal launches under a runner on a different host.
|
|
"""
|
|
# Patch shutil.which on the shared module so the function's local
|
|
# ``import shutil`` sees the fake; make every known shell "installed".
|
|
monkeypatch.setattr("shutil.which", lambda name: f"/usr/bin/{name}")
|
|
for shell_path, expected in (
|
|
("/bin/zsh", "zsh"),
|
|
("/usr/local/bin/fish", "fish"),
|
|
("/bin/bash", "bash"),
|
|
):
|
|
monkeypatch.setenv("SHELL", shell_path)
|
|
assert _platform.default_interactive_shell() == expected
|
|
|
|
|
|
@pytest.mark.posix_only
|
|
def test_default_interactive_shell_falls_back_to_bash(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Unset / unknown / not-on-PATH ``$SHELL`` all fall back to bash."""
|
|
# Unset $SHELL → bash (known shells resolve on PATH here).
|
|
monkeypatch.setattr("shutil.which", lambda name: f"/usr/bin/{name}")
|
|
monkeypatch.delenv("SHELL", raising=False)
|
|
assert _platform.default_interactive_shell() == "bash"
|
|
# An unknown shell name is never honored, even if on PATH.
|
|
monkeypatch.setenv("SHELL", "/opt/weird/nushell")
|
|
assert _platform.default_interactive_shell() == "bash"
|
|
# A known shell that does not resolve on PATH also falls back.
|
|
monkeypatch.setattr("shutil.which", lambda name: None)
|
|
monkeypatch.setenv("SHELL", "/bin/zsh")
|
|
assert _platform.default_interactive_shell() == "bash"
|
|
|
|
|
|
@pytest.mark.posix_only
|
|
def test_installed_interactive_shells_lists_default_first_then_alternatives(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""The default ($SHELL) leads; installed alternatives follow, deduped."""
|
|
monkeypatch.setattr("shutil.which", lambda name: f"/usr/bin/{name}")
|
|
monkeypatch.setenv("SHELL", "/bin/zsh")
|
|
# zsh is the default → first; bash/fish follow in offer order; no dupes.
|
|
assert _platform.installed_interactive_shells() == ["zsh", "bash", "fish"]
|
|
|
|
|
|
@pytest.mark.posix_only
|
|
def test_installed_interactive_shells_skips_uninstalled(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Alternatives that don't resolve on PATH are omitted."""
|
|
# Only bash and zsh installed; fish absent.
|
|
monkeypatch.setattr(
|
|
"shutil.which", lambda name: f"/usr/bin/{name}" if name in {"bash", "zsh"} else None
|
|
)
|
|
monkeypatch.setenv("SHELL", "/bin/bash")
|
|
assert _platform.installed_interactive_shells() == ["bash", "zsh"]
|
|
|
|
|
|
@pytest.mark.posix_only
|
|
def test_installed_interactive_shells_always_nonempty(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""With nothing resolvable the bash fallback is still offered alone."""
|
|
monkeypatch.setattr("shutil.which", lambda name: None)
|
|
monkeypatch.delenv("SHELL", raising=False)
|
|
assert _platform.installed_interactive_shells() == ["bash"]
|
|
|
|
|
|
def test_stable_user_id_is_stable_and_path_safe() -> None:
|
|
uid = _platform.stable_user_id()
|
|
assert uid == _platform.stable_user_id()
|
|
assert uid and not set(uid) & set("/\\: ")
|
|
|
|
|
|
@pytest.mark.windows_only
|
|
def test_resolve_repo_symlink_dereferences_git_stub(tmp_path: Path) -> None:
|
|
# Real target the Git symlink points at.
|
|
target = tmp_path / "examples" / "polly"
|
|
target.mkdir(parents=True)
|
|
(target / "config.yaml").write_text("name: polly\n", encoding="utf-8")
|
|
# Stub file Git leaves on a no-symlink Windows checkout: content is the
|
|
# relative link target, no trailing newline.
|
|
stub = tmp_path / "resources" / "polly"
|
|
stub.parent.mkdir(parents=True)
|
|
stub.write_text("../examples/polly", encoding="utf-8")
|
|
|
|
resolved = _platform.resolve_repo_symlink(stub)
|
|
assert resolved == target.resolve()
|
|
|
|
|
|
@pytest.mark.windows_only
|
|
def test_resolve_repo_symlink_leaves_real_specs_untouched(tmp_path: Path) -> None:
|
|
# A genuine single-file YAML spec must not be mistaken for a symlink stub.
|
|
spec = tmp_path / "agent.yaml"
|
|
spec.write_text("name: hello\nharness: claude-sdk\n", encoding="utf-8")
|
|
assert _platform.resolve_repo_symlink(spec) == spec
|
|
# A real directory is returned unchanged.
|
|
d = tmp_path / "bundle"
|
|
d.mkdir()
|
|
assert _platform.resolve_repo_symlink(d) == d
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# _proc
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
def test_spawn_kwargs_shape_matches_platform() -> None:
|
|
kw = _proc.spawn_kwargs()
|
|
if os.name == "nt":
|
|
assert kw == {"creationflags": subprocess.CREATE_NEW_PROCESS_GROUP}
|
|
else:
|
|
assert kw == {"start_new_session": True}
|
|
|
|
|
|
def test_process_alive_is_a_nondestructive_probe() -> None:
|
|
proc = subprocess.Popen(_spin_cmd(), **_proc.spawn_kwargs())
|
|
try:
|
|
assert _proc.process_alive(proc.pid) is True
|
|
# Probing repeatedly must NOT kill the process (the os.kill(pid, 0)
|
|
# bug on Windows would terminate it here).
|
|
for _ in range(3):
|
|
assert _proc.process_alive(proc.pid) is True
|
|
finally:
|
|
_proc.kill_tree(proc)
|
|
proc.wait(timeout=5)
|
|
assert _proc.process_alive(proc.pid) is False
|
|
|
|
|
|
def test_process_alive_false_for_bogus_pid() -> None:
|
|
assert _proc.process_alive(2_000_000_000) is False
|
|
assert _proc.process_alive(-1) is False
|
|
|
|
|
|
def test_terminate_tree_stops_the_process() -> None:
|
|
proc = subprocess.Popen(_spin_cmd(), **_proc.spawn_kwargs())
|
|
_proc.terminate_tree(proc, grace=5)
|
|
proc.wait(timeout=5)
|
|
assert _proc.process_alive(proc.pid) is False
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Harness IPC endpoint abstraction
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
def test_endpoint_uds_variant_shape() -> None:
|
|
from omnigent.runtime.harnesses.process_manager import _HarnessEndpoint
|
|
|
|
ep = _HarnessEndpoint(socket_path=Path("/tmp/x/conv.sock"))
|
|
assert ep.is_uds is True
|
|
assert ep.spawn_args() == ["--socket", str(Path("/tmp/x/conv.sock"))]
|
|
assert ep.base_url == "http://harness.local"
|
|
|
|
|
|
def test_endpoint_tcp_variant_shape() -> None:
|
|
from omnigent.runtime.harnesses.process_manager import _HarnessEndpoint
|
|
|
|
ep = _HarnessEndpoint(host="127.0.0.1", port=54321)
|
|
assert ep.is_uds is False
|
|
assert ep.spawn_args() == ["--bind", "127.0.0.1:54321"]
|
|
assert ep.base_url == "http://127.0.0.1:54321"
|
|
|
|
|
|
def test_endpoint_create_picks_platform_transport() -> None:
|
|
from omnigent.runtime.harnesses.process_manager import _HarnessEndpoint
|
|
|
|
ep = _HarnessEndpoint.create(Path("/tmp/inst"), "conv_x")
|
|
assert ep.is_uds == (os.name != "nt")
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# windows_jobobject backend
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.windows_only
|
|
def test_windows_jobobject_is_platform_default() -> None:
|
|
from omnigent.inner import sandbox
|
|
|
|
assert sandbox._default_sandbox_for_platform().type == "windows_jobobject"
|
|
|
|
|
|
@pytest.mark.windows_only
|
|
def test_windows_jobobject_kill_on_close_terminates_tree() -> None:
|
|
from omnigent.inner.sandbox import SandboxPolicy
|
|
from omnigent.inner.windows_jobobject_sandbox import WindowsJobObjectSandboxBackend
|
|
|
|
backend = WindowsJobObjectSandboxBackend()
|
|
policy = SandboxPolicy(
|
|
backend_type="windows_jobobject",
|
|
active=True,
|
|
read_roots=None,
|
|
write_roots=[],
|
|
write_files=[],
|
|
allow_network=True,
|
|
)
|
|
proc = subprocess.Popen(_spin_cmd())
|
|
handle = backend.post_spawn(policy, proc.pid)
|
|
assert handle is not None
|
|
assert _proc.process_alive(proc.pid) is True
|
|
# Closing the job handle must terminate the contained process.
|
|
handle.close()
|
|
time.sleep(0.5)
|
|
assert _proc.process_alive(proc.pid) is False
|
|
if _proc.process_alive(proc.pid):
|
|
proc.kill()
|
|
|
|
|
|
@pytest.mark.windows_only
|
|
def test_explicit_bwrap_errors_loudly_on_windows() -> None:
|
|
from omnigent.inner import sandbox
|
|
from omnigent.inner.datamodel import OSEnvSandboxSpec, OSEnvSpec
|
|
|
|
sandbox._ensure_builtin_backends()
|
|
backend = sandbox.get_backend("linux_bwrap")
|
|
with pytest.raises(OSError):
|
|
backend.resolve(OSEnvSpec(sandbox=OSEnvSandboxSpec(type="linux_bwrap")), Path("."))
|
|
|
|
|
|
@pytest.mark.posix_only
|
|
def test_posix_default_sandbox_is_not_jobobject() -> None:
|
|
from omnigent.inner import sandbox
|
|
|
|
assert sandbox._default_sandbox_for_platform().type in {"linux_bwrap", "darwin_seatbelt"}
|
|
|
|
|
|
@pytest.mark.windows_only
|
|
def test_helper_env_keeps_systemroot_so_child_can_import_asyncio() -> None:
|
|
# Regression: a filtered (active-sandbox) helper env that drops SYSTEMROOT
|
|
# makes any spawned `python -m omnigent...` die at `import asyncio` with
|
|
# WinError 10106 (Winsock loads providers from %SystemRoot%). The os_env
|
|
# allowlist must carry the Windows system vars.
|
|
from omnigent.inner.os_env import build_helper_env
|
|
from omnigent.inner.sandbox import SandboxPolicy
|
|
|
|
policy = SandboxPolicy(
|
|
backend_type="windows_jobobject",
|
|
active=True,
|
|
read_roots=None,
|
|
write_roots=[],
|
|
write_files=[],
|
|
allow_network=True,
|
|
)
|
|
env = build_helper_env(os.environ, policy)
|
|
assert "SYSTEMROOT" in env
|
|
result = subprocess.run(
|
|
[__import__("sys").executable, "-c", "import asyncio"],
|
|
env=env,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert result.returncode == 0, result.stderr
|
|
|
|
|
|
@pytest.mark.windows_only
|
|
def test_parent_death_watchdog_does_not_false_fire_on_windows() -> None:
|
|
# Regression: on Windows os.getppid() does not match the spawner (the venv
|
|
# launcher breaks the parent link), so the getppid-based orphan check
|
|
# reported the runner orphaned the instant it started and tore it down
|
|
# (clean exit 0). With a live parent_pid the runner must NOT be orphaned.
|
|
from omnigent.runner._entry import _parent_is_orphaned
|
|
|
|
assert _parent_is_orphaned(os.getpid()) is False
|
|
assert _parent_is_orphaned(2_000_000_000) is True
|
|
|
|
|
|
@pytest.mark.windows_only
|
|
def test_host_runner_env_lets_child_import_asyncio_and_resolve_home() -> None:
|
|
# Regression: the host->runner env allowlist dropped SYSTEMROOT (asyncio /
|
|
# WinError 10106) and USERPROFILE (Path.home() -> "Could not determine home
|
|
# directory"). Both must pass through so a spawned runner can boot.
|
|
import sys
|
|
|
|
from omnigent.host.connect import _build_runner_env
|
|
|
|
env = _build_runner_env(
|
|
base_env=os.environ,
|
|
server_url="http://x",
|
|
runner_id="r1",
|
|
binding_token="t",
|
|
workspace=".",
|
|
parent_pid=os.getpid(),
|
|
)
|
|
assert "SYSTEMROOT" in env
|
|
assert "USERPROFILE" in env
|
|
result = subprocess.run(
|
|
[sys.executable, "-c", "import asyncio; from pathlib import Path; Path.home()"],
|
|
env=env,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert result.returncode == 0, result.stderr
|