From 4a341e2dca32dc0bfb4c4d7aeb8a795868ea4cef Mon Sep 17 00:00:00 2001 From: MaXFeeD Date: Mon, 22 Sep 2025 02:35:20 +0300 Subject: [PATCH] Example (unit test) running support, fixed recursive dependency collection for file configs, project name --- src/icmtoolchain/cli.py | 31 ++++++++++++++++++++++++++++--- src/icmtoolchain/project_graph.py | 11 +++++------ 2 files changed, 33 insertions(+), 9 deletions(-) diff --git a/src/icmtoolchain/cli.py b/src/icmtoolchain/cli.py index 8c090a4..52a626a 100644 --- a/src/icmtoolchain/cli.py +++ b/src/icmtoolchain/cli.py @@ -1,5 +1,7 @@ import sys from itertools import tee +from os import listdir +from os.path import dirname, isdir, isfile, join from typing import TYPE_CHECKING, MutableSequence, MutableSet, Optional from .project_graph import ProjectEdge, ProjectGraph @@ -95,7 +97,11 @@ def run(argv: Optional[MutableSequence[str]] = None): show_available_tasks() exit(0) if "--cli-test" in argv: - run_test() + run_cli_test() + exit(0) + if "--example" in argv: + example_offset = argv.index("--example") + run_example_test(argv[example_offset + 1] if len(argv) > example_offset + 1 else "complex") exit(0) from time import time @@ -154,7 +160,7 @@ def run(argv: Optional[MutableSequence[str]] = None): startup_millis = time() - startup_millis success(f"Tasks successfully completed in {startup_millis:.2f}s!") -def run_test(): +def run_cli_test(): import asyncio from itertools import cycle from random import randint, random @@ -346,5 +352,24 @@ def run_test(): except (KeyboardInterrupt, EOFError): pretty_print("Tasks stopped gracefully.") +def run_example_test(name: str): + examples_directory = join(dirname(__file__), "..", "..", "examples") + if not isdir(examples_directory): + failure("Examples directory is not available.") + return + example_executable = join(examples_directory, name, "__main__.py") + if not isfile(example_executable): + failure(f"No such example {name!r}. It should be one of: {', '.join(listdir(examples_directory))}.") + return + + from importlib.util import module_from_spec, spec_from_file_location + example_name = f"{__name__[:__name__.rindex('.')]}.examples.{name}" + example_spec = spec_from_file_location(example_name, example_executable) + assert example_spec + example_module = module_from_spec(example_spec) + sys.modules[example_name] = example_module + assert example_spec.loader + example_spec.loader.exec_module(example_module) + if __name__ == "__main__": - run_test() + run_cli_test() diff --git a/src/icmtoolchain/project_graph.py b/src/icmtoolchain/project_graph.py index 04bdbd4..0c132a5 100644 --- a/src/icmtoolchain/project_graph.py +++ b/src/icmtoolchain/project_graph.py @@ -8,7 +8,7 @@ from typing import (Any, Callable, Dict, Final, List, MutableSequence, MutableSet, Optional, Tuple, Type, Union) from . import GLOBALS -from .config import Config +from .config import Config, FileConfig from .language import MakeDataConfig from .shell import abort, attention, confirm_prompt, pretty_print from .utils import ensure_not_whitespace, remove_tree @@ -101,7 +101,7 @@ class ProjectEdge: return f"ProjectEdge(project={self.project}, dependencies=({', '.join(str(edge.project) for edge in self.dependencies)}), references=({', '.join(str(edge.project) for edge in self.references)}))" def __str__(self) -> str: - return f"Project {self.project} ({len(self.dependencies)} dependencies)" + return f"Project {basename(self.project.directory) if isinstance(self.project, FileConfig) else self.project} ({len(self.dependencies)} dependencies)" class ProjectGraph(Dict[Union[MakeDataConfig, Artifact], ProjectEdge]): def __init__(self, project: MakeDataConfig) -> None: @@ -169,10 +169,9 @@ class ProjectGraph(Dict[Union[MakeDataConfig, Artifact], ProjectEdge]): if not parent: parent = self.root for dependency in config.iterate_dependencies(): - child = self.obtain_edge(dependency) - if isinstance(dependency, MakeDataConfig): - self.collect_dependencies(dependency, child) - self.depend_on(parent, child) + if isinstance(dependency, MakeDataConfig) and not dependency in self: + self.collect_dependencies(dependency, self.obtain_edge(dependency)) + self.depend_on(parent, self.obtain_edge(dependency)) def resolve_dependencies(self, edge: Optional[ProjectEdge] = None, keep_unused: bool = False, traversed_references: Optional[MutableSet[ProjectEdge]] = None) -> MutableSet[ProjectEdge]: if not edge: