Fixed test_webui only failing in nix_sandbox
This commit is contained in:
@@ -5,6 +5,7 @@ import shlex
|
||||
import stat
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable, Dict, List, Optional
|
||||
|
||||
@@ -17,7 +18,42 @@ def command_exec(cmd: List[str], work_dir: Path, env: Dict[str, str]) -> None:
|
||||
subprocess.run(cmd, check=True, env=env, cwd=work_dir.resolve())
|
||||
|
||||
|
||||
def repro_env_break(
|
||||
def block_for_input() -> None:
|
||||
log = logging.getLogger(__name__)
|
||||
procid = os.getpid()
|
||||
command = f"echo 'continue' > /sys/proc/{procid}/fd/{sys.stdin.fileno()}"
|
||||
|
||||
while True:
|
||||
log.warning("Use sudo cntr attach <pid> to attach to the container.")
|
||||
log.warning("Resume execution by executing '%s' in cntr shell", command)
|
||||
res = input("Input 'continue' to resume execution: ")
|
||||
if res == "continue":
|
||||
break
|
||||
time.sleep(1)
|
||||
log.info("Resuming execution.")
|
||||
|
||||
|
||||
def breakpoint_container(
|
||||
work_dir: Path,
|
||||
env: Optional[Dict[str, str]] = None,
|
||||
cmd: Optional[List[str]] = None,
|
||||
) -> None:
|
||||
if env is None:
|
||||
env = os.environ.copy()
|
||||
else:
|
||||
env = env.copy()
|
||||
|
||||
dump_env(env, work_dir / "env.sh")
|
||||
|
||||
if cmd is not None:
|
||||
log.debug("Command: %s", shlex.join(cmd))
|
||||
mycommand = shlex.join(cmd)
|
||||
write_command(mycommand, work_dir / "cmd.sh")
|
||||
|
||||
block_for_input()
|
||||
|
||||
|
||||
def breakpoint_shell(
|
||||
work_dir: Path,
|
||||
env: Optional[Dict[str, str]] = None,
|
||||
cmd: Optional[List[str]] = None,
|
||||
@@ -32,7 +68,6 @@ def repro_env_break(
|
||||
if cmd is not None:
|
||||
mycommand = shlex.join(cmd)
|
||||
write_command(mycommand, work_dir / "cmd.sh")
|
||||
print(f"Adding to zsh history the command: {mycommand}", file=sys.stderr)
|
||||
proc = spawn_process(func=command_exec, cmd=args, work_dir=work_dir, env=env)
|
||||
|
||||
try:
|
||||
@@ -42,6 +77,7 @@ def repro_env_break(
|
||||
|
||||
|
||||
def write_command(command: str, loc: Path) -> None:
|
||||
log.info("Dumping command to %s", loc)
|
||||
with open(loc, "w") as f:
|
||||
f.write("#!/usr/bin/env bash\n")
|
||||
f.write(command)
|
||||
@@ -53,13 +89,14 @@ def spawn_process(func: Callable, **kwargs: Any) -> mp.Process:
|
||||
if mp.get_start_method(allow_none=True) is None:
|
||||
mp.set_start_method(method="spawn")
|
||||
|
||||
proc = mp.Process(target=func, kwargs=kwargs)
|
||||
proc = mp.Process(target=func, name="python-debug-process", kwargs=kwargs)
|
||||
proc.start()
|
||||
return proc
|
||||
|
||||
|
||||
def dump_env(env: Dict[str, str], loc: Path) -> None:
|
||||
cenv = env.copy()
|
||||
log.info("Dumping environment variables to %s", loc)
|
||||
with open(loc, "w") as f:
|
||||
f.write("#!/usr/bin/env bash\n")
|
||||
for k, v in cenv.items():
|
||||
|
||||
Reference in New Issue
Block a user