"""Shared test helpers for decomposition BDD step files. Provides factory functions for creating temporary files, dependency graphs, and other test fixtures used by both the primary and coverage step implementations. """ from __future__ import annotations import itertools import os from cleveragents.application.services.decomposition_models import ( DependencyEdge, DependencyGraph, DependencyType, ) def make_files(tmpdir: str, count: int, levels: int = 1) -> list[str]: """Create *count* empty files spread across *levels* directories.""" paths: list[str] = [] for i in range(count): level = i % levels parts = [f"level{d}" for d in range(level + 1)] dirpath = os.path.join(tmpdir, *parts) os.makedirs(dirpath, exist_ok=True) fpath = os.path.join(dirpath, f"file_{i:05d}.py") with open(fpath, "w") as fh: fh.write("x" * 200) # ~50 tokens paths.append(fpath) return sorted(paths) def make_lang_files(tmpdir: str, py_count: int, ts_count: int) -> list[str]: """Create Python and TypeScript files in separate dirs.""" paths: list[str] = [] py_dir = os.path.join(tmpdir, "src", "python") ts_dir = os.path.join(tmpdir, "src", "typescript") os.makedirs(py_dir, exist_ok=True) os.makedirs(ts_dir, exist_ok=True) for i in range(py_count): fpath = os.path.join(py_dir, f"mod_{i:04d}.py") with open(fpath, "w") as fh: fh.write("x" * 200) paths.append(fpath) for i in range(ts_count): fpath = os.path.join(ts_dir, f"mod_{i:04d}.ts") with open(fpath, "w") as fh: fh.write("x" * 200) paths.append(fpath) return sorted(paths) def make_dir_files(tmpdir: str) -> list[str]: """Create files in two distinct directories.""" paths: list[str] = [] for sub in ("src/api", "src/web"): dirpath = os.path.join(tmpdir, sub) os.makedirs(dirpath, exist_ok=True) for i in range(50): fpath = os.path.join(dirpath, f"f_{i:03d}.py") with open(fpath, "w") as fh: fh.write("x" * 200) paths.append(fpath) return sorted(paths) def make_same_dir_files(tmpdir: str, count: int) -> list[str]: """Create *count* files all in the same directory with same extension.""" dirpath = os.path.join(tmpdir, "flat") os.makedirs(dirpath, exist_ok=True) paths: list[str] = [] for i in range(count): fpath = os.path.join(dirpath, f"f_{i:04d}.py") with open(fpath, "w") as fh: fh.write("x" * 200) paths.append(fpath) return sorted(paths) def chain_graph(*names: str) -> DependencyGraph: """Build a chain A -> B -> C -> ...""" g = DependencyGraph() for n in names: g.add_node(n) for src, tgt in itertools.pairwise(names): g.add_edge(DependencyEdge(src, tgt, DependencyType.IMPORT)) return g def diamond_graph() -> DependencyGraph: """Build A->B, A->C, B->D, C->D.""" g = DependencyGraph() for n in ("A", "B", "C", "D"): g.add_node(n) g.add_edge(DependencyEdge("A", "B", DependencyType.IMPORT)) g.add_edge(DependencyEdge("A", "C", DependencyType.IMPORT)) g.add_edge(DependencyEdge("B", "D", DependencyType.IMPORT)) g.add_edge(DependencyEdge("C", "D", DependencyType.IMPORT)) return g def cycle_graph() -> DependencyGraph: """Build A -> B -> C -> A.""" g = DependencyGraph() for n in ("A", "B", "C"): g.add_node(n) g.add_edge(DependencyEdge("A", "B", DependencyType.IMPORT)) g.add_edge(DependencyEdge("B", "C", DependencyType.IMPORT)) g.add_edge(DependencyEdge("C", "A", DependencyType.IMPORT)) return g