Archives
Trending
Support
Login
clear text
XML
Django
JavaScript
MATLAB
C
C++
Python
SQL
Shell
Markdown
YAML
JSON
CSS
PHP
Java
Ruby
Go
Rust
Swift
Kotlin
TypeScript
Perl
Lua
R
Scala
Haskell
Groovy
Dart
Clojure
VB.NET
Objective-C
PowerShell
Bash
CoffeeScript
Verilog
import argparse import sys from collections import defaultdict from pathlib import Path from typing import Iterable, List, Set, NoReturn, Generator, Optional from glotter.batch import batch from glotter.download import download, remove_images from glotter.source import get_sources from glotter.settings import Settings from glotter.test import test RUN_EVERYTHING_PATHS = { Path(".glotter.yml"), Path("repo-config.yml"), Path("pyproject.toml"), Path("poetry.lock"), Path(".github/workflows/test-suite.yml"), Path("scripts/run_tests.py"), } def remove_deleted_file_changes(files_changed: Iterable[str]) -> Set[Path]: return {Path(path) for path in files_changed if Path(path).exists()} def should_run_everything(event: Optional[str], paths_changed: Set[Path]) -> bool: return event == "schedule" or bool(paths_changed & RUN_EVERYTHING_PATHS) def run_everything(parsed_args: argparse.Namespace) -> NoReturn: print("=== Running all tests ===\n", flush=True) glotter_args = argparse.Namespace( num_batches=parsed_args.num_batches, batch=None, parallel=parsed_args.parallel, remove=parsed_args.remove, ) batch(glotter_args) def should_run_languages(paths_changed: Set[Path]) -> bool: return any(path.name == "testinfo.yml" for path in paths_changed) def run_languages(paths_changed: Set[Path], parsed_args: argparse.Namespace) -> NoReturn: languages = _get_languages(paths_changed) print(f"=== Running tests for {', '.join(languages)} ===\n", flush=True) exit_code = 0 for language_set in _do_batches(languages, parsed_args): test_args = argparse.Namespace( source=None, project=None, language=language_set, parallel=parsed_args.parallel ) try: test(test_args) except SystemExit as e: exit_code = exit_code or int(e.code) sys.exit(exit_code) def should_run_sample_programs(paths_changed: Set[Path]) -> bool: return any(path.name != "README.md" for path in paths_changed) def run_sample_programs(paths_changed: Set[Path], parsed_args: argparse.Namespace) -> NoReturn: print( f"=== Running tests for {', '.join(sorted(str(path) for path in paths_changed))} ===\n", flush=True, ) languages = _get_languages(paths_changed) all_sources = get_sources(Settings().source_root) abs_paths_changed = {path.absolute() for path in paths_changed} exit_code = 0 for language_set in _do_batches(languages, parsed_args): # Get selected language/project combinations for this language set languages_and_projects = sorted( (source.language.lower(), project) for project, sources in all_sources.items() for source in sources if source.language.lower() in language_set and Path(source.full_path) in abs_paths_changed ) # Get list of selected projects for each language sample_programs = defaultdict(list) for language, project in languages_and_projects: sample_programs[language].append(project) # Test each selected project for each language for language, projects in sample_programs.items(): for project in projects: test_args = argparse.Namespace( source=None, project=project, language=language, parallel=parsed_args.parallel ) try: test(test_args) except SystemExit as e: exit_code = exit_code or int(e.code) sys.exit(exit_code) def _get_languages(paths_changed: Set[Path]) -> List[str]: all_sources = get_sources(Settings().source_root) dirs_changed = {path.parent.absolute() for path in paths_changed} return sorted( { source.language.lower() for sources in all_sources.values() for source in sources if Path(source.path) in dirs_changed } ) def _do_batches( languages: List[str], parsed_args: argparse.Namespace ) -> Generator[Set[str], None, None]: num_languages = len(languages) num_batches = min(num_languages, parsed_args.num_batches) for n in range(0, num_batches): # Get languages for this batch batch_args = argparse.Namespace( source=None, project=None, language=set( languages[ (n * num_languages // num_batches) : ((n + 1) * num_languages // num_batches) ] ), parallel=parsed_args.parallel, ) # Download images for this batch _display_batch("Downloading images", n, num_batches) containers = download(batch_args) # Tell caller languages for this batch yield batch_args.language # If removing images, remove images for this batch if parsed_args.remove: _display_batch("Removing images", n, num_batches) remove_images(containers, parsed_args.parallel) def _display_batch(prefix, n, num_batches) -> None: print(f"\n*** {prefix} for batch {n + 1} of {num_batches} ***", flush=True) def main() -> NoReturn: parser = argparse.ArgumentParser() parser.add_argument("--event", help="GitHub event") parser.add_argument("--num-batches", type=int, help="number of glotter batches", required=True) parser.add_argument("--parallel", action="store_true", help="run glotter in parallel") parser.add_argument( "--remove", action="store_true", help="remove docker images after each batch is finished" ) parser.add_argument("files_changed", nargs="*", help="files that have changed") parsed_args = parser.parse_args() paths_changed = remove_deleted_file_changes(parsed_args.files_changed) if should_run_everything(parsed_args.event, paths_changed): run_everything(parsed_args) if should_run_languages(paths_changed): run_languages(paths_changed, parsed_args) if should_run_sample_programs(paths_changed): run_sample_programs(paths_changed, parsed_args) print("Nothing to do") sys.exit(0) if __name__ == "__main__": main()
Mark as private
for 30 minutes
for 6 hours
for 1 day
for 1 week
for 1 month
for 1 year