63 lines
2.3 KiB
Python
63 lines
2.3 KiB
Python
"""Static guard for direct filesystem mutation outside adapters/actions."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from pathlib import Path
|
|
|
|
|
|
MUTATING_PATTERNS = (
|
|
re.compile(r"\.(mkdir|unlink|write_text|write_bytes|symlink_to|chmod)\("),
|
|
re.compile(r"\b(os\.replace|shutil\.rmtree|shutil\.copy2|shutil\.copytree)\("),
|
|
re.compile(r"runtime\.fs\.(create_symlink|remove_symlink|write_json|write_text|write_bytes|copy_file|copy_tree|remove_file|remove_tree)\("),
|
|
)
|
|
|
|
COMMAND_PATTERNS = (
|
|
re.compile(r"runtime\.runner\.(run|run_shell)\("),
|
|
re.compile(r"runtime\.git\.run\("),
|
|
re.compile(r"runtime\.containers\.(run_container|start|stop|kill|rm)\("),
|
|
re.compile(r"runtime\.tmux\.(new_session|set_option|respawn_pane)\("),
|
|
re.compile(r"self\.(rt|tmux)\.(run_container|start|stop|kill|rm|new_session|set_option|respawn_pane)\("),
|
|
)
|
|
|
|
ALLOWED_PREFIXES = (
|
|
Path("src/flow/adapters"),
|
|
Path("src/flow/actions"),
|
|
Path("tests"),
|
|
)
|
|
|
|
ALLOWED_FILES = {
|
|
Path("src/flow/core/paths.py"),
|
|
}
|
|
|
|
def test_no_direct_filesystem_mutation_outside_action_boundary():
|
|
root = Path(__file__).resolve().parents[1]
|
|
offenders: list[str] = []
|
|
for path in sorted((root / "src" / "flow").rglob("*.py")):
|
|
rel = path.relative_to(root)
|
|
if rel in ALLOWED_FILES:
|
|
continue
|
|
if any(rel.is_relative_to(prefix) for prefix in ALLOWED_PREFIXES):
|
|
continue
|
|
for line_no, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
|
|
if "Service(" in line:
|
|
continue
|
|
if any(pattern.search(line) for pattern in MUTATING_PATTERNS):
|
|
offenders.append(f"{rel}:{line_no}: {line.strip()}")
|
|
|
|
assert offenders == []
|
|
|
|
|
|
def test_no_direct_command_mutation_outside_action_boundary():
|
|
root = Path(__file__).resolve().parents[1]
|
|
offenders: list[str] = []
|
|
for path in sorted((root / "src" / "flow").rglob("*.py")):
|
|
rel = path.relative_to(root)
|
|
if any(rel.is_relative_to(prefix) for prefix in ALLOWED_PREFIXES):
|
|
continue
|
|
for line_no, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
|
|
if any(pattern.search(line) for pattern in COMMAND_PATTERNS):
|
|
offenders.append(f"{rel}:{line_no}: {line.strip()}")
|
|
|
|
assert offenders == []
|