rustclamp 1.1.0

A minimal synchronous entrypoint for Clamp applications.
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

#!/usr/bin/env python3
"""Run the coordinated test suite or Kernel microbench with clear status marks."""

import argparse
import os
from pathlib import Path
import re
import subprocess
import sys


ROOT = Path(__file__).resolve().parents[2]
COMMANDS = {
    "test": ["cargo", "test", "--offline", "--locked", "--workspace", "--all-features"],
    "bench": ["cargo", "bench", "--offline", "--locked", "-p", "rustclamp-kernel", "--bench", "resolution"],
}
COLORS = {"green": "32", "yellow": "33", "red": "31", "cyan": "36"}


def style(text, color, mode):
    enabled = mode == "always" or (
        mode == "auto" and "NO_COLOR" not in os.environ and sys.stdout.isatty()
    )
    if not enabled or mode == "never":
        return text
    return f"\033[{COLORS[color]}m{text}\033[0m"


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("mode", choices=COMMANDS)
    parser.add_argument("--color", choices=("auto", "always", "never"), default="auto")
    args = parser.parse_args()
    commands = [COMMANDS[args.mode]]
    if args.mode == "bench":
        commands.append([
            "cargo", "bench", "--offline", "--locked", "-p", "rustclamp-kernel",
            "--bench", "process_projection",
        ])

    suite = "Cargo tests"
    suite_count = 0
    test_count = 0
    failed_tests = 0
    bench_count = 0
    warnings = []
    try:
        return_code = 0
        for command in commands:
            print(style("▶", "cyan", args.color), " ".join(command), flush=True)
            process = subprocess.Popen(
                command,
                cwd=ROOT,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                text=True,
                bufsize=1,
            )
            assert process.stdout is not None
            for line in process.stdout:
                stripped = line.strip()
                bench_line = re.match(r"✓ BENCH (.+)", stripped)
                if args.mode == "bench" and bench_line:
                    bench_count += 1
                    print(style(f"✓ BENCH {bench_line.group(1)}", "green", args.color), flush=True)
                    continue
                print(line, end="", flush=True)
                if stripped.startswith("Running "):
                    suite = stripped.removeprefix("Running ").split(" (", 1)[0]
                elif stripped.startswith("Doc-tests "):
                    suite = stripped
                result = re.search(r"test result: (?:ok|FAILED)\. (\d+) passed; (\d+) failed;", stripped)
                if result:
                    passed, failed = map(int, result.groups())
                    suite_count += 1
                    test_count += passed
                    failed_tests += failed
                    mark = "✗" if failed else "✓"
                    color = "red" if failed else "green"
                    print(style(f"{mark} TEST SUITE {suite}: {passed} passed, {failed} failed", color, args.color), flush=True)
                if "warning:" in stripped.lower():
                    warnings.append(stripped)
            return_code = process.wait()
            if return_code:
                break
    except OSError as error:
        print(style(f"✗ Could not start {args.mode}: {error}", "red", args.color))
        print(style(f"✗ {args.mode.upper()} FAILED", "red", args.color))
        return 1

    if args.mode == "test":
        if return_code or failed_tests:
            print(style(f"✗ TESTS FAILED ({test_count} passed, {failed_tests} failed)", "red", args.color))
            return return_code or 1
        if warnings:
            print(style(f"⚠ TESTS COMPLETED WITH WARNINGS ({test_count} passed, {suite_count} suites)", "yellow", args.color))
            for warning in warnings:
                print(style(f"⚠ {warning}", "yellow", args.color))
            return 0
        print(style(f"✓ TESTS SUCCESSFULLY COMPLETED ({test_count} passed, {suite_count} suites)", "green", args.color))
        return 0

    if return_code or bench_count == 0:
        print(style(f"✗ BENCH FAILED (exit={return_code}, measured_paths={bench_count})", "red", args.color))
        return return_code or 1
    print(style(f"⚠ BENCH COMPLETED WITH WARNINGS ({bench_count} paths measured)", "yellow", args.color))
    print(style("⚠ Single-host microbenchmark; timings are advisory and have no regression threshold.", "yellow", args.color))
    if warnings:
        for warning in warnings:
            print(style(f"⚠ {warning}", "yellow", args.color))
    return 0


if __name__ == "__main__":
    sys.exit(main())