1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
|
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
import shlex
import subprocess
from typing import Iterable, Sequence
from .config import DeviceConfig
@dataclass(frozen=True)
class CommandResult:
argv: tuple[str, ...]
returncode: int
stdout: str
stderr: str
@property
def ok(self) -> bool:
return self.returncode == 0
def public_dict(self, limit: int = 20000) -> dict[str, object]:
stdout, stdout_truncated = truncate(self.stdout, limit)
stderr, stderr_truncated = truncate(self.stderr, limit)
return {
"argv": list(self.argv),
"returncode": self.returncode,
"stdout": stdout,
"stderr": stderr,
"stdout_truncated": stdout_truncated,
"stderr_truncated": stderr_truncated,
}
def truncate(text: str, limit: int) -> tuple[str, bool]:
if len(text) <= limit:
return text, False
return text[:limit] + f"\n[truncated after {limit} characters]", True
def run(
argv: Sequence[str],
*,
cwd: str | Path | None = None,
timeout: int = 60,
) -> CommandResult:
try:
completed = subprocess.run(
list(argv),
cwd=str(cwd) if cwd is not None else None,
text=True,
capture_output=True,
timeout=timeout,
check=False,
)
return CommandResult(
argv=tuple(str(arg) for arg in argv),
returncode=completed.returncode,
stdout=completed.stdout,
stderr=completed.stderr,
)
except subprocess.TimeoutExpired as exc:
return CommandResult(
argv=tuple(str(arg) for arg in argv),
returncode=124,
stdout=exc.stdout or "",
stderr=(exc.stderr or "") + f"\ncommand timed out after {timeout}s",
)
def ssh_argv(device: DeviceConfig, ssh_config: Path | None, remote: str) -> list[str]:
argv = ["ssh"]
if ssh_config:
argv += ["-F", str(ssh_config)]
argv += [device.ssh_target, remote]
return argv
def scp_to_argv(
device: DeviceConfig,
ssh_config: Path | None,
local_path: Path,
remote_path: str,
) -> list[str]:
argv = ["scp"]
if ssh_config:
argv += ["-F", str(ssh_config)]
argv += [str(local_path), f"{device.ssh_target}:{remote_path}"]
return argv
def scp_from_argv(
device: DeviceConfig,
ssh_config: Path | None,
remote_path: str,
local_path: Path,
) -> list[str]:
argv = ["scp"]
if ssh_config:
argv += ["-F", str(ssh_config)]
argv += [f"{device.ssh_target}:{remote_path}", str(local_path)]
return argv
def remote_command(argv: Iterable[str]) -> str:
return shlex.join([str(arg) for arg in argv])
def user_bus_env(device: DeviceConfig) -> list[str]:
return [
"env",
f"XDG_RUNTIME_DIR={device.user_bus_runtime_dir}",
f"DBUS_SESSION_BUS_ADDRESS={device.user_bus_address}",
]
|
