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
|
#!/usr/bin/env nix-shell
#!nix-shell -p nix -p python3 -i python
import argparse
import multiprocessing
import re
import subprocess
import sys
from pathlib import Path
from typing import List, Tuple
TEST_ROOT = Path(__file__).resolve().parent
ROOT = TEST_ROOT.parent
GREEN = "\033[92m"
RED = "\033[91m"
RESET = "\033[0m"
def parse_readme() -> List[str]:
profiles = set()
with open(ROOT.joinpath("README.md")) as f:
for line in f:
results = re.findall(r"<nixos-hardware/[^>]+>", line)
profiles.update(results)
return list(profiles)
def build_profile(profile: str) -> Tuple[str, subprocess.CompletedProcess]:
# Hard-code this for now until we have enough other architectures to care about this.
system = "x86_64-linux"
if "raspberry-pi/2" in profile:
system = "armv7l-linux"
cmd = [
"nix-build",
"-I",
f"nixos-hardware={ROOT}",
"--dry-run",
"--show-trace",
"build-profile.nix",
"--system",
system,
"--arg",
"profile",
profile,
]
res = subprocess.run(
cmd, cwd=TEST_ROOT, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
)
return (profile, res)
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Run hardware tests")
parser.add_argument(
"--jobs",
type=int,
default=multiprocessing.cpu_count(),
help="Number of parallel evaluations."
"If set to 1 it disable multi processing (suitable for debugging)",
)
parser.add_argument("profiles", nargs="*")
return parser.parse_args()
def main() -> None:
args = parse_args()
if len(args.profiles) == 0:
profiles = parse_readme()
else:
profiles = args.profiles
failed_profiles = []
def eval_finished(args: Tuple[str, subprocess.CompletedProcess]) -> None:
profile, res = args
if res.returncode == 0:
print(f"{GREEN}OK {profile}{RESET}")
else:
print(f"{RED}FAIL {profile}:{RESET}", file=sys.stderr)
if res.stdout != "":
print(f"{RED}{res.stdout.rstrip()}{RESET}", file=sys.stderr)
print(f"{RED}{res.stderr.rstrip()}{RESET}", file=sys.stderr)
failed_profiles.append(profile)
if len(profiles) == 0 or args.jobs == 1:
for profile in profiles:
eval_finished(build_profile(profile))
else:
pool = multiprocessing.Pool(processes=args.jobs)
for r in pool.imap(build_profile, profiles):
eval_finished(r)
if len(failed_profiles) > 0:
print(f"\n{RED}The following {len(failed_profiles)} test(s) failed:{RESET}")
for profile in failed_profiles:
print(f"{sys.argv[0]} '{profile}'")
sys.exit(1)
if __name__ == "__main__":
main()
|
