[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.
This commit is contained in:
Arvid Norberg
2024-10-10 13:56:33 -07:00
committed by GitHub
parent 8e4e84690d
commit 5f01fccdc6
4 changed files with 1135 additions and 0 deletions
+7
View File
@@ -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
+591
View File
@@ -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)"
+532
View File
@@ -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: <name>
# 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()
+5
View File
@@ -0,0 +1,5 @@
exclude_paths:
ignore:
packages:
files:
edges: