244 lines
8.1 KiB
Python
244 lines
8.1 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import threading
|
|
import time
|
|
import traceback
|
|
import warnings
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
import psutil
|
|
|
|
from ci.util.running_process_manager import RunningProcessManagerSingleton
|
|
from ci.util.test_args import parse_args
|
|
from ci.util.test_commands import run_command
|
|
from ci.util.test_env import (
|
|
dump_thread_stacks,
|
|
get_process_tree_info,
|
|
setup_environment,
|
|
setup_force_exit,
|
|
setup_watchdog,
|
|
)
|
|
from ci.util.test_runner import runner as test_runner
|
|
from ci.util.test_types import (
|
|
FingerprintResult,
|
|
calculate_fingerprint,
|
|
process_test_flags,
|
|
)
|
|
|
|
|
|
_CANCEL_WATCHDOG = threading.Event()
|
|
|
|
_TIMEOUT_EVERYTHING = 600
|
|
|
|
if os.environ.get("_GITHUB"):
|
|
_TIMEOUT_EVERYTHING = 1200 # Extended timeout for GitHub Linux builds
|
|
print(
|
|
f"GitHub Windows environment detected - using extended timeout: {_TIMEOUT_EVERYTHING} seconds"
|
|
)
|
|
|
|
|
|
def make_watch_dog_thread(
|
|
seconds: int,
|
|
) -> threading.Thread: # 60 seconds default timeout
|
|
def watchdog_timer() -> None:
|
|
time.sleep(seconds)
|
|
if _CANCEL_WATCHDOG.is_set():
|
|
return
|
|
|
|
warnings.warn(f"Watchdog timer expired after {seconds} seconds.")
|
|
|
|
dump_thread_stacks()
|
|
print(f"Watchdog timer expired after {seconds} seconds - forcing exit")
|
|
|
|
# Dump outstanding running processes (if any)
|
|
try:
|
|
RunningProcessManagerSingleton.dump_active()
|
|
except Exception as e:
|
|
print(f"Failed to dump active processes: {e}")
|
|
|
|
traceback.print_stack()
|
|
time.sleep(0.5)
|
|
|
|
os._exit(2) # Exit with error code 2 to indicate timeout (SIGTERM)
|
|
|
|
thr = threading.Thread(target=watchdog_timer, daemon=True, name="WatchdogTimer")
|
|
thr.start()
|
|
return thr
|
|
|
|
|
|
def main() -> None:
|
|
try:
|
|
# Record start time
|
|
start_time = time.time()
|
|
|
|
# Change to script directory first
|
|
os.chdir(Path(__file__).parent)
|
|
|
|
# Parse and process arguments
|
|
args = parse_args()
|
|
|
|
# Force sequential execution to avoid out of memory errors
|
|
args.no_parallel = True
|
|
|
|
args = process_test_flags(args)
|
|
|
|
timeout = _TIMEOUT_EVERYTHING
|
|
# Adjust watchdog timeout based on test configuration
|
|
# Sequential examples compilation can take up to 30 minutes
|
|
if args.examples is not None and args.no_parallel:
|
|
# 35 minutes for sequential examples compilation
|
|
timeout = 2100
|
|
print(
|
|
f"Adjusted watchdog timeout for sequential examples compilation: {timeout} seconds"
|
|
)
|
|
|
|
# Set up watchdog timer
|
|
watchdog = make_watch_dog_thread(seconds=timeout)
|
|
|
|
# Handle --no-interactive flag
|
|
if args.no_interactive:
|
|
os.environ["FASTLED_CI_NO_INTERACTIVE"] = "true"
|
|
os.environ["GITHUB_ACTIONS"] = (
|
|
"true" # This ensures all subprocess also run in non-interactive mode
|
|
)
|
|
|
|
# Handle --interactive flag
|
|
if args.interactive:
|
|
os.environ.pop("FASTLED_CI_NO_INTERACTIVE", None)
|
|
os.environ.pop("GITHUB_ACTIONS", None)
|
|
|
|
# Set up remaining environment based on arguments
|
|
setup_environment(args)
|
|
|
|
# Handle stack trace control
|
|
enable_stack_trace = not args.no_stack_trace
|
|
if enable_stack_trace:
|
|
print("Stack trace dumping enabled for test timeouts")
|
|
else:
|
|
print("Stack trace dumping disabled for test timeouts")
|
|
|
|
# Validate conflicting arguments
|
|
if args.no_interactive and args.interactive:
|
|
print(
|
|
"Error: --interactive and --no-interactive cannot be used together",
|
|
file=sys.stderr,
|
|
)
|
|
sys.exit(1)
|
|
|
|
# Set up fingerprint caching
|
|
cache_dir = Path(".cache")
|
|
cache_dir.mkdir(exist_ok=True)
|
|
fingerprint_file = cache_dir / "fingerprint.json"
|
|
|
|
def write_fingerprint(fingerprint: FingerprintResult) -> None:
|
|
fingerprint_dict = {
|
|
"hash": fingerprint.hash,
|
|
"elapsed_seconds": fingerprint.elapsed_seconds,
|
|
"status": fingerprint.status,
|
|
}
|
|
with open(fingerprint_file, "w") as f:
|
|
json.dump(fingerprint_dict, f, indent=2)
|
|
|
|
def read_fingerprint() -> FingerprintResult | None:
|
|
if fingerprint_file.exists():
|
|
with open(fingerprint_file, "r") as f:
|
|
try:
|
|
data = json.load(f)
|
|
return FingerprintResult(
|
|
hash=data.get("hash", ""),
|
|
elapsed_seconds=data.get("elapsed_seconds"),
|
|
status=data.get("status"),
|
|
)
|
|
except json.JSONDecodeError:
|
|
print("Invalid fingerprint file. Recalculating...")
|
|
return None
|
|
|
|
# Calculate and save fingerprint
|
|
prev_fingerprint = read_fingerprint()
|
|
fingerprint_data = calculate_fingerprint()
|
|
src_code_change = (
|
|
True
|
|
if prev_fingerprint is None
|
|
else fingerprint_data.hash != prev_fingerprint.hash
|
|
)
|
|
write_fingerprint(fingerprint_data)
|
|
|
|
# Run tests using the test runner with sequential example compilation
|
|
# Check if we need to use sequential execution to avoid resource conflicts
|
|
if not args.unit and not args.examples and not args.py and args.full:
|
|
# Full test mode - use sequential execution for example compilation
|
|
from ci.util.running_process import RunningProcess
|
|
from ci.util.test_runner import (
|
|
create_examples_test_process,
|
|
create_python_test_process,
|
|
)
|
|
|
|
# Create Python test process (runs first)
|
|
python_process = create_python_test_process(
|
|
enable_stack_trace=False, full_tests=True
|
|
)
|
|
|
|
# Create examples compilation process with auto_run=False
|
|
examples_process = create_examples_test_process(
|
|
args, not args.no_stack_trace
|
|
)
|
|
examples_process.auto_run = False
|
|
|
|
# Set up callback to start examples compilation after Python tests complete
|
|
def start_examples_compilation():
|
|
print("Python tests completed - starting examples compilation...")
|
|
examples_process.run()
|
|
|
|
python_process.on_complete = start_examples_compilation
|
|
|
|
# Start the Python test process (since auto_run=False)
|
|
python_process.run()
|
|
|
|
if args.verbose:
|
|
with python_process.line_iter(timeout=None) as line_iter:
|
|
for line in line_iter:
|
|
print(line)
|
|
|
|
# Wait for Python tests to complete (which will trigger examples compilation)
|
|
python_return_code = python_process.wait()
|
|
if python_return_code != 0:
|
|
python_stdout = python_process.stdout
|
|
print(f"Python test process stdout: {python_stdout}")
|
|
sys.exit(python_return_code)
|
|
|
|
if args.verbose:
|
|
with examples_process.line_iter(timeout=None) as line_iter:
|
|
for line in line_iter:
|
|
print(line)
|
|
|
|
# Wait for examples compilation to complete
|
|
examples_return_code = examples_process.wait()
|
|
if examples_return_code != 0:
|
|
sys.exit(examples_return_code)
|
|
|
|
print("Sequential test execution completed successfully")
|
|
else:
|
|
# Use normal test runner for other cases
|
|
test_runner(args, src_code_change)
|
|
|
|
# Set up force exit daemon and exit
|
|
daemon_thread = setup_force_exit()
|
|
_CANCEL_WATCHDOG.set()
|
|
|
|
# Print total execution time
|
|
elapsed_time = time.time() - start_time
|
|
print(f"\nTotal execution time: {elapsed_time:.2f} seconds")
|
|
|
|
sys.exit(0)
|
|
|
|
except KeyboardInterrupt:
|
|
sys.exit(130) # Standard Unix practice: 128 + SIGINT's signal number (2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|