forked from cleveragents/cleveragents-core
321 lines
12 KiB
Python
321 lines
12 KiB
Python
"""Step definitions for architecture validation."""
|
|
|
|
import ast
|
|
import re
|
|
from pathlib import Path
|
|
|
|
from behave import given, then, when
|
|
|
|
|
|
@given('the ADR directory exists at "{path}"')
|
|
def step_adr_directory_exists(context, path):
|
|
"""Check if the ADR directory exists."""
|
|
context.adr_dir = Path(path)
|
|
assert context.adr_dir.exists(), f"ADR directory {path} does not exist"
|
|
assert context.adr_dir.is_dir(), f"{path} is not a directory"
|
|
|
|
|
|
@when("I check for ADR files")
|
|
def step_check_adr_files(context):
|
|
"""Find all ADR files in the directory."""
|
|
context.adr_files = list(context.adr_dir.glob("ADR-*.md"))
|
|
|
|
|
|
@then("I should find at least {count:d} ADR files")
|
|
def step_verify_adr_count(context, count):
|
|
"""Verify minimum number of ADR files."""
|
|
assert len(context.adr_files) >= count, (
|
|
f"Expected at least {count} ADR files, found {len(context.adr_files)}"
|
|
)
|
|
|
|
|
|
@then("each ADR should have a valid format with status")
|
|
def step_verify_adr_format(context):
|
|
"""Verify each ADR follows the expected format."""
|
|
for adr_file in context.adr_files:
|
|
content = adr_file.read_text()
|
|
# Check for required sections
|
|
assert "## Status" in content, f"{adr_file.name} missing Status section"
|
|
assert "## Context" in content, f"{adr_file.name} missing Context section"
|
|
assert "## Decision" in content, f"{adr_file.name} missing Decision section"
|
|
assert "## Consequences" in content, (
|
|
f"{adr_file.name} missing Consequences section"
|
|
)
|
|
|
|
|
|
@given('the source directory exists at "{path}"')
|
|
def step_source_directory_exists(context, path):
|
|
"""Check if the source directory exists."""
|
|
context.src_dir = Path(path)
|
|
assert context.src_dir.exists(), f"Source directory {path} does not exist"
|
|
|
|
|
|
@when("I check the package structure")
|
|
def step_check_package_structure(context):
|
|
"""List all packages in the source directory."""
|
|
try:
|
|
# Ensure the directory exists
|
|
if not context.src_dir.exists():
|
|
raise AssertionError(f"Source directory {context.src_dir} does not exist")
|
|
|
|
# List all items in the directory with better error handling
|
|
context.packages = {}
|
|
for item in context.src_dir.iterdir():
|
|
try:
|
|
if item.is_dir() and not item.name.startswith("__"):
|
|
# Store all packages including discovery - we'll handle expected vs actual in verification
|
|
context.packages[item.name] = item
|
|
except (OSError, PermissionError) as e:
|
|
# Log the error but continue processing other items
|
|
print(f"Warning: Could not check {item}: {e}")
|
|
continue
|
|
|
|
except Exception as e:
|
|
raise AssertionError(f"Failed to check package structure: {e}")
|
|
|
|
|
|
@then("I should find these main packages:")
|
|
def step_verify_packages(context):
|
|
"""Verify expected packages exist and are proper Python packages."""
|
|
# Check if packages were properly collected
|
|
if not hasattr(context, "packages"):
|
|
raise AssertionError(
|
|
"Package structure was not checked properly - 'packages' attribute missing"
|
|
)
|
|
|
|
# Parse the table to get expected packages and their descriptions
|
|
expected_packages = {
|
|
row["package"]: row.get("description", "") for row in context.table
|
|
}
|
|
actual_packages = set(context.packages.keys())
|
|
|
|
# Check for missing packages
|
|
missing = set(expected_packages.keys()) - actual_packages
|
|
if missing:
|
|
extra = actual_packages - set(expected_packages.keys())
|
|
error_msg = f"Missing packages: {missing}"
|
|
if extra:
|
|
error_msg += f"\nUnexpected packages found: {extra}"
|
|
error_msg += f"\nActual packages: {sorted(actual_packages)}"
|
|
error_msg += f"\nExpected packages: {sorted(expected_packages.keys())}"
|
|
raise AssertionError(error_msg)
|
|
|
|
# Verify each expected package is a proper Python package
|
|
for package_name in expected_packages:
|
|
package_path = context.packages[package_name]
|
|
init_file = package_path / "__init__.py"
|
|
if not init_file.exists():
|
|
raise AssertionError(
|
|
f"Package '{package_name}' at {package_path} is not a proper Python package "
|
|
f"(__init__.py not found)"
|
|
)
|
|
|
|
|
|
@given("the package structure is defined")
|
|
def step_package_structure_defined(context):
|
|
"""Set up package structure for import analysis."""
|
|
context.src_dir = Path("src/cleveragents")
|
|
context.packages = {
|
|
p.name: p
|
|
for p in context.src_dir.iterdir()
|
|
if p.is_dir() and not p.name.startswith("__")
|
|
}
|
|
|
|
|
|
@when("I analyze package imports")
|
|
def step_analyze_imports(context):
|
|
"""Analyze imports in all Python files."""
|
|
context.imports = {}
|
|
for package_name, package_path in context.packages.items():
|
|
package_imports = set()
|
|
for py_file in package_path.rglob("*.py"):
|
|
if py_file.name == "__init__.py" and py_file.read_text().strip() == "":
|
|
continue
|
|
try:
|
|
tree = ast.parse(py_file.read_text())
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Import):
|
|
for alias in node.names:
|
|
if alias.name.startswith("cleveragents."):
|
|
package_imports.add(alias.name.split(".")[1])
|
|
elif isinstance(node, ast.ImportFrom):
|
|
if node.module and node.module.startswith("cleveragents."):
|
|
package_imports.add(node.module.split(".")[1])
|
|
except SyntaxError:
|
|
pass # Skip files with syntax errors
|
|
context.imports[package_name] = package_imports
|
|
|
|
|
|
@then("{package} should not import from {target}")
|
|
def step_verify_no_import(context, package, target):
|
|
"""Verify a package doesn't import from another."""
|
|
if package not in context.imports:
|
|
return # Package doesn't exist yet
|
|
imports = context.imports.get(package, set())
|
|
assert target not in imports, f"{package} imports from {target}"
|
|
|
|
|
|
@then("the core package should be self-contained")
|
|
def step_verify_core_self_contained(context):
|
|
"""Verify core package doesn't import from any other internal package."""
|
|
package = "core"
|
|
if package not in context.imports:
|
|
return # Package doesn't exist yet
|
|
imports = context.imports.get(package, set())
|
|
# Filter out self-imports and allowed imports
|
|
# Core can import from config for dependency injection container setup
|
|
allowed_imports = {"core", "config"}
|
|
other_imports = imports - allowed_imports
|
|
assert not other_imports, f"{package} imports from: {other_imports}"
|
|
|
|
|
|
@given("the settings module exists")
|
|
def step_settings_module_exists(context):
|
|
"""Check if settings module exists."""
|
|
context.settings_file = Path("src/cleveragents/config/settings.py")
|
|
assert context.settings_file.exists(), "Settings module does not exist"
|
|
|
|
|
|
@when("I check environment variable definitions")
|
|
def step_check_env_vars(context):
|
|
"""Extract environment variable definitions from settings."""
|
|
content = context.settings_file.read_text()
|
|
# Find all environment variable references
|
|
context.env_vars = re.findall(r'["\']([A-Z_]+)["\']', content)
|
|
|
|
|
|
@then("all application variables should use {prefix} prefix")
|
|
def step_verify_env_prefix(context, prefix):
|
|
"""Verify application environment variables use correct prefix."""
|
|
# Provider prefixes that should be excluded
|
|
provider_prefixes = [
|
|
"OPENAI_",
|
|
"ANTHROPIC_",
|
|
"CLAUDE_",
|
|
"GEMINI_",
|
|
"AZURE_",
|
|
"GOOGLE_",
|
|
"OPENROUTER_",
|
|
"HF_",
|
|
]
|
|
|
|
# Common constants and system variables to exclude
|
|
excluded_constants = [
|
|
"HOME",
|
|
"PATH",
|
|
"USER",
|
|
"SHELL",
|
|
"PWD", # System vars
|
|
"DEBUG",
|
|
"INFO",
|
|
"WARNING",
|
|
"ERROR",
|
|
"CRITICAL", # Log levels
|
|
"GET",
|
|
"POST",
|
|
"PUT",
|
|
"DELETE",
|
|
"PATCH", # HTTP methods
|
|
"JSON",
|
|
"TEXT",
|
|
"HTML",
|
|
"XML", # Content types
|
|
]
|
|
|
|
app_vars = [
|
|
var
|
|
for var in context.env_vars
|
|
if not any(var.startswith(p) for p in provider_prefixes)
|
|
and var not in excluded_constants
|
|
]
|
|
|
|
invalid_vars = [var for var in app_vars if not var.startswith(prefix)]
|
|
assert not invalid_vars, f"Variables not using {prefix} prefix: {invalid_vars}"
|
|
|
|
|
|
@then("provider variables should keep their original prefix")
|
|
def step_verify_provider_vars(context):
|
|
"""Verify provider variables keep their original prefix."""
|
|
provider_prefixes = [
|
|
"OPENAI_",
|
|
"ANTHROPIC_",
|
|
"CLAUDE_",
|
|
"GEMINI_",
|
|
"AZURE_",
|
|
"GOOGLE_",
|
|
]
|
|
provider_vars = [
|
|
var
|
|
for var in context.env_vars
|
|
if any(var.startswith(p) for p in provider_prefixes)
|
|
]
|
|
# Just verify they exist and aren't renamed
|
|
for var in provider_vars:
|
|
assert not var.startswith("CLEVERAGENTS_"), f"Provider var renamed: {var}"
|
|
|
|
|
|
@given("the source code exists")
|
|
def step_source_code_exists(context):
|
|
"""Set up source code for analysis."""
|
|
context.src_dir = Path("src/cleveragents")
|
|
assert context.src_dir.exists()
|
|
|
|
|
|
@when("I check for type annotations")
|
|
def step_check_type_annotations(context):
|
|
"""Analyze type annotations in the codebase."""
|
|
context.functions_with_hints = 0
|
|
context.functions_without_hints = 0
|
|
context.pydantic_models = 0
|
|
context.regular_dataclasses = 0
|
|
|
|
for py_file in context.src_dir.rglob("*.py"):
|
|
if "discovery" in str(py_file):
|
|
continue # Skip discovery module as per pyright config
|
|
|
|
try:
|
|
tree = ast.parse(py_file.read_text())
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.FunctionDef):
|
|
# Check if function has return type annotation
|
|
if node.returns is not None:
|
|
context.functions_with_hints += 1
|
|
else:
|
|
# Skip __init__, __str__, __repr__ methods
|
|
if not node.name.startswith("_"):
|
|
context.functions_without_hints += 1
|
|
|
|
elif isinstance(node, ast.ClassDef):
|
|
# Check if it's a Pydantic model or dataclass
|
|
for base in node.bases:
|
|
base_name = (
|
|
ast.unparse(base) if hasattr(ast, "unparse") else str(base)
|
|
)
|
|
if "BaseModel" in base_name or "BaseSettings" in base_name:
|
|
context.pydantic_models += 1
|
|
elif "dataclass" in [
|
|
d.id for d in node.decorator_list if isinstance(d, ast.Name)
|
|
]:
|
|
context.regular_dataclasses += 1
|
|
except:
|
|
pass # Skip files with syntax errors
|
|
|
|
|
|
@then("all public functions should have type hints")
|
|
def step_verify_type_hints(context):
|
|
"""Verify public functions have type hints."""
|
|
if context.functions_with_hints + context.functions_without_hints > 0:
|
|
hint_ratio = context.functions_with_hints / (
|
|
context.functions_with_hints + context.functions_without_hints
|
|
)
|
|
assert hint_ratio >= 0.8, f"Only {hint_ratio:.1%} of functions have type hints"
|
|
|
|
|
|
@then("all dataclasses should use Pydantic models")
|
|
def step_verify_pydantic_models(context):
|
|
"""Verify dataclasses use Pydantic."""
|
|
# For now, just verify we're using Pydantic models
|
|
assert context.pydantic_models > 0 or context.regular_dataclasses == 0, (
|
|
"Should use Pydantic models for data validation"
|
|
)
|