From 5f01fccdc6cee68b8c49fdb4c30f0aa7eb7c3e3d Mon Sep 17 00:00:00 2001 From: Arvid Norberg Date: Thu, 10 Oct 2024 22:56:33 +0200 Subject: [PATCH] [CHIA-598] virtual project structure (#18616) slightly modified subset of Quexington's virtual project structure. All files default to be treated as belonging to chia-blockchain project. No default annotations. no default exclusions. --- .pre-commit-config.yaml | 7 + chia/_tests/tools/test_virtual_project.py | 591 ++++++++++++++++++++++ chia/util/virtual_project_analysis.py | 532 +++++++++++++++++++ virtual_project.yaml | 5 + 4 files changed, 1135 insertions(+) create mode 100644 chia/_tests/tools/test_virtual_project.py create mode 100644 chia/util/virtual_project_analysis.py create mode 100644 virtual_project.yaml diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index efceeee634..7e7afd4e9d 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -71,6 +71,13 @@ repos: entry: ./activated.py python tools/chialispp.py . language: system pass_filenames: false + - repo: local + hooks: + - id: virtual_project_dependency_cycles + name: Check for dependency cycles in project packages + entry: ./activated.py python chia/util/virtual_project_analysis.py print_cycles --directory chia --config virtual_project.yaml + language: system + pass_filenames: false - repo: local hooks: - id: build mypy.ini diff --git a/chia/_tests/tools/test_virtual_project.py b/chia/_tests/tools/test_virtual_project.py new file mode 100644 index 0000000000..1b4bc5d37d --- /dev/null +++ b/chia/_tests/tools/test_virtual_project.py @@ -0,0 +1,591 @@ +from __future__ import annotations + +import textwrap +from pathlib import Path +from typing import Any, Callable, Dict, List + +import click +import pytest +import yaml +from click.testing import CliRunner + +from chia.util.virtual_project_analysis import ( + Annotation, + ChiaFile, + Config, + DirectoryParameters, + File, + Package, + build_dependency_graph, + build_virtual_dependency_graph, + cli, + config, + find_cycles, + parse_file_or_package, +) + + +@pytest.mark.parametrize( + "file_string, expected, annotated", + [ + ("# Package: example\n# Some other comment", "example", True), + ("# Some comment\n# Some other comment", "chia-blockchain", False), + ], +) +def test_parse_annotation(file_string: str, expected: str, annotated: bool) -> None: + """ + Test that parse returns an Annotation instance for a valid annotation or + raises ValueError for an invalid one. + """ + annotation = Annotation.parse(file_string) + assert isinstance(annotation, Annotation) + assert annotation.package == expected + assert annotation.is_annotated == annotated + + +# Temporary directory fixture to create test files +@pytest.fixture +def create_test_file(tmp_path: Path) -> Callable[[str, str], Path]: + def _create_test_file(name: str, content: str) -> Path: + file_path = tmp_path / name + file_path.write_text(content, encoding="utf-8") + return file_path + + return _create_test_file + + +def test_parse_with_annotation(create_test_file: Callable[[str, str], Path]) -> None: + """Test parsing a file that contains a valid annotation.""" + file_content = "# Package: test_package\n# Some other comment" + test_file = create_test_file("annotated_file.txt", file_content) + + parsed_file = ChiaFile.parse(test_file) + + assert parsed_file.path == test_file + assert isinstance(parsed_file.annotations, Annotation) + assert parsed_file.annotations.package == "test_package" + + +def test_parse_without_annotation(create_test_file: Callable[[str, str], Path]) -> None: + """Test parsing a file that does not contain any annotations.""" + file_content = "# Some comment\n# Some other comment" + test_file = create_test_file("non_annotated_file.txt", file_content) + + parsed_file = ChiaFile.parse(test_file) + + assert parsed_file.path == test_file + assert not parsed_file.annotations.is_annotated + + +# This test is optional and can be adapted based on expected behavior for non-existent files +def test_parse_nonexistent_file() -> None: + """Test attempting to parse a non-existent file.""" + with pytest.raises(FileNotFoundError): + ChiaFile.parse(Path("/path/to/nonexistent/file.txt")) + + +# Helper function to create a non-empty Python file +def create_python_file(dir_path: Path, name: str, content: str) -> Path: + file_path = dir_path / name + file_path.write_text(content, encoding="utf-8") + return file_path + + +# Helper function to create an empty Python file +def create_empty_python_file(dir_path: Path, name: str) -> Path: + file_path = dir_path / name + file_path.touch() + return file_path + + +def test_gather_non_empty_python_files(tmp_path: Path) -> None: + # Set up directory structure + dir_path = tmp_path / "test_dir" + dir_path.mkdir() + excluded_dir = tmp_path / "excluded_dir" + excluded_dir.mkdir() + + # Create test files + non_empty_file = create_python_file(dir_path, "non_empty.py", "print('Hello World')") + create_empty_python_file(dir_path, "empty.py") + create_python_file(excluded_dir, "excluded.py", "print('Hello World')") + + # Initialize DirectoryParameters with excluded paths + dir_params = DirectoryParameters(dir_path=dir_path, excluded_paths=[excluded_dir]) + + # Perform the test + python_files = dir_params.gather_non_empty_python_files() + + # Assertions + assert len(python_files) == 1 # Only one non-empty Python file should be found + assert python_files[0].path == non_empty_file # The path of the gathered file should match the non-empty file + + +def test_gather_with_nested_directories_and_exclusions(tmp_path: Path) -> None: + # Set up directory structure + base_dir = tmp_path / "base_dir" + base_dir.mkdir() + nested_dir = base_dir / "nested_dir" + nested_dir.mkdir() + excluded_dir = base_dir / "excluded_dir" + excluded_dir.mkdir() + + # Create test files + nested_file = create_python_file(nested_dir, "nested.py", "print('Hello World')") + create_empty_python_file(nested_dir, "nested_empty.py") + create_python_file(excluded_dir, "excluded.py", "print('Hello World')") + + # Initialize DirectoryParameters without excluded paths + dir_params = DirectoryParameters(dir_path=base_dir, excluded_paths=[excluded_dir]) + + # Perform the test + python_files = dir_params.gather_non_empty_python_files() + + # Assertions + assert len(python_files) == 1 # Only the non-empty Python file in the nested directory should be found + assert python_files[0].path == nested_file # The path of the gathered file should match the nested non-empty file + + +def test_find_missing_annotations(tmp_path: Path) -> None: + # Set up directory structure + dir_path = tmp_path / "test_dir" + dir_path.mkdir() + + # Create test files + create_python_file(dir_path, "non_empty.py", "print('Hello World')") + + # Run the command + runner = CliRunner() + result = runner.invoke(cli, ["find_missing_annotations", "--directory", str(dir_path)]) + assert result.output == f"{dir_path / 'non_empty.py'}\n" + + # Rewrite file to have annotation + create_python_file(dir_path, "non_empty.py", "# Package: misc\n") + + # Run the command again with no results + result = runner.invoke(cli, ["find_missing_annotations", "--directory", str(dir_path)]) + assert result.output == "" + + +def test_parse_file_or_package() -> None: + assert parse_file_or_package("example.py") == File(Path("example.py")) + assert parse_file_or_package("example.py (extra info)") == File(Path("example.py")) + assert parse_file_or_package("(package_name)") == Package("package_name") + assert parse_file_or_package("package_name") == Package("package_name") + assert parse_file_or_package("package_name(") == Package("package_name(") + assert parse_file_or_package("(package_name") == Package("(package_name") + assert parse_file_or_package("package_name)") == Package("package_name)") + + +@pytest.fixture +def chia_package_structure(tmp_path: Path) -> Path: + base_dir = tmp_path / "chia_project" + base_dir.mkdir() + chia_dir = base_dir / "chia" + chia_dir.mkdir() + + # Create some files within the chia package + create_python_file(chia_dir, "module1.py", "def func1(): pass") + create_python_file(chia_dir, "module2.py", "def func2(): pass\nfrom chia.module1 import func1") + create_python_file(chia_dir, "module3.py", "def func3(): pass\nimport chia.module2") + + return chia_dir + + +def test_build_dependency_graph(chia_package_structure: Path) -> None: + chia_dir = chia_package_structure + dir_params = DirectoryParameters(dir_path=chia_dir) + graph = build_dependency_graph(dir_params) + assert chia_dir / "module1.py" in graph + assert chia_dir / "module2.py" in graph + assert chia_dir / "module3.py" in graph + assert chia_dir / "module1.py" in graph[chia_dir / "module2.py"] + assert chia_dir / "module2.py" in graph[chia_dir / "module3.py"] + + +def test_print_dependency_graph(chia_package_structure: Path) -> None: + # Run the command + runner = CliRunner() + result = runner.invoke(cli, ["print_dependency_graph", "--directory", str(chia_package_structure)]) + assert "module1.py" in result.output + assert "module2.py" in result.output + assert "module3.py" in result.output + + +# Mock the build_dependency_graph function to control its output +def mock_build_dependency_graph(dir_params: DirectoryParameters) -> Dict[Path, List[Path]]: + return { + Path("/path/to/package1/module1.py"): [ + Path("/path/to/package2/module2.py"), + Path("/path/to/package3/module3.py"), + ], + Path("/path/to/package2/module2.py"): [], + Path("/path/to/package3/module3.py"): [Path("/path/to/package2/module2.py")], + } + + +# Helper function to simulate ChiaFile.parse for testing +def mock_chia_file_parse(path: Path) -> ChiaFile: + annotations_map = { + Path("/path/to/package1/module1.py"): Annotation("Package1", True), + Path("/path/to/package2/module2.py"): Annotation("Package2", True), + Path("/path/to/package3/module3.py"): Annotation("Package3", True), + } + return ChiaFile(path=Path(path), annotations=annotations_map.get(path, Annotation("chia-blockchain", False))) + + +@pytest.fixture +def prepare_mocks(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("chia.util.virtual_project_analysis.build_dependency_graph", mock_build_dependency_graph) + monkeypatch.setattr("chia.util.virtual_project_analysis.ChiaFile.parse", mock_chia_file_parse) + + +def test_build_virtual_dependency_graph(prepare_mocks: None) -> None: + dir_params = DirectoryParameters(dir_path=Path("/path/to/package1")) + virtual_graph = build_virtual_dependency_graph(dir_params) + + assert "Package2" in virtual_graph["Package1"] + assert "Package3" in virtual_graph["Package1"] + assert virtual_graph["Package2"] == [] + assert "Package2" in virtual_graph["Package3"] + + +def test_print_virtual_dependency_graph(tmp_path: Path) -> None: + chia_dir = tmp_path / "chia" + chia_dir.mkdir() + + # Create some files within the chia package + create_python_file(chia_dir, "module1.py", "# Package: one\ndef func1(): pass") + create_python_file(chia_dir, "module2.py", "# Package: two\ndef func2(): pass\nfrom chia.module1 import func1") + create_python_file(chia_dir, "module3.py", "# Package: three\ndef func3(): pass\nimport chia.module2") + + # Run the command + runner = CliRunner() + result = runner.invoke(cli, ["print_virtual_dependency_graph", "--directory", str(chia_dir)]) + assert "one" in result.output + assert "two" in result.output + assert "three" in result.output + + +# Helper function to simulate ChiaFile.parse for testing +def mock_chia_file_parse2(path: Path) -> ChiaFile: + annotations_map = { + Path("/path/to/package1/module1.py"): Annotation("Package1", True), + Path("/path/to/package2/module2.py"): Annotation("Package2", True), + Path("/path/to/package3/module3.py"): Annotation("Package1", True), + } + return ChiaFile(path=Path(path), annotations=annotations_map.get(path, Annotation("chia-blockchain", False))) + + +@pytest.fixture +def prepare_mocks2(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("chia.util.virtual_project_analysis.ChiaFile.parse", mock_chia_file_parse2) + + +def test_cycle_detection(prepare_mocks2: None) -> None: + # Example graph with a simple cycle + graph: Dict[Path, List[Path]] = { + Path("/path/to/package1/module1.py"): [Path("/path/to/package2/module2.py")], + Path("/path/to/package2/module2.py"): [Path("/path/to/package3/module3.py")], # Cycle here + Path("/path/to/package3/module3.py"): [], + } + cycles = find_cycles( + graph, + build_virtual_dependency_graph(DirectoryParameters(dir_path=Path("path")), existing_graph=graph), + excluded_paths=[], + ignore_cycles_in=[], + ignore_specific_files=[], + ignore_specific_edges=[], + ) + # \path\to\package1\module1.py (Package1) -> \path\to\package2\module2.py (Package2) -> (Package1) + # \path\to\package2\module2.py (Package2) -> \path\to\package3\module3.py (Package1) -> (Package2) + assert len(cycles) == 2 + + +def test_print_cycles(tmp_path: Path) -> None: + chia_dir = tmp_path / "chia" + chia_dir.mkdir() + + # Create some files within the chia package + create_python_file(chia_dir, "module1.py", "# Package: one\ndef func1(): pass\nfrom chia.module2 import func2") + create_python_file(chia_dir, "module2.py", "# Package: two\ndef func2(): pass\nfrom chia.module3 import func3") + create_python_file(chia_dir, "module3.py", "# Package: one\ndef func3(): pass\n") + + # Run the command + runner = CliRunner() + result = runner.invoke(cli, ["print_cycles", "--directory", str(chia_dir)]) + assert "module1.py (one) -> " in result.output + + +def test_check_config(tmp_path: Path) -> None: + chia_dir = tmp_path / "chia" + chia_dir.mkdir() + + # Create some files within the chia package + create_python_file( + chia_dir, + "module1.py", + textwrap.dedent( + """ + # Package: one + def func1(): pass + from chia.module2 import func2 + """ + ), + ) + create_python_file( + chia_dir, + "module1b.py", + textwrap.dedent( + """ + # Package: one + def func1b(): pass + """ + ), + ) + create_python_file( + chia_dir, + "module2.py", + textwrap.dedent( + """ + # Package: two + def func2(): pass + from chia.module3 import func3 + from chia.module1b import func1b + """ + ), + ) + create_python_file( + chia_dir, + "module3.py", + textwrap.dedent( + """ + # Package: three + def func3(): pass + """ + ), + ) + + # Run the command + runner = CliRunner() + result = runner.invoke( + cli, + [ + "check_config", + "--directory", + str(chia_dir), + "--ignore-cycles-in", + "three", + "--ignore-specific-file", + str(chia_dir / "module3.py"), + "--ignore-specific-edge", + str(chia_dir / "module2.py") + " -> " + str(chia_dir / "module3.py"), + "--ignore-specific-edge", + str(chia_dir / "module2.py") + " -> " + str(chia_dir / "module1b.py"), + ], + ) + assert " module three ignored but no cycles were found" in result.output + assert f" file {str(chia_dir / 'module3.py')} ignored but no cycles were found" in result.output + assert ( + f"edge {str(chia_dir / 'module2.py') + ' -> ' + str(chia_dir / 'module3.py')} ignored but no cycles were found" + in result.output + ) + assert ( + f"edge {str(chia_dir / 'module2.py') + ' -> ' + str(chia_dir / 'module1b.py')} ignored but no cycles were found" + not in result.output + ) + + +def test_excluded_paths_handling(prepare_mocks2: None) -> None: + # Graph where module2.py is excluded + graph = { + Path("/path/to/package1/module1.py"): [Path("/path/to/package2/module2.py")], + Path("/path/to/package2/module2.py"): [Path("/path/to/package1/module1.py")], + } + cycles = find_cycles( + graph, + build_virtual_dependency_graph(DirectoryParameters(dir_path=Path("path")), existing_graph=graph), + excluded_paths=[Path("/path/to/package2/module2.py")], + ignore_cycles_in=[], + ignore_specific_files=[], + ignore_specific_edges=[], + ) + assert len(cycles) == 0 # No cycles due to exclusion + + +def test_ignore_cycles_in_specific_packages(prepare_mocks2: None) -> None: + graph: Dict[Path, List[Path]] = { + Path("/path/to/package1/module1.py"): [Path("/path/to/package2/module2.py")], + Path("/path/to/package2/module2.py"): [Path("/path/to/package3/module3.py")], + Path("/path/to/package3/module3.py"): [], + } + # Assuming module1.py and module3.py belong to Package1, which is ignored + cycles = find_cycles( + graph, + build_virtual_dependency_graph(DirectoryParameters(dir_path=Path("path")), existing_graph=graph), + excluded_paths=[], + ignore_cycles_in=["Package1"], + ignore_specific_files=[], + ignore_specific_edges=[], + ) + assert len(cycles) == 1 # Cycles in Package1 are ignored + + +def test_ignore_cycles_with_specific_edges(monkeypatch: pytest.MonkeyPatch) -> None: + def _mock_chia_file_parse(path: Path) -> ChiaFile: + annotations_map = { + Path("/path/to/package1/module1a.py"): Annotation("Package1", True), + Path("/path/to/package2/module2.py"): Annotation("Package2", True), + Path("/path/to/package3/module3.py"): Annotation("Package3", True), + Path("/path/to/package1/module1b.py"): Annotation("Package1", True), + } + return ChiaFile(path=Path(path), annotations=annotations_map.get(path, Annotation("chia-blockchain", False))) + + monkeypatch.setattr("chia.util.virtual_project_analysis.ChiaFile.parse", _mock_chia_file_parse) + + graph = { + Path("/path/to/package1/module1a.py"): [Path("/path/to/package2/module2.py")], + Path("/path/to/package2/module2.py"): [Path("/path/to/package3/module3.py")], + Path("/path/to/package3/module3.py"): [Path("/path/to/package1/module1b.py")], + Path("/path/to/package1/module1b.py"): [], + } + virtual_graph = build_virtual_dependency_graph(DirectoryParameters(dir_path=Path("path")), existing_graph=graph) + cycles = find_cycles( + graph, + virtual_graph, + excluded_paths=[], + ignore_cycles_in=[], + ignore_specific_files=[], + ignore_specific_edges=[ + (File(Path("/path/to/package3/module3.py")), File(Path("/path/to/package2/module2.py"))), + (Package("Package3"), Package("Package2")), + ], + ) + assert len(cycles) == 0 + + +def test_ignore_cycles_with_specific_files(prepare_mocks2: None) -> None: + graph = { + Path("/path/to/package1/module1.py"): [Path("/path/to/package2/module2.py")], + Path("/path/to/package2/module2.py"): [Path("/path/to/package3/module3.py")], # Cycle here + } + cycles = find_cycles( + graph, + build_virtual_dependency_graph(DirectoryParameters(dir_path=Path("path")), existing_graph=graph), + excluded_paths=[], + ignore_cycles_in=[], + ignore_specific_files=[Path("/path/to/package1/module1.py"), Path("/path/to/package2/module2.py")], + ignore_specific_edges=[], + ) + assert len(cycles) == 0 + + +# Sample function to use with the decorator for testing +@click.command("blah") +@config +def sample_function(config: Config) -> None: + print(config) + + +# Helper function to create a temporary YAML configuration file +@pytest.fixture +def create_yaml_config(tmp_path: Path) -> Callable[[Dict[str, Any]], Path]: + def _create_yaml_config(content: Dict[str, Any]) -> Path: + path = tmp_path / "config.yaml" + with open(path, "w") as f: + yaml.dump(content, f) + return path + + return _create_yaml_config + + +def test_config_with_yaml(create_yaml_config: Callable[[Dict[str, Any]], Path]) -> None: + # Create a temporary YAML configuration file + yaml_config = { + "exclude_paths": ["path/to/exclude"], + "ignore": { + "packages": ["ignored.package"], + "files": ["ignored_file.py"], + "edges": ["ignored_parent -> ignored_child"], + }, + } + config_path = create_yaml_config(yaml_config) + + runner = CliRunner() + + # Invoke the CLI with the --config option + result = runner.invoke(sample_function, ["--directory", ".", "--config", str(config_path)]) + + # Check if the command ran successfully + assert result.exit_code == 0 + + # Verify the config object created by the decorator + config = result.output + path_type = config_path.__class__.__name__ + assert config == ( + "Config(" + "directory_parameters=DirectoryParameters(" + f"dir_path={path_type}('.'), " + f"excluded_paths=[{path_type}('path/to/exclude')]" + "), " + "ignore_cycles_in=['ignored.package'], " + f"ignore_specific_files=[{path_type}('ignored_file.py')], " + f"ignore_specific_edges=[(Package(name='ignored_child', is_file=False), " + f"Package(name='ignored_parent', is_file=False))]" + ")\n" + ) + + +def test_parse_edges(tmp_path: Path) -> None: + chia_dir = tmp_path / "chia" + chia_dir.mkdir() + + # Create some files within the chia package + create_python_file( + chia_dir, + "module1.py", + textwrap.dedent( + """ + # Package: one + def func1(): pass + from chia.module2 import func2 + from chia.module3 import func3 + """ + ), + ) + create_python_file( + chia_dir, + "module2.py", + textwrap.dedent( + """ + # Package: two + def func2(): pass + """ + ), + ) + create_python_file( + chia_dir, + "module3.py", + textwrap.dedent( + """ + # Package: three + def func3(): pass + """ + ), + ) + + # Run the command + runner = CliRunner() + result = runner.invoke( + cli, + [ + "print_edges", + "--directory", + str(chia_dir), + "--dependent-package", + "one", + "--provider-package", + "two", + ], + ) + assert result.output.strip() == f"{str(chia_dir / 'module1.py')} (one) -> {str(chia_dir / 'module2.py')} (two)" diff --git a/chia/util/virtual_project_analysis.py b/chia/util/virtual_project_analysis.py new file mode 100644 index 0000000000..2b6c31c13e --- /dev/null +++ b/chia/util/virtual_project_analysis.py @@ -0,0 +1,532 @@ +from __future__ import annotations + +import ast +import json +import os +import re +import sys +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Callable, Dict, List, Literal, Optional, Tuple, Union + +import click +import yaml + +# This tool enforces digraph dependencies within a "virtual project structure". +# i.e. files grouped together forming a project are not allowed to have cyclical +# dependencies on other such groups. + +# by default, all files are considered part of the "chia-blockchain" project. + +# To pull out a sub project, annotate its files with a comment (on the first +# line): +# Package: + +# if chia-blockchain depends on this new sub-project, the sub-project may not +# depend back on chia-blockchain. + + +@dataclass(frozen=True) +class Annotation: + package: str + is_annotated: bool + + @classmethod + def parse(cls, file_string: str) -> Annotation: + result = re.search(r"^# Package: (.+)$", file_string, re.MULTILINE) + if result is None: + return cls("chia-blockchain", False) + + return cls(result.group(1).strip(), True) + + +@dataclass(frozen=True) +class ChiaFile: + path: Path + annotations: Annotation + + @classmethod + def parse(cls, file_path: Path) -> ChiaFile: + with open(file_path, encoding="utf-8", errors="ignore") as f: + file_string = f.read().strip() + return cls(file_path, Annotation.parse(file_string)) + + +def build_dependency_graph(dir_params: DirectoryParameters) -> Dict[Path, List[Path]]: + dependency_graph: Dict[Path, List[Path]] = {} + for chia_file in dir_params.gather_non_empty_python_files(): + dependency_graph[chia_file.path] = [] + with open(chia_file.path, encoding="utf-8", errors="ignore") as f: + filestring = f.read() + tree = ast.parse(filestring, filename=chia_file.path) + for node in ast.iter_child_nodes(tree): + if isinstance(node, ast.ImportFrom): + if node.module is not None and node.module.startswith(dir_params.dir_path.stem): + imported_path = os.path.join(dir_params.dir_path.parent, node.module.replace(".", "/") + ".py") + paths_to_search = [ + imported_path, + *(os.path.join(imported_path[:-3], alias.name + ".py") for alias in node.names), + ] + for path_to_search in paths_to_search: + if os.path.exists(path_to_search): + dependency_graph[chia_file.path].append(Path(path_to_search)) + elif isinstance(node, ast.Import): + for alias in node.names: + if alias.name.startswith(dir_params.dir_path.stem): + imported_path = os.path.join( + dir_params.dir_path.parent, alias.name.replace(".", "/") + ".py" + ) + if os.path.exists(imported_path): + dependency_graph[chia_file.path].append(Path(imported_path)) + return dependency_graph + + +def build_virtual_dependency_graph( + dir_params: DirectoryParameters, *, existing_graph: Optional[Dict[Path, List[Path]]] = None +) -> Dict[str, List[str]]: + if existing_graph is None: + graph = build_dependency_graph(dir_params) + else: + graph = existing_graph + + virtual_graph: Dict[str, List[str]] = {} + for file, imports in graph.items(): + file_path = Path(file) + root_file = ChiaFile.parse(file_path) + if root_file.annotations is None: + continue + root = root_file.annotations.package + virtual_graph.setdefault(root, []) + + dependency_files = [ChiaFile.parse(Path(imp)) for imp in imports] + dependencies = [f.annotations.package for f in dependency_files if f.annotations is not None] + + virtual_graph[root].extend(dependencies) + + # Filter out self before returning the list + return {k: list({v for v in vs if v != k}) for k, vs in virtual_graph.items()} + + +@dataclass(frozen=True) +class Cycle: + dependent_path: Path + dependent_package: str + provider_path: Path + provider_package: str + packages_after_provider: List[str] + + def __repr__(self) -> str: + return "".join( + ( + f"{self.dependent_path} ({self.dependent_package}) -> ", + f"{self.provider_path} ({self.provider_package}) -> ", + *(f"({extra}) -> " for extra in self.packages_after_provider), + ) + )[:-4] + + def possible_edge_interpretations(self) -> List[Tuple[FileOrPackage, FileOrPackage]]: + edges_after_initial_files = [] + provider = self.packages_after_provider[0] + for next_provider in self.packages_after_provider[1:]: + edges_after_initial_files.append((Package(next_provider), Package(provider))) + provider = next_provider + + return [ + # Dependent -> Provider + (File(self.provider_path), File(self.dependent_path)), + (Package(self.provider_package), File(self.dependent_path)), + (File(self.provider_path), Package(self.dependent_package)), + (Package(self.provider_package), Package(self.dependent_package)), + # Provider -> Dependent/Other Packages + (Package(self.packages_after_provider[0]), File(self.provider_path)), + (Package(self.packages_after_provider[0]), Package(self.provider_package)), + # the rest + *edges_after_initial_files, + ] + + +def find_all_dependency_paths(dependency_graph: Dict[str, List[str]], start: str, end: str) -> List[List[str]]: + all_paths = [] + visited = set() + + def dfs(current: str, target: str, path: List[str]) -> None: + if current in visited: + return + if current == target and len(path) > 0: + all_paths.append(path[1:] + [current]) + return + visited.add(current) + for provider in sorted(dependency_graph.get(current, [])): + dfs(provider, target, path + [current]) + + dfs(start, end, []) + return all_paths + + +def find_cycles( + graph: Dict[Path, List[Path]], + virtual_graph: Dict[str, List[str]], + excluded_paths: List[Path], + ignore_cycles_in: List[str], + ignore_specific_files: List[Path], + ignore_specific_edges: List[Tuple[FileOrPackage, FileOrPackage]], +) -> List[Cycle]: + # Initialize an accumulator for paths that are part of cycles. + path_accumulator = [] + # Iterate over each package (parent) in the graph. + for dependent in sorted(graph): + if dependent in excluded_paths: + continue + # Parse the parent package file. + dependent_file = ChiaFile.parse(dependent) + # Skip this package if it has no annotations or should be ignored in cycle detection. + if ( + dependent_file.annotations is None + or dependent_file.annotations.package in ignore_cycles_in + or dependent in ignore_specific_files + ): + continue + + for provider in sorted(graph[dependent]): + if provider in excluded_paths: + continue + provider_file = ChiaFile.parse(provider) + if ( + provider_file.annotations is None + or provider_file.annotations.package == dependent_file.annotations.package + ): + continue + + dependency_paths = find_all_dependency_paths( + virtual_graph, provider_file.annotations.package, dependent_file.annotations.package + ) + if dependency_paths is None: + continue + + for dependency_path in dependency_paths: + possible_cycle = Cycle( + dependent_file.path, + dependent_file.annotations.package, + provider_file.path, + provider_file.annotations.package, + dependency_path, + ) + + for edge in possible_cycle.possible_edge_interpretations(): + if edge in ignore_specific_edges: + break + else: + path_accumulator.append(possible_cycle) + + # Format and return the accumulated paths as strings showing the cycles. + return path_accumulator + + +def print_graph(graph: Union[Dict[str, List[str]], Dict[Path, List[Path]]]) -> None: + print(json.dumps({str(k): list(str(v) for v in vs) for k, vs in graph.items()}, indent=4)) + + +@click.group(help="A utility for grouping different parts of the repo into separate projects") +def cli() -> None: + pass + + +@dataclass(frozen=True) +class DirectoryParameters: + dir_path: Path + excluded_paths: List[Path] = field(default_factory=list) + + def gather_non_empty_python_files(self) -> List[ChiaFile]: + """ + Gathers non-empty Python files in the specified directory while + ignoring files and directories in the excluded paths. + + Returns: + A list of paths to non-empty Python files. + """ + python_files = [] + for root, dirs, files in os.walk(self.dir_path, topdown=True): + # Modify dirs in-place to remove excluded directories from search + dirs[:] = [d for d in dirs if Path(os.path.join(root, d)) not in self.excluded_paths] + + for file in files: + file_path = Path(os.path.join(root, file)) + # Check if the file is a Python file and not in the excluded paths + if file_path.suffix == ".py" and file_path not in self.excluded_paths: + # Check if the file is non-empty + if os.path.getsize(file_path) > 0: + python_files.append(ChiaFile.parse(file_path)) + + return python_files + + +@dataclass(frozen=True) +class Config: + directory_parameters: DirectoryParameters + ignore_cycles_in: List[str] + ignore_specific_files: List[Path] + ignore_specific_edges: List[Tuple[FileOrPackage, FileOrPackage]] # (parent, child) + + +@dataclass(frozen=True) +class File: + name: Path + is_file: Literal[True] = True + + +@dataclass(frozen=True) +class Package: + name: str + is_file: Literal[False] = False + + +FileOrPackage = Union[File, Package] + + +def parse_file_or_package(identifier: str) -> FileOrPackage: + if ".py" in identifier: + if "(" not in identifier: + return File(Path(identifier)) + else: + return File(Path(identifier.split("(")[0].strip())) + + if ".py" not in identifier and identifier[0] == "(" and identifier[-1] == ")": + return Package(identifier[1:-1]) # strip parens + + return Package(identifier) + + +def parse_edge(user_string: str) -> Tuple[FileOrPackage, FileOrPackage]: + split_string = user_string.split("->") + dependent_side = split_string[0].strip() + provider_side = split_string[1].strip() + + return parse_file_or_package(provider_side), parse_file_or_package(dependent_side) + + +def config(func: Callable[..., None]) -> Callable[..., None]: + @click.option( + "--directory", + "include_dir", + type=click.Path(exists=True, file_okay=False, dir_okay=True), + required=True, + help="The directory to include.", + ) + @click.option( + "--exclude-path", + "excluded_paths", + multiple=True, + type=click.Path(exists=False, file_okay=True, dir_okay=True), + help="Optional paths to exclude.", + ) + @click.option( + "--config", + "config_path", + type=click.Path(exists=True), + required=False, + default=None, + help="Path to the YAML configuration file.", + ) + def inner(config_path: Optional[str], *args: Any, **kwargs: Any) -> None: + exclude_paths = [] + ignore_cycles_in: List[str] = [] + ignore_specific_files: List[str] = [] + ignore_specific_edges: List[str] = [] + if config_path is not None: + # Reading from the YAML configuration file + with open(config_path) as file: + config_data = yaml.safe_load(file) + + # Extracting required configuration values + exclude_paths = [Path(p) for p in config_data.get("exclude_paths") or []] + ignore_cycles_in = config_data["ignore"].get("packages") or [] + ignore_specific_files = config_data["ignore"].get("files") or [] + ignore_specific_edges = config_data["ignore"].get("edges") or [] + + # Instantiate DirectoryParameters with the provided options + dir_params = DirectoryParameters( + dir_path=Path(kwargs.pop("include_dir")), + excluded_paths=[*(Path(p) for p in kwargs.pop("excluded_paths")), *exclude_paths], + ) + + # Make the ignored edge dictionary + ignore_specific_edges_graph = [] + for ignore in (*kwargs.pop("ignore_specific_edges", []), *ignore_specific_edges): + parent, child = parse_edge(ignore) + ignore_specific_edges_graph.append((parent, child)) + + # Instantiating the Config object + config = Config( + directory_parameters=dir_params, + ignore_cycles_in=[*kwargs.pop("ignore_cycles_in", []), *ignore_cycles_in], + ignore_specific_files=[Path(p) for p in (*kwargs.pop("ignore_specific_files", []), *ignore_specific_files)], + ignore_specific_edges=ignore_specific_edges_graph, + ) + + # Calling the wrapped function with the Config object and other arguments + return func(config, *args, **kwargs) + + return inner + + +@click.command("find_missing_annotations", short_help="Search a directory for chia files without annotations") +@config +def find_missing_annotations(config: Config) -> None: + flag = False + for file in config.directory_parameters.gather_non_empty_python_files(): + if not file.annotations.is_annotated: + print(file.path) + flag = True + + if flag: + sys.exit(1) + + +@click.command("print_dependency_graph", short_help="Output a dependency graph of all the files in a directory") +@config +def print_dependency_graph(config: Config) -> None: + print_graph(build_dependency_graph(config.directory_parameters)) + + +@click.command( + "print_virtual_dependency_graph", short_help="Output a dependency graph of all the packages in a directory" +) +@config +def print_virtual_dependency_graph(config: Config) -> None: + print_graph(build_virtual_dependency_graph(config.directory_parameters)) + + +@click.command("print_cycles", short_help="Output cycles found in the virtual dependency graph") +@click.option( + "--ignore-cycles-in", + "ignore_cycles_in", + multiple=True, + type=str, + help="Ignore dependency cycles in a package", +) +@click.option( + "--ignore-specific-file", + "ignore_specific_files", + multiple=True, + type=click.Path(exists=True, file_okay=True, dir_okay=False), + help="Ignore cycles involving specific files", +) +@click.option( + "--ignore-specific-edge", + "ignore_specific_edges", + multiple=True, + type=str, + help="Ignore specific problematic dependencies (format: path/to/file1 -> path/to/file2)", +) +@config +def print_cycles(config: Config) -> None: + flag = False + graph = build_dependency_graph(config.directory_parameters) + for cycle in find_cycles( + graph, + build_virtual_dependency_graph(config.directory_parameters, existing_graph=graph), + config.directory_parameters.excluded_paths, + config.ignore_cycles_in, + config.ignore_specific_files, + config.ignore_specific_edges, + ): + print(cycle) + flag = True + + if flag: + sys.exit(1) + + +@click.command("check_config", short_help="Check the config is as specific as it can be") +@click.option( + "--ignore-cycles-in", + "ignore_cycles_in", + multiple=True, + type=str, + help="Ignore dependency cycles in a package", +) +@click.option( + "--ignore-specific-file", + "ignore_specific_files", + multiple=True, + type=click.Path(exists=True, file_okay=True, dir_okay=False), + help="Ignore cycles involving specific files", +) +@click.option( + "--ignore-specific-edge", + "ignore_specific_edges", + multiple=True, + type=str, + help="Ignore specific problematic dependencies (format: path/to/file1 -> path/to/file2)", +) +@config +def check_config(config: Config) -> None: + graph = build_dependency_graph(config.directory_parameters) + cycles = find_cycles( + graph, + build_virtual_dependency_graph(config.directory_parameters, existing_graph=graph), + config.directory_parameters.excluded_paths, + [], + [], + [], + ) + modules_found = set() + files_found = set() + edges_found = set() + for cycle in cycles: + modules_found.add(cycle.dependent_package) + files_found.add(cycle.dependent_path) + edges_found.update(set(cycle.possible_edge_interpretations())) + + for module in config.ignore_cycles_in: + if module not in modules_found: + print(f" module {module} ignored but no cycles were found") + print() + for file in config.ignore_specific_files: + if file not in files_found: + print(f" file {file} ignored but no cycles were found") + print() + for edge in config.ignore_specific_edges: + if edge not in edges_found: + print(f" edge {edge[1].name} -> {edge[0].name} ignored but no cycles were found") + + +@click.command("print_edges", short_help="Check for all of the ways a package immediately depends on another") +@click.option( + "--dependent-package", + "from_package", + type=str, + help="The package that depends on the other", +) +@click.option( + "--provider-package", + "to_package", + type=str, + help="The package that the dependent package imports from", +) +@config +def print_edges(config: Config, from_package: str, to_package: str) -> None: + graph = build_dependency_graph(config.directory_parameters) + for dependent, providers in graph.items(): + dependent_file = ChiaFile.parse(dependent) + assert dependent_file.annotations is not None + if dependent_file.annotations.package == from_package: + for provider in providers: + provider_file = ChiaFile.parse(provider) + assert provider_file.annotations is not None + if provider_file.annotations.package == to_package: + print( + f"{dependent} ({dependent_file.annotations.package}) -> " + f"{provider} ({provider_file.annotations.package})" + ) + + +cli.add_command(find_missing_annotations) +cli.add_command(print_dependency_graph) +cli.add_command(print_virtual_dependency_graph) +cli.add_command(print_cycles) +cli.add_command(check_config) +cli.add_command(print_edges) + +if __name__ == "__main__": + cli() diff --git a/virtual_project.yaml b/virtual_project.yaml new file mode 100644 index 0000000000..0774d3b636 --- /dev/null +++ b/virtual_project.yaml @@ -0,0 +1,5 @@ +exclude_paths: +ignore: + packages: + files: + edges: