···
from typing import Tuple, Any, Callable, Dict, Iterator, Optional, List, Iterable
from xml.sax.saxutils import XMLGenerator
from colorama import Style
···
···
-
monitor_socket_path: pathlib.Path,
-
shell_socket_path: pathlib.Path,
allow_reboot: bool = False, # TODO: unused, legacy?
···
-
state_dir: pathlib.Path,
-
shared_dir: pathlib.Path,
# We make a copy to not update the current environment
···
-
state_dir: pathlib.Path,
-
shared_dir: pathlib.Path,
-
monitor_socket_path: pathlib.Path,
-
shell_socket_path: pathlib.Path,
self.cmd(monitor_socket_path, shell_socket_path),
···
netBackendArgs: Optional[str] = None,
netFrontendArgs: Optional[str] = None,
-
hda: Optional[Tuple[pathlib.Path, str]] = None,
cdrom: Optional[str] = None,
usb: Optional[str] = None,
bios: Optional[str] = None,
···
the machine lifecycle with the help of a start script / command."""
-
shared_dir: pathlib.Path
-
state_dir: pathlib.Path
-
monitor_path: pathlib.Path
-
shell_path: pathlib.Path
start_command: StartCommand
···
start_command: StartCommand,
keep_vm_state: bool = False,
···
hda_arg: str = args.get("hda", "")
-
hda_arg_path: pathlib.Path = pathlib.Path(hda_arg)
hda = (hda_arg_path, args.get("hdaInterface", ""))
return LegacyStartCommand(
netBackendArgs=args.get("netBackendArgs"),
···
"""Copy a file from the host into the guest via the `shared_dir` shared
among all the VMs (using a temporary directory).
-
host_src = pathlib.Path(source)
-
vm_target = pathlib.Path(target)
with tempfile.TemporaryDirectory(dir=self.shared_dir) as shared_td:
-
shared_temp = pathlib.Path(shared_td)
host_intermediate = shared_temp / host_src.name
-
vm_shared_temp = pathlib.Path("/tmp/shared") / shared_temp.name
vm_intermediate = vm_shared_temp / host_src.name
self.succeed(make_command(["mkdir", "-p", vm_shared_temp]))
···
all the VMs (using a temporary directory).
# Compute the source, target, and intermediate shared file names
-
out_dir = pathlib.Path(os.environ.get("out", os.getcwd()))
-
vm_src = pathlib.Path(source)
with tempfile.TemporaryDirectory(dir=self.shared_dir) as shared_td:
-
shared_temp = pathlib.Path(shared_td)
-
vm_shared_temp = pathlib.Path("/tmp/shared") / shared_temp.name
vm_intermediate = vm_shared_temp / vm_src.name
intermediate = shared_temp / vm_src.name
# Copy the file to the shared directory inside VM
···
-
def clear(path: pathlib.Path) -> pathlib.Path:
-
def create_socket(path: pathlib.Path) -> socket.socket:
s = socket.socket(family=socket.AF_UNIX, type=socket.SOCK_STREAM)
···
-
socket_dir: pathlib.Path
process: subprocess.Popen
···
def __repr__(self) -> str:
return f"<Vlan Nr. {self.nr}>"
-
def __init__(self, nr: int, tmp_dir: pathlib.Path):
self.socket_dir = tmp_dir / f"vde{self.nr}.ctl"
···
-
tmp_dir = pathlib.Path(os.environ.get("TMPDIR", tempfile.gettempdir()))
tmp_dir.mkdir(mode=0o700, exist_ok=True)
with rootlog.nested("start all VLans"):
···
"Using legacy create_machine(), please instantiate the"
"Machine class directly, instead"
-
tmp_dir = pathlib.Path(os.environ.get("TMPDIR", tempfile.gettempdir()))
tmp_dir.mkdir(mode=0o700, exist_ok=True)
if args.get("startCommand"):
···
help="the test script to run",
args = arg_parser.parse_args()
···
from typing import Tuple, Any, Callable, Dict, Iterator, Optional, List, Iterable
from xml.sax.saxutils import XMLGenerator
from colorama import Style
+
from pathlib import Path
···
···
+
monitor_socket_path: Path,
+
shell_socket_path: Path,
allow_reboot: bool = False, # TODO: unused, legacy?
···
# We make a copy to not update the current environment
···
+
monitor_socket_path: Path,
+
shell_socket_path: Path,
self.cmd(monitor_socket_path, shell_socket_path),
···
netBackendArgs: Optional[str] = None,
netFrontendArgs: Optional[str] = None,
+
hda: Optional[Tuple[Path, str]] = None,
cdrom: Optional[str] = None,
usb: Optional[str] = None,
bios: Optional[str] = None,
···
the machine lifecycle with the help of a start script / command."""
start_command: StartCommand
···
start_command: StartCommand,
keep_vm_state: bool = False,
···
hda_arg: str = args.get("hda", "")
+
hda_arg_path: Path = Path(hda_arg)
hda = (hda_arg_path, args.get("hdaInterface", ""))
return LegacyStartCommand(
netBackendArgs=args.get("netBackendArgs"),
···
"""Copy a file from the host into the guest via the `shared_dir` shared
among all the VMs (using a temporary directory).
+
host_src = Path(source)
+
vm_target = Path(target)
with tempfile.TemporaryDirectory(dir=self.shared_dir) as shared_td:
+
shared_temp = Path(shared_td)
host_intermediate = shared_temp / host_src.name
+
vm_shared_temp = Path("/tmp/shared") / shared_temp.name
vm_intermediate = vm_shared_temp / host_src.name
self.succeed(make_command(["mkdir", "-p", vm_shared_temp]))
···
all the VMs (using a temporary directory).
# Compute the source, target, and intermediate shared file names
+
out_dir = Path(os.environ.get("out", os.getcwd()))
with tempfile.TemporaryDirectory(dir=self.shared_dir) as shared_td:
+
shared_temp = Path(shared_td)
+
vm_shared_temp = Path("/tmp/shared") / shared_temp.name
vm_intermediate = vm_shared_temp / vm_src.name
intermediate = shared_temp / vm_src.name
# Copy the file to the shared directory inside VM
···
+
def clear(path: Path) -> Path:
+
def create_socket(path: Path) -> socket.socket:
s = socket.socket(family=socket.AF_UNIX, type=socket.SOCK_STREAM)
···
process: subprocess.Popen
···
def __repr__(self) -> str:
return f"<Vlan Nr. {self.nr}>"
+
def __init__(self, nr: int, tmp_dir: Path):
self.socket_dir = tmp_dir / f"vde{self.nr}.ctl"
···
+
tmp_dir = Path(os.environ.get("TMPDIR", tempfile.gettempdir()))
tmp_dir.mkdir(mode=0o700, exist_ok=True)
with rootlog.nested("start all VLans"):
···
"Using legacy create_machine(), please instantiate the"
"Machine class directly, instead"
+
tmp_dir = Path(os.environ.get("TMPDIR", tempfile.gettempdir()))
tmp_dir.mkdir(mode=0o700, exist_ok=True)
if args.get("startCommand"):
···
help="the test script to run",
args = arg_parser.parse_args()
pyrox.dev