mirror of
https://github.com/Chia-Network/chia-blockchain.git
synced 2024-10-26 20:40:51 +03:00
185 lines
6.0 KiB
Python
185 lines
6.0 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import dataclasses
|
|
import json
|
|
import logging
|
|
import types
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List
|
|
|
|
import testconfig
|
|
|
|
root_path = Path(__file__).parent.absolute()
|
|
project_root_path = root_path.parent
|
|
|
|
|
|
def skip(path: Path) -> bool:
|
|
return any(part.startswith(("_", ".")) for part in path.parts)
|
|
|
|
|
|
def subdirs(per: str) -> List[Path]:
|
|
if per == "directory":
|
|
glob_pattern = "**/"
|
|
elif per == "file":
|
|
glob_pattern = "**/test_*.py"
|
|
else:
|
|
raise Exception(f"Unrecognized per: {per!r}")
|
|
|
|
paths = [path for path in root_path.rglob(glob_pattern) if not skip(path=path)]
|
|
|
|
if per == "directory":
|
|
filtered_paths = []
|
|
for path in paths:
|
|
relative_path = path.relative_to(root_path)
|
|
logging.info(f"Considering: {relative_path}")
|
|
if len([f for f in path.glob("test_*.py")]) == 0:
|
|
logging.info(f"Skipping {relative_path}: no tests collected")
|
|
continue
|
|
|
|
filtered_paths.append(path)
|
|
|
|
paths = filtered_paths
|
|
|
|
return sorted(paths)
|
|
|
|
|
|
def module_dict(module: types.ModuleType) -> Dict[str, Any]:
|
|
return {k: v for k, v in module.__dict__.items() if not k.startswith("_") and k != "annotations"}
|
|
|
|
|
|
def dir_config(dir: Path) -> Dict[str, Any]:
|
|
import importlib
|
|
|
|
module_name = ".".join([*dir.relative_to(root_path).parts, "config"])
|
|
try:
|
|
return module_dict(importlib.import_module(module_name))
|
|
except ModuleNotFoundError:
|
|
return {}
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class SpecifiedDefaultsError(Exception):
|
|
overlap: Dict[str, Any]
|
|
|
|
def __post_init__(self) -> None:
|
|
super().__init__()
|
|
|
|
|
|
# Overwrite with directory specific values
|
|
def update_config(parent: Dict[str, Any], child: Dict[str, Any]) -> Dict[str, Any]:
|
|
if child is None:
|
|
return parent
|
|
conf = child
|
|
|
|
# avoid manual configuration set to default values
|
|
common_keys = set(parent.keys()).intersection(child.keys())
|
|
specified_defaulted_values = {k: parent[k] for k in common_keys if parent[k] == child[k]}
|
|
if len(specified_defaulted_values) > 0:
|
|
raise SpecifiedDefaultsError(overlap=specified_defaulted_values)
|
|
|
|
for k, v in parent.items():
|
|
if k not in child:
|
|
conf[k] = v
|
|
|
|
return conf
|
|
|
|
|
|
# args
|
|
arg_parser = argparse.ArgumentParser(description="Generate GitHub test matrix configuration")
|
|
arg_parser.add_argument("--per", type=str, choices=["directory", "file"], required=True)
|
|
arg_parser.add_argument("--verbose", "-v", action="store_true")
|
|
arg_parser.add_argument("--only", action="append", default=[])
|
|
arg_parser.add_argument("--duplicates", type=int, default=1)
|
|
arg_parser.add_argument("--timeout-multiplier", type=float, default=1)
|
|
args = arg_parser.parse_args()
|
|
|
|
if args.verbose:
|
|
logging.basicConfig(format="%(asctime)s:%(message)s", level=logging.DEBUG)
|
|
|
|
# main
|
|
if len(args.only) == 0:
|
|
test_paths = subdirs(per=args.per)
|
|
else:
|
|
test_paths = [root_path.joinpath(path) for path in args.only]
|
|
|
|
test_paths = [path for path in test_paths for _ in range(args.duplicates)]
|
|
|
|
configuration = []
|
|
|
|
specified_defaults: Dict[Path, Dict[str, Any]] = {}
|
|
pytest_monitor_enabling_paths: List[Path] = []
|
|
|
|
for path in test_paths:
|
|
if path.is_dir():
|
|
test_files = sorted(path.glob("test_*.py"))
|
|
test_file_paths = [file.relative_to(project_root_path) for file in test_files]
|
|
paths_for_cli = " ".join(path.as_posix() for path in test_file_paths)
|
|
config_path = path
|
|
else:
|
|
paths_for_cli = path.relative_to(project_root_path).as_posix()
|
|
config_path = path.parent
|
|
|
|
try:
|
|
conf = update_config(module_dict(testconfig), dir_config(config_path))
|
|
except SpecifiedDefaultsError as e:
|
|
specified_defaults[root_path.joinpath(config_path, "config.py")] = e.overlap
|
|
continue
|
|
|
|
# TODO: design a configurable system for this
|
|
process_count = {
|
|
"macos": {False: 0, True: 4}.get(conf["parallel"], conf["parallel"]),
|
|
"ubuntu": {False: 0, True: 4}.get(conf["parallel"], conf["parallel"]),
|
|
"windows": {False: 0, True: 3}.get(conf["parallel"], conf["parallel"]),
|
|
}
|
|
pytest_parallel_args = {os: f" -n {count}" for os, count in process_count.items()}
|
|
|
|
enable_pytest_monitor = conf["check_resource_usage"]
|
|
|
|
if enable_pytest_monitor:
|
|
# NOTE: do not use until the hangs are fixed
|
|
# https://github.com/CFMTech/pytest-monitor/issues/53
|
|
# https://github.com/pythonprofilers/memory_profiler/issues/342
|
|
|
|
pytest_monitor_enabling_paths.append(path)
|
|
|
|
for_matrix = {
|
|
"check_resource_usage": conf["check_resource_usage"],
|
|
"enable_pytest_monitor": "-p monitor" if enable_pytest_monitor else "",
|
|
"job_timeout": round(conf["job_timeout"] * args.timeout_multiplier),
|
|
"pytest_parallel_args": pytest_parallel_args,
|
|
"checkout_blocks_and_plots": conf["checkout_blocks_and_plots"],
|
|
"install_timelord": conf["install_timelord"],
|
|
"test_files": paths_for_cli,
|
|
"name": ".".join(path.relative_to(root_path).with_suffix("").parts),
|
|
"legacy_keyring_required": conf.get("legacy_keyring_required", False),
|
|
}
|
|
for_matrix = dict(sorted(for_matrix.items()))
|
|
configuration.append(for_matrix)
|
|
|
|
messages: List[str] = []
|
|
|
|
if len(specified_defaults) > 0:
|
|
message = f"Found {len(specified_defaults)} directories with specified defaults"
|
|
messages.append(message)
|
|
logging.error(f"{message}:")
|
|
for path, overlap in sorted(specified_defaults.items()):
|
|
logging.error(f" {path} : {overlap}")
|
|
|
|
if len(pytest_monitor_enabling_paths) > 0:
|
|
message = f"Found {len(pytest_monitor_enabling_paths)} directories with pytest-monitor enabled"
|
|
messages.append(message)
|
|
logging.error(f"{message}:")
|
|
for path in sorted(pytest_monitor_enabling_paths):
|
|
logging.error(f" {path}")
|
|
|
|
if len(messages) > 0:
|
|
raise Exception("\n".join(messages))
|
|
|
|
configuration_json = json.dumps(configuration)
|
|
|
|
for line in json.dumps(configuration, indent=4).splitlines():
|
|
logging.info(line)
|
|
|
|
print(f"{configuration_json}")
|