test_host_policy_applies_prlimit() — langchain Function Reference

Source Code

libs/langchain_v1/tests/unit_tests/agents/middleware/implementations/test_shell_execution_policies.py lines 74–109

def test_host_policy_applies_prlimit(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    fake_resource = _make_resource(with_prlimit=True)
    monkeypatch.setattr(_execution, "resource", fake_resource, raising=False)
    monkeypatch.setattr(sys, "platform", "linux")

    recorded: dict[str, Any] = {}

    def fake_launch(
        command: Sequence[str],
        *,
        env: Mapping[str, str],
        cwd: Path,
        preexec_fn: Callable[[], None] | None,
        start_new_session: bool,
    ) -> subprocess.Popen[str]:
        recorded["command"] = list(command)
        recorded["env"] = dict(env)
        recorded["cwd"] = cwd
        recorded["preexec_fn"] = preexec_fn
        recorded["start_new_session"] = start_new_session
        return Mock(spec=subprocess.Popen, pid=1234)

    monkeypatch.setattr(_execution, "_launch_subprocess", fake_launch)

    policy = HostExecutionPolicy(cpu_time_seconds=2, memory_bytes=4096)
    env = {"PATH": os.environ.get("PATH", ""), "VAR": "1"}
    process = policy.spawn(workspace=tmp_path, env=env, command=("/bin/sh",))

    assert process is not None
    assert recorded["preexec_fn"] is None
    assert recorded["start_new_session"] is True
    assert fake_resource.prlimit_calls == [
        (1234, fake_resource.RLIMIT_CPU, (2, 2)),
        (1234, fake_resource.RLIMIT_AS, (4096, 4096)),
    ]
    assert fake_resource.setrlimit_calls == []