wehub-resource-sync
741 lines
24 KiB
Python
741 lines
24 KiB
Python
"""Declarative static import contracts; dynamic imports are deliberately unscanned."""
|
|
|
|
import ast
|
|
import subprocess
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
_REPO_ROOT = Path(__file__).resolve().parents[2]
|
|
_PACKAGE_ROOT = _REPO_ROOT / "src" / "free_claude_code"
|
|
_PACKAGE_NAME = "free_claude_code"
|
|
|
|
ALLOWED_PACKAGE_DEPENDENCIES: dict[str, set[str]] = {
|
|
"config": set(),
|
|
"core": set(),
|
|
"application": {"config", "core"},
|
|
"messaging": {"core"},
|
|
"providers": {"application", "config", "core"},
|
|
"api": {"application", "config", "core"},
|
|
"cli": {"config", "core"},
|
|
"runtime": {
|
|
"api",
|
|
"application",
|
|
"cli",
|
|
"config",
|
|
"core",
|
|
"messaging",
|
|
"providers",
|
|
},
|
|
}
|
|
|
|
IMPORT_EXCEPTIONS: dict[tuple[str, str], str] = {
|
|
(
|
|
"free_claude_code.cli.entrypoints",
|
|
"free_claude_code.runtime.bootstrap",
|
|
): (
|
|
"Owner: installed server entrypoint. "
|
|
"Reason: the executable delegates construction to the process composition root."
|
|
),
|
|
}
|
|
|
|
FACADE_ONLY_BOUNDARIES = {
|
|
"free_claude_code.core.openai_responses",
|
|
"free_claude_code.messaging.trees",
|
|
"free_claude_code.providers.openai_chat",
|
|
}
|
|
|
|
OPTIONAL_IMPORT_OWNERS = {
|
|
"librosa": "free_claude_code.messaging.transcription",
|
|
"torch": "free_claude_code.messaging.transcription",
|
|
"transformers": "free_claude_code.messaging.transcription",
|
|
"riva": "free_claude_code.providers.nvidia_nim.voice",
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ImportRecord:
|
|
importer: str
|
|
imported: str
|
|
path: str
|
|
line: int
|
|
inside_function: bool
|
|
|
|
def describe(self) -> str:
|
|
return f"{self.path}:{self.line}: {self.importer} -> {self.imported}"
|
|
|
|
|
|
class _ImportVisitor(ast.NodeVisitor):
|
|
def __init__(
|
|
self,
|
|
*,
|
|
importer: str,
|
|
importer_is_package: bool,
|
|
modules: set[str],
|
|
path: str,
|
|
) -> None:
|
|
self._importer = importer
|
|
self._importer_is_package = importer_is_package
|
|
self._modules = modules
|
|
self._path = path
|
|
self._function_depth = 0
|
|
self.records: list[ImportRecord] = []
|
|
|
|
def visit_Import(self, node: ast.Import) -> None:
|
|
for alias in node.names:
|
|
self._record(alias.name, node.lineno)
|
|
|
|
def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
|
|
base = _resolve_import_from_base(
|
|
importer=self._importer,
|
|
importer_is_package=self._importer_is_package,
|
|
level=node.level,
|
|
module=node.module,
|
|
)
|
|
if base is None:
|
|
return
|
|
for alias in node.names:
|
|
candidate = f"{base}.{alias.name}"
|
|
imported = candidate if candidate in self._modules else base
|
|
self._record(imported, node.lineno)
|
|
|
|
def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
|
|
self._visit_function(node)
|
|
|
|
def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
|
|
self._visit_function(node)
|
|
|
|
def _visit_function(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> None:
|
|
self._function_depth += 1
|
|
self.generic_visit(node)
|
|
self._function_depth -= 1
|
|
|
|
def _record(self, imported: str, line: int) -> None:
|
|
self.records.append(
|
|
ImportRecord(
|
|
importer=self._importer,
|
|
imported=imported,
|
|
path=self._path,
|
|
line=line,
|
|
inside_function=self._function_depth > 0,
|
|
)
|
|
)
|
|
|
|
|
|
def test_package_dependencies_follow_declarative_policy() -> None:
|
|
modules = _module_paths(_PACKAGE_ROOT)
|
|
records = _scan_imports(_PACKAGE_ROOT)
|
|
actual_packages = _ownership_roots(set(modules), _PACKAGE_NAME)
|
|
|
|
assert set(ALLOWED_PACKAGE_DEPENDENCIES) == actual_packages
|
|
assert all(
|
|
(_PACKAGE_ROOT / package / "__init__.py").is_file()
|
|
for package in actual_packages
|
|
)
|
|
assert all(
|
|
dependency in actual_packages and dependency != package
|
|
for package, dependencies in ALLOWED_PACKAGE_DEPENDENCIES.items()
|
|
for dependency in dependencies
|
|
)
|
|
|
|
observed_dependencies: set[tuple[str, str]] = set()
|
|
observed_exceptions: set[tuple[str, str]] = set()
|
|
offenders: list[str] = []
|
|
for record in records:
|
|
source_package = _top_level_package(record.importer)
|
|
target_package = _top_level_package(record.imported)
|
|
if source_package is None or target_package is None:
|
|
continue
|
|
if source_package == target_package:
|
|
continue
|
|
exact_edge = (record.importer, record.imported)
|
|
if exact_edge in IMPORT_EXCEPTIONS:
|
|
observed_exceptions.add(exact_edge)
|
|
continue
|
|
if target_package in ALLOWED_PACKAGE_DEPENDENCIES[source_package]:
|
|
observed_dependencies.add((source_package, target_package))
|
|
continue
|
|
offenders.append(record.describe())
|
|
|
|
declared_dependencies = {
|
|
(package, dependency)
|
|
for package, dependencies in ALLOWED_PACKAGE_DEPENDENCIES.items()
|
|
for dependency in dependencies
|
|
}
|
|
assert sorted(offenders) == []
|
|
assert declared_dependencies - observed_dependencies == set()
|
|
assert set(IMPORT_EXCEPTIONS) - observed_exceptions == set()
|
|
assert all(
|
|
"Owner:" in reason and "Reason:" in reason
|
|
for reason in IMPORT_EXCEPTIONS.values()
|
|
)
|
|
for source, target in IMPORT_EXCEPTIONS:
|
|
source_package = _top_level_package(source)
|
|
target_package = _top_level_package(target)
|
|
assert source_package is not None
|
|
assert target_package is not None
|
|
assert target_package not in ALLOWED_PACKAGE_DEPENDENCIES[source_package]
|
|
expected_package_modules = {
|
|
_PACKAGE_NAME,
|
|
*(f"{_PACKAGE_NAME}.{package}" for package in actual_packages),
|
|
}
|
|
assert set(modules) >= expected_package_modules
|
|
|
|
|
|
def test_first_party_imports_use_the_installable_namespace() -> None:
|
|
offenders = _legacy_first_party_import_offenders(
|
|
_scan_imports(_PACKAGE_ROOT),
|
|
set(ALLOWED_PACKAGE_DEPENDENCIES),
|
|
)
|
|
|
|
assert offenders == []
|
|
|
|
|
|
def test_openai_chat_collaborators_have_explicit_ownership_boundaries() -> None:
|
|
provider_root = _PACKAGE_ROOT / "providers" / "openai_chat"
|
|
|
|
assert _provider_backchannel_offenders(provider_root) == []
|
|
|
|
|
|
def test_provider_backchannel_detector_reports_untyped_private_access(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
provider_root = tmp_path / "openai_chat"
|
|
_write_module(
|
|
provider_root / "sample" / "runner.py",
|
|
"from typing import Any\n"
|
|
"\n"
|
|
"class Runner:\n"
|
|
" def __init__(self, provider: object | Any) -> None:\n"
|
|
" self._provider = provider\n"
|
|
"\n"
|
|
" def run(self) -> object:\n"
|
|
" return self._provider._send()\n",
|
|
)
|
|
|
|
assert _provider_backchannel_offenders(provider_root) == [
|
|
"sample/runner.py:4: untyped provider collaborator",
|
|
"sample/runner.py:8: private provider member _send outside provider.py",
|
|
]
|
|
|
|
|
|
def test_legacy_first_party_import_detector_rejects_bare_owner_names() -> None:
|
|
record = ImportRecord(
|
|
importer="free_claude_code.api.routes",
|
|
imported="core.anthropic",
|
|
path="free_claude_code/api/routes.py",
|
|
line=7,
|
|
inside_function=False,
|
|
)
|
|
|
|
assert _legacy_first_party_import_offenders([record], {"api", "core"}) == [
|
|
"free_claude_code/api/routes.py:7: "
|
|
"free_claude_code.api.routes -> core.anthropic"
|
|
]
|
|
|
|
|
|
def test_ownership_root_discovery_includes_modules_and_namespace_directories(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
package_root = tmp_path / "sample"
|
|
_write_module(package_root / "__init__.py")
|
|
_write_module(package_root / "declared" / "__init__.py")
|
|
_write_module(package_root / "namespace" / "module.py")
|
|
_write_module(package_root / "root_module.py")
|
|
|
|
roots = _ownership_roots(set(_module_paths(package_root)), "sample")
|
|
|
|
assert roots == {"declared", "namespace", "root_module"}
|
|
|
|
|
|
def test_descendants_do_not_import_ancestor_package_facades() -> None:
|
|
modules = _module_paths(_PACKAGE_ROOT)
|
|
packages = {
|
|
module for module, path in modules.items() if path.name == "__init__.py"
|
|
}
|
|
offenders = _ancestor_facade_offenders(_scan_imports(_PACKAGE_ROOT), packages)
|
|
|
|
assert offenders == []
|
|
|
|
|
|
def test_static_first_party_import_graph_is_acyclic() -> None:
|
|
modules = set(_module_paths(_PACKAGE_ROOT))
|
|
graph = {module: set() for module in modules}
|
|
for record in _scan_imports(_PACKAGE_ROOT):
|
|
if record.imported in modules:
|
|
graph[record.importer].add(record.imported)
|
|
|
|
assert _cyclic_components(graph) == []
|
|
|
|
|
|
def test_cycle_detector_reports_exact_strongly_connected_components() -> None:
|
|
graph = {
|
|
"package.a": {"package.b"},
|
|
"package.b": {"package.a"},
|
|
"package.c": set(),
|
|
"package.self": {"package.self"},
|
|
}
|
|
|
|
assert _cyclic_components(graph) == [
|
|
("package.a", "package.b"),
|
|
("package.self",),
|
|
]
|
|
|
|
|
|
def test_external_consumers_use_owned_package_facades() -> None:
|
|
offenders: list[str] = []
|
|
for record in _scan_imports(_PACKAGE_ROOT):
|
|
for facade in FACADE_ONLY_BOUNDARIES:
|
|
if record.importer == facade or record.importer.startswith(f"{facade}."):
|
|
continue
|
|
if record.imported.startswith(f"{facade}."):
|
|
offenders.append(record.describe())
|
|
|
|
assert sorted(offenders) == []
|
|
|
|
|
|
def test_import_scanner_resolves_absolute_relative_and_lazy_imports(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
package_root = tmp_path / "sample"
|
|
_write_module(package_root / "__init__.py")
|
|
_write_module(package_root / "alpha" / "__init__.py", "PUBLIC = object()\n")
|
|
_write_module(package_root / "alpha" / "sibling.py")
|
|
_write_module(package_root / "beta" / "__init__.py")
|
|
_write_module(package_root / "beta" / "absolute.py")
|
|
_write_module(package_root / "beta" / "relative.py")
|
|
_write_module(package_root / "beta" / "lazy.py")
|
|
_write_module(
|
|
package_root / "alpha" / "consumer.py",
|
|
"\n".join(
|
|
(
|
|
"import sample.beta.absolute",
|
|
"from ..beta import relative",
|
|
"from . import sibling",
|
|
"from . import PUBLIC",
|
|
"def load():",
|
|
" import sample.beta.lazy",
|
|
"",
|
|
)
|
|
),
|
|
)
|
|
|
|
records = _scan_imports(package_root)
|
|
resolved = {
|
|
(record.imported, record.inside_function)
|
|
for record in records
|
|
if record.importer == "sample.alpha.consumer"
|
|
}
|
|
|
|
assert resolved == {
|
|
("sample.alpha", False),
|
|
("sample.alpha.sibling", False),
|
|
("sample.beta.absolute", False),
|
|
("sample.beta.relative", False),
|
|
("sample.beta.lazy", True),
|
|
}
|
|
|
|
|
|
def test_import_scanner_distinguishes_facade_symbols_from_sibling_modules(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
package_root = tmp_path / "sample"
|
|
_write_module(package_root / "__init__.py")
|
|
_write_module(package_root / "owner" / "__init__.py", "PUBLIC = object()\n")
|
|
_write_module(package_root / "owner" / "sibling.py")
|
|
_write_module(
|
|
package_root / "owner" / "consumer.py",
|
|
"from . import PUBLIC, sibling\n",
|
|
)
|
|
modules = _module_paths(package_root)
|
|
packages = {
|
|
module for module, path in modules.items() if path.name == "__init__.py"
|
|
}
|
|
|
|
offenders = _ancestor_facade_offenders(_scan_imports(package_root), packages)
|
|
|
|
assert offenders == [
|
|
"sample/owner/consumer.py:1: sample.owner.consumer -> sample.owner"
|
|
]
|
|
|
|
|
|
def test_anthropic_request_boundaries_use_the_protocol_model() -> None:
|
|
"""Known Messages fields must not cross core/provider boundaries by duck typing."""
|
|
roots = [
|
|
_PACKAGE_ROOT / "core" / "anthropic",
|
|
_PACKAGE_ROOT / "providers",
|
|
]
|
|
request_names = {"request", "request_data"}
|
|
protocol_fields = {
|
|
"extra_body",
|
|
"max_tokens",
|
|
"messages",
|
|
"model",
|
|
"stop_sequences",
|
|
"system",
|
|
"temperature",
|
|
"thinking",
|
|
"tool_choice",
|
|
"tools",
|
|
"top_k",
|
|
"top_p",
|
|
}
|
|
offenders: list[str] = []
|
|
|
|
for root in roots:
|
|
for path in root.rglob("*.py"):
|
|
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
|
|
relative = path.relative_to(_REPO_ROOT).as_posix()
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef):
|
|
arguments = [
|
|
*node.args.posonlyargs,
|
|
*node.args.args,
|
|
*node.args.kwonlyargs,
|
|
]
|
|
for argument in arguments:
|
|
if argument.arg.lstrip("_") not in request_names:
|
|
continue
|
|
annotation_names = (
|
|
{
|
|
child.id
|
|
for child in ast.walk(argument.annotation)
|
|
if isinstance(child, ast.Name)
|
|
}
|
|
if argument.annotation is not None
|
|
else set()
|
|
)
|
|
if argument.annotation is None or annotation_names & {
|
|
"Any",
|
|
"Mapping",
|
|
}:
|
|
offenders.append(
|
|
f"{relative}:{argument.lineno}: "
|
|
f"{node.name}({argument.arg}) is not concrete"
|
|
)
|
|
if not isinstance(node, ast.Call) or not (
|
|
isinstance(node.func, ast.Name)
|
|
and node.func.id == "getattr"
|
|
and len(node.args) >= 2
|
|
and isinstance(node.args[0], ast.Name)
|
|
and node.args[0].id.lstrip("_") in request_names
|
|
and isinstance(node.args[1], ast.Constant)
|
|
and node.args[1].value in protocol_fields
|
|
):
|
|
continue
|
|
offenders.append(
|
|
f"{relative}:{node.lineno}: "
|
|
f"getattr({node.args[0].id}, {node.args[1].value!r})"
|
|
)
|
|
|
|
assert sorted(offenders) == []
|
|
|
|
|
|
def test_core_does_not_import_provider_transport_sdks() -> None:
|
|
forbidden_roots = {"aiohttp", "httpx", "openai"}
|
|
offenders = [
|
|
record.describe()
|
|
for record in _scan_imports(_PACKAGE_ROOT)
|
|
if (
|
|
record.importer == "free_claude_code.core"
|
|
or record.importer.startswith("free_claude_code.core.")
|
|
)
|
|
and record.imported.split(".", 1)[0] in forbidden_roots
|
|
]
|
|
|
|
assert sorted(offenders) == []
|
|
|
|
|
|
def test_providers_do_not_own_wire_error_type_literals() -> None:
|
|
wire_types = {
|
|
"api_error",
|
|
"authentication_error",
|
|
"billing_error",
|
|
"invalid_request_error",
|
|
"not_found_error",
|
|
"overloaded_error",
|
|
"permission_error",
|
|
"rate_limit_error",
|
|
"request_too_large",
|
|
"timeout_error",
|
|
}
|
|
offenders: list[str] = []
|
|
for path in (_PACKAGE_ROOT / "providers").rglob("*.py"):
|
|
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
|
|
offenders.extend(
|
|
f"{path.relative_to(_REPO_ROOT).as_posix()}:{node.lineno}: {node.value}"
|
|
for node in ast.walk(tree)
|
|
if isinstance(node, ast.Constant) and node.value in wire_types
|
|
)
|
|
|
|
assert sorted(offenders) == []
|
|
|
|
|
|
def test_optional_dependencies_have_one_lazy_owner() -> None:
|
|
seen: set[str] = set()
|
|
offenders: list[str] = []
|
|
for record in _scan_imports(_PACKAGE_ROOT):
|
|
dependency = record.imported.split(".", 1)[0]
|
|
owner = OPTIONAL_IMPORT_OWNERS.get(dependency)
|
|
if owner is None:
|
|
continue
|
|
seen.add(dependency)
|
|
if record.importer != owner or not record.inside_function:
|
|
offenders.append(record.describe())
|
|
|
|
assert seen == set(OPTIONAL_IMPORT_OWNERS)
|
|
assert sorted(offenders) == []
|
|
|
|
|
|
def test_runtime_imports_without_optional_voice_dependencies() -> None:
|
|
blocked = sorted(OPTIONAL_IMPORT_OWNERS)
|
|
script = "\n".join(
|
|
(
|
|
"import importlib.abc",
|
|
"import sys",
|
|
f"BLOCKED = {blocked!r}",
|
|
"class Blocker(importlib.abc.MetaPathFinder):",
|
|
" def find_spec(self, fullname, path=None, target=None):",
|
|
" if fullname.split('.', 1)[0] in BLOCKED:",
|
|
" raise ModuleNotFoundError(fullname)",
|
|
" return None",
|
|
"sys.meta_path.insert(0, Blocker())",
|
|
"import free_claude_code.runtime.bootstrap",
|
|
"import free_claude_code.api.app",
|
|
)
|
|
)
|
|
|
|
completed = subprocess.run(
|
|
[sys.executable, "-c", script],
|
|
capture_output=True,
|
|
check=False,
|
|
text=True,
|
|
)
|
|
|
|
assert completed.returncode == 0, completed.stderr or completed.stdout
|
|
|
|
|
|
def test_supported_messaging_facade_is_explicit() -> None:
|
|
import free_claude_code.messaging as facade
|
|
from free_claude_code.messaging.managed_protocols import (
|
|
ManagedClaudeSessionManagerProtocol,
|
|
ManagedClaudeSessionProtocol,
|
|
)
|
|
from free_claude_code.messaging.models import IncomingMessage, MessageScope
|
|
from free_claude_code.messaging.platforms.ports import OutboundMessenger
|
|
|
|
expected = {
|
|
"IncomingMessage": IncomingMessage,
|
|
"ManagedClaudeSessionManagerProtocol": ManagedClaudeSessionManagerProtocol,
|
|
"ManagedClaudeSessionProtocol": ManagedClaudeSessionProtocol,
|
|
"MessageScope": MessageScope,
|
|
"OutboundMessenger": OutboundMessenger,
|
|
}
|
|
|
|
assert set(facade.__all__) == set(expected)
|
|
assert all(getattr(facade, name) is value for name, value in expected.items())
|
|
|
|
|
|
def test_message_tree_mutability_stays_behind_its_facade() -> None:
|
|
import free_claude_code.messaging.trees as facade
|
|
|
|
for internal_owner in {
|
|
"MessageNode",
|
|
"MessageTree",
|
|
"TreeQueueProcessor",
|
|
"TreeRepository",
|
|
}:
|
|
assert internal_owner not in facade.__all__
|
|
assert not hasattr(facade, internal_owner)
|
|
|
|
|
|
def _module_paths(package_root: Path) -> dict[str, Path]:
|
|
return {
|
|
_module_name(package_root, path): path for path in package_root.rglob("*.py")
|
|
}
|
|
|
|
|
|
def _module_name(package_root: Path, path: Path) -> str:
|
|
relative = path.relative_to(package_root)
|
|
module_parts = (
|
|
relative.parent.parts
|
|
if path.name == "__init__.py"
|
|
else relative.with_suffix("").parts
|
|
)
|
|
return ".".join((package_root.name, *module_parts))
|
|
|
|
|
|
def _scan_imports(package_root: Path) -> list[ImportRecord]:
|
|
module_paths = _module_paths(package_root)
|
|
modules = set(module_paths)
|
|
records: list[ImportRecord] = []
|
|
for importer, path in sorted(module_paths.items()):
|
|
visitor = _ImportVisitor(
|
|
importer=importer,
|
|
importer_is_package=path.name == "__init__.py",
|
|
modules=modules,
|
|
path=path.relative_to(package_root.parent).as_posix(),
|
|
)
|
|
visitor.visit(ast.parse(path.read_text(encoding="utf-8"), filename=str(path)))
|
|
records.extend(visitor.records)
|
|
return records
|
|
|
|
|
|
def _resolve_import_from_base(
|
|
*,
|
|
importer: str,
|
|
importer_is_package: bool,
|
|
level: int,
|
|
module: str | None,
|
|
) -> str | None:
|
|
if level == 0:
|
|
return module
|
|
package_parts = importer.split(".")
|
|
if not importer_is_package:
|
|
package_parts.pop()
|
|
parents_to_remove = level - 1
|
|
if parents_to_remove > len(package_parts):
|
|
return None
|
|
if parents_to_remove:
|
|
del package_parts[-parents_to_remove:]
|
|
if module is not None:
|
|
package_parts.extend(module.split("."))
|
|
return ".".join(package_parts) or None
|
|
|
|
|
|
def _top_level_package(module: str) -> str | None:
|
|
parts = module.split(".")
|
|
if len(parts) < 2 or parts[0] != _PACKAGE_NAME:
|
|
return None
|
|
return parts[1]
|
|
|
|
|
|
def _ownership_roots(modules: set[str], package_name: str) -> set[str]:
|
|
roots: set[str] = set()
|
|
for module in modules:
|
|
parts = module.split(".")
|
|
if len(parts) >= 2 and parts[0] == package_name:
|
|
roots.add(parts[1])
|
|
return roots
|
|
|
|
|
|
def _legacy_first_party_import_offenders(
|
|
records: list[ImportRecord],
|
|
owner_names: set[str],
|
|
) -> list[str]:
|
|
offenders = [
|
|
record.describe()
|
|
for record in records
|
|
if record.imported.split(".", 1)[0] in owner_names
|
|
]
|
|
return sorted(offenders)
|
|
|
|
|
|
def _provider_backchannel_offenders(provider_root: Path) -> list[str]:
|
|
offenders: list[str] = []
|
|
for path in sorted(provider_root.rglob("*.py")):
|
|
relative_path = path.relative_to(provider_root).as_posix()
|
|
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef):
|
|
arguments = (
|
|
*node.args.posonlyargs,
|
|
*node.args.args,
|
|
*node.args.kwonlyargs,
|
|
)
|
|
offenders.extend(
|
|
f"{relative_path}:{argument.lineno}: untyped provider collaborator"
|
|
for argument in arguments
|
|
if argument.arg == "provider"
|
|
and _annotation_is_any(argument.annotation)
|
|
)
|
|
if (
|
|
path.name != "provider.py"
|
|
and isinstance(node, ast.Attribute)
|
|
and node.attr.startswith("_")
|
|
and _is_provider_reference(node.value)
|
|
):
|
|
offenders.append(
|
|
f"{relative_path}:{node.lineno}: private provider member "
|
|
f"{node.attr} outside provider.py"
|
|
)
|
|
return sorted(offenders)
|
|
|
|
|
|
def _annotation_is_any(annotation: ast.expr | None) -> bool:
|
|
if annotation is None:
|
|
return False
|
|
return any(
|
|
(isinstance(node, ast.Name) and node.id == "Any")
|
|
or (isinstance(node, ast.Attribute) and node.attr == "Any")
|
|
for node in ast.walk(annotation)
|
|
)
|
|
|
|
|
|
def _is_provider_reference(expression: ast.expr) -> bool:
|
|
return (isinstance(expression, ast.Name) and expression.id == "provider") or (
|
|
isinstance(expression, ast.Attribute)
|
|
and isinstance(expression.value, ast.Name)
|
|
and expression.value.id == "self"
|
|
and expression.attr == "_provider"
|
|
)
|
|
|
|
|
|
def _ancestor_facade_offenders(
|
|
records: list[ImportRecord],
|
|
packages: set[str],
|
|
) -> list[str]:
|
|
offenders = [
|
|
record.describe()
|
|
for record in records
|
|
if record.imported in packages
|
|
and record.importer.startswith(f"{record.imported}.")
|
|
]
|
|
return sorted(offenders)
|
|
|
|
|
|
def _cyclic_components(graph: dict[str, set[str]]) -> list[tuple[str, ...]]:
|
|
index = 0
|
|
indices: dict[str, int] = {}
|
|
lowlinks: dict[str, int] = {}
|
|
stack: list[str] = []
|
|
on_stack: set[str] = set()
|
|
components: list[tuple[str, ...]] = []
|
|
|
|
def visit(node: str) -> None:
|
|
nonlocal index
|
|
indices[node] = index
|
|
lowlinks[node] = index
|
|
index += 1
|
|
stack.append(node)
|
|
on_stack.add(node)
|
|
|
|
for successor in sorted(graph[node]):
|
|
if successor not in indices:
|
|
visit(successor)
|
|
lowlinks[node] = min(lowlinks[node], lowlinks[successor])
|
|
elif successor in on_stack:
|
|
lowlinks[node] = min(lowlinks[node], indices[successor])
|
|
|
|
if lowlinks[node] != indices[node]:
|
|
return
|
|
component: list[str] = []
|
|
while True:
|
|
member = stack.pop()
|
|
on_stack.remove(member)
|
|
component.append(member)
|
|
if member == node:
|
|
break
|
|
if len(component) > 1 or node in graph[node]:
|
|
components.append(tuple(sorted(component)))
|
|
|
|
for node in sorted(graph):
|
|
if node not in indices:
|
|
visit(node)
|
|
return sorted(components)
|
|
|
|
|
|
def _write_module(path: Path, text: str = "") -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(text, encoding="utf-8")
|