cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

feat(validation): add semantic validation service #449

Merged
CoreRasurae merged 1 commits from feature/m6-semantic-validation into master 2026-02-28 19:06:59 +00:00
Conversation 3 Commits 1 Files Changed 11
+3197
11 changed files with 3197 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+3
View File
@@ -2,6 +2,9 @@
## Unreleased
- Added semantic validation service with AST-based rules for syntax errors, missing imports,
broken references, duplicate imports, API misuse, and missing symbols. Includes rule registry,
file-hash LRU cache, severity mapping, and ValidationPipeline integration. (#448)
- Added comprehensive M6 autonomy acceptance test suite covering ACP local facade dispatch
(session/plan/registry/context/event operations), event queue pub/sub with local callbacks
and close semantics, HTTP transport stub rejection, version negotiation, automation profile
benchmarks/semantic_validation_bench.py
+235
View File
@@ -0,0 +1,235 @@
"""ASV benchmarks for the semantic validation service.
Measures the performance of:
- Individual rule checks (syntax, import, reference, cycle, misuse, symbol)
- Rule registry operations (register, lookup, list)
- Validation cache operations (put, get, compute_hash)
- Service orchestration (check_file, as_pipeline_results)
"""
from __future__ import annotations
import sys
from pathlib import Path
from typing import Any
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
from cleveragents.application.services.semantic_validation_rules import (
APIMisuseRule,
BrokenReferenceRule,
DuplicateImportRule,
MissingImportRule,
MissingSymbolRule,
SemanticCheckResult,
SemanticValidationSeverity,
SyntaxCheckRule,
)
from cleveragents.application.services.semantic_validation_service import (
SemanticRuleRegistry,
SemanticValidationCache,
SemanticValidationService,
create_default_registry,
)
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
VALID_PYTHON = """\
import os
import sys
import json
x = 10
def hello():
return x
class Greeter:
def greet(self):
return hello()
def compute():
return x + 1
"""
LARGE_PYTHON = VALID_PYTHON * 50 # ~800 lines
# ---------------------------------------------------------------------------
# Rule benchmark suites
# ---------------------------------------------------------------------------
class SyntaxCheckRuleSuite:
"""Benchmarks for SyntaxCheckRule."""
timeout = 60
def setup(self) -> None:
self.rule = SyntaxCheckRule()
self.source = VALID_PYTHON
self.large = LARGE_PYTHON
def time_syntax_check_small(self) -> None:
self.rule.check(self.source, "test.py")
def time_syntax_check_large(self) -> None:
self.rule.check(self.large, "test.py")
class MissingImportRuleSuite:
"""Benchmarks for MissingImportRule."""
timeout = 60
def setup(self) -> None:
self.rule = MissingImportRule()
self.source = VALID_PYTHON
def time_import_check(self) -> None:
self.rule.check(self.source, "test.py")
class BrokenReferenceRuleSuite:
"""Benchmarks for BrokenReferenceRule."""
timeout = 60
def setup(self) -> None:
self.rule = BrokenReferenceRule()
self.source = VALID_PYTHON
def time_reference_check(self) -> None:
self.rule.check(self.source, "test.py")
class APIMisuseRuleSuite:
"""Benchmarks for APIMisuseRule."""
timeout = 60
def setup(self) -> None:
self.rule = APIMisuseRule()
self.source = VALID_PYTHON
def time_api_misuse_check(self) -> None:
self.rule.check(self.source, "test.py")
class DuplicateImportRuleSuite:
"""Benchmarks for DuplicateImportRule."""
timeout = 60
def setup(self) -> None:
self.rule = DuplicateImportRule()
self.source = VALID_PYTHON
def time_duplicate_import_check(self) -> None:
self.rule.check(self.source, "test.py")
class MissingSymbolRuleSuite:
"""Benchmarks for MissingSymbolRule."""
timeout = 60
def setup(self) -> None:
self.rule = MissingSymbolRule()
self.source = VALID_PYTHON
def time_symbol_check(self) -> None:
self.rule.check(self.source, "test.py")
# ---------------------------------------------------------------------------
# Registry benchmarks
# ---------------------------------------------------------------------------
class RegistrySuite:
"""Benchmarks for SemanticRuleRegistry."""
timeout = 60
def setup(self) -> None:
self.registry = create_default_registry()
def time_list_rules(self) -> None:
self.registry.list_rules()
def time_lookup_rule(self) -> None:
self.registry.get("syntax_error")
def time_all_rules(self) -> None:
self.registry.all_rules()
# ---------------------------------------------------------------------------
# Cache benchmarks
# ---------------------------------------------------------------------------
class CacheSuite:
"""Benchmarks for SemanticValidationCache."""
timeout = 60
def setup(self) -> None:
self.cache = SemanticValidationCache()
self.result = SemanticCheckResult(
passed=True,
message="ok",
severity=SemanticValidationSeverity.INFO,
)
self.cache.put("rule", "hash1", self.result)
def time_compute_hash(self) -> None:
SemanticValidationCache.compute_hash(VALID_PYTHON)
def time_cache_get_hit(self) -> None:
self.cache.get("rule", "hash1")
def time_cache_get_miss(self) -> None:
self.cache.get("rule", "missing")
def time_cache_put(self) -> None:
self.cache.put("rule", "hash2", self.result)
# ---------------------------------------------------------------------------
# Service benchmarks
# ---------------------------------------------------------------------------
class ServiceSuite:
"""Benchmarks for SemanticValidationService orchestration."""
timeout = 120
def setup(self) -> None:
self.service = SemanticValidationService()
self.source = VALID_PYTHON
self.large = LARGE_PYTHON
def time_check_file_small(self) -> None:
# Clear cache to measure full check cost
self.service.cache.clear()
self.service.check_file(self.source, "test.py")
def time_check_file_large(self) -> None:
self.service.cache.clear()
self.service.check_file(self.large, "test.py")
def time_check_file_cached(self) -> None:
# First call populates cache, second uses it
self.service.check_file(self.source, "test.py")
self.service.check_file(self.source, "test.py")
def time_pipeline_results(self) -> None:
self.service.cache.clear()
self.service.as_pipeline_results(self.source, "test.py")
docs/reference/semantic_validation.md
+265
View File
@@ -0,0 +1,265 @@
# Semantic Validation
The **Semantic Validation Service** provides lightweight, AST-based semantic
analysis for Python projects during the strategize/execute phases. It detects
syntax errors, missing imports, broken references, duplicate imports, API
misuse, and missing symbols before code is applied. Non-Python files are
automatically skipped.
## Overview
Semantic checks are exposed as Validation tools attachable per resource via the
Tool Registry. Results integrate into the `ValidationPipeline` as informational
by default, meaning failures are reported but do not block the plan lifecycle
unless configured otherwise.
```
Source Code ──► SemanticValidationService.check_file() ──► [SemanticCheckResult, ...]
│
├── SyntaxCheckRule
├── MissingImportRule
├── BrokenReferenceRule
├── DuplicateImportRule
├── APIMisuseRule
└── MissingSymbolRule
```
> **Note:** `check_file()` returns `[]` immediately for non-Python files
> (files without `.py` or `.pyi` extension).
## Built-in Rules
| Rule | Description | Default Severity |
|--------------------|-------------------------------------------------------|-----------------|
| `syntax_error` | Parses source via `ast.parse` | `error` |
| `missing_import` | Flags imports not in `sys.stdlib_module_names` | `warn` |
| `broken_reference` | Detects names not defined in any scope (scope-aware) | `warn` |
| `duplicate_import` | Finds duplicate relative imports | `error` |
| `api_misuse` | AST-based detection of dangerous calls (see below) | `warn` |
| `missing_symbol` | Finds undefined names in functions and class methods | `info` |
> **Rename:** The rule was previously registered as `dependency_cycle` and is
> now registered as `duplicate_import` to better reflect its purpose.
> The class alias `DependencyCycleRule` remains available for backward
> compatibility.
### APIMisuseRule Dangerous Calls
The `api_misuse` rule uses AST analysis (not regex) to detect calls to:
- `eval()`, `exec()`, `compile()`
- `__import__()`
- `os.system()`, `os.popen()`
- `subprocess.call()`, `subprocess.run()`, `subprocess.Popen()`
- `pickle.load()`, `pickle.loads()`
- `marshal.loads()`
Because the rule inspects `ast.Call` nodes, it does **not** false-positive on
string literals that happen to contain these function names.
## Severity Mapping
Each rule produces findings with a severity level. The severity determines
how the finding integrates with the `ValidationPipeline`:
| Severity | Pipeline Mode | Behaviour |
|----------|------------------|------------------------------------|
| `error` | `required` | Failure **blocks** the operation |
| `warn` | `informational` | Failure is reported, does not block |
| `info` | `informational` | Informational only |
The mapping is configurable via `validation.semantic.severity_mapping`.
## Required vs Informational Attachment Modes
Semantic validations attach to the `ValidationPipeline` in one of two modes:
- **Required** (`error` severity): If a required semantic validation fails,
`ValidationSummary.all_required_passed` becomes `False`, blocking the plan
from proceeding to the Apply phase.
- **Informational** (`warn`/`info` severity): If an informational validation
fails, the finding is logged at INFO level for review but does **not**
affect the `all_required_passed` property.
This allows teams to start with informational semantic checks and
progressively promote them to required as confidence grows.
## Configuration
Three configuration keys control semantic validation:
| Key | Type | Default | Description |
|------------------------------------------|--------|---------|----------------------------------------|
| `validation.semantic.enabled` | `bool` | `true` | Global on/off for semantic validation |
| `validation.semantic.python.enabled` | `bool` | `true` | Python-specific checks on/off |
| `validation.semantic.severity_mapping` | `dict` | (below) | Per-rule severity overrides |
Default severity mapping:
```yaml
validation.semantic.severity_mapping:
syntax_error: error
missing_import: warn
broken_reference: warn
duplicate_import: error
api_misuse: warn
missing_symbol: info
```
### Per-project Configuration
Set `validation.semantic.enabled: false` in the project configuration to
disable semantic validation for a specific project:
```yaml
# .cleveragents/config.yaml
validation:
semantic:
enabled: false
```
### Per-plan Configuration
Override in plan metadata to enable/disable per plan execution:
```yaml
validation:
semantic:
enabled: true
python:
enabled: true
```
## Rule Registry
The `SemanticRuleRegistry` manages pluggable validation rules:
```python
from cleveragents.application.services import (
SemanticRuleRegistry,
create_default_registry,
)
# Create with built-in rules
registry = create_default_registry()
# List registered rules
print(registry.list_rules())
# ['api_misuse', 'broken_reference', 'duplicate_import',
# 'missing_import', 'missing_symbol', 'syntax_error']
#
# Note: 'duplicate_import' is provided by DuplicateImportRule
# Look up a rule
rule = registry.get("syntax_error")
# Register a custom rule
registry.register(my_custom_rule)
```
### Custom Rule Protocol
Any object implementing the `SemanticValidationRule` protocol can be
registered:
```python
class MyCustomRule:
@property
def name(self) -> str:
return "my_custom_check"
def check(self, source: str, filename: str) -> SemanticCheckResult:
# ... run analysis ...
return SemanticCheckResult(
passed=True,
message="All good",
)
```
## Caching
The `SemanticValidationCache` avoids re-running checks on unchanged files
by caching results keyed by `(rule_name, file_sha256)`. The cache uses an
**LRU (Least Recently Used)** eviction policy with a configurable
`max_size` (default: 4096 entries). All operations are **thread-safe**.
```python
from cleveragents.application.services import SemanticValidationCache
cache = SemanticValidationCache(max_size=256)
file_hash = cache.compute_hash(source_code)
# Check cache before running expensive rule
cached = cache.get("syntax_error", file_hash)
if cached is None:
result = rule.check(source_code, filename)
cache.put("syntax_error", file_hash, result)
```
The service automatically uses caching when `check_file()` is called.
Calling `check_file()` with the same source content will return cached
results without re-executing rules.
When the cache reaches `max_size`, the least-recently-used entry is
automatically evicted to make room for the new one.
## Output Schema Normalisation
All semantic check results follow a consistent normalised schema:
```json
{
"passed": true,
"message": "No syntax errors in test.py",
"data": null
}
```
The `data` field contains structured finding details when present:
```json
{
"passed": false,
"message": "Syntax error at test.py:3: invalid syntax",
"data": {
"filename": "test.py",
"line": 3,
"detail": "invalid syntax"
}
}
```
The `normalise_output()` method on `SemanticValidationService` produces
this schema from any `SemanticCheckResult`.
## Pipeline Integration
Use `as_pipeline_results()` to get results compatible with the
`ValidationPipeline`:
```python
service = SemanticValidationService()
results = service.as_pipeline_results(source, "module.py")
# Each result dict has:
# - passed: bool
# - message: str
# - data: dict | None
# - severity: str ("info" | "warn" | "error")
# - mode: str ("required" | "informational")
```
## API Reference
### `SemanticValidationService`
| Method | Description |
|-----------------------|------------------------------------------------|
| `check_file()` | Run all (or selected) rules on source code |
| `as_pipeline_results()` | Same as `check_file` but returns pipeline dicts |
| `normalise_output()` | Normalise a single result to passed/message/data |
| `enabled` | Whether semantic validation is globally enabled |
| `python_enabled` | Whether Python-specific checks are enabled |
| `registry` | The rule registry instance |
| `cache` | The result cache instance |
features/semantic_validation.feature
+467
View File
@@ -0,0 +1,467 @@
Feature: Semantic validation service
As a plan executor
I want semantic validation checks for Python source code
So that syntax errors, missing imports, and broken references are caught early
Background:
Given a semantic validation test environment
# ── Severity enum ──────────────────────────────────────────────
Scenario: SemanticValidationSeverity has three levels
Then the semantic severity enum has values "info", "warn", "error"
# ── SemanticCheckResult model ──────────────────────────────────
Scenario: SemanticCheckResult normalised output has passed message data
Given a semantic check result with passed true and message "All good"
Then the semantic check result passed is true
And the semantic check result message is "All good"
And the semantic check result data is none
Scenario: SemanticCheckResult with failure and data
Given a semantic check result with passed false and message "Error found" and data
Then the semantic check result passed is false
And the semantic check result has data key "detail"
# ── SyntaxCheckRule ────────────────────────────────────────────
Scenario: SyntaxCheckRule passes for valid Python
Given valid Python source code
When the syntax check rule runs
Then the semantic check result passed is true
Scenario: SyntaxCheckRule fails for invalid Python
Given Python source code with a syntax error
When the syntax check rule runs
Then the semantic check result passed is false
And the semantic check result message contains "Syntax error"
And the semantic check result severity is "error"
# ── MissingImportRule ──────────────────────────────────────────
Scenario: MissingImportRule passes for valid imports
Given Python source with standard imports
When the missing import rule runs
Then the semantic check result passed is true
Scenario: MissingImportRule flags suspicious private imports
Given Python source with suspicious private imports
When the missing import rule runs
Then the semantic check result passed is false
And the semantic check result message contains "suspicious import"
Scenario: MissingImportRule skips on syntax error
Given Python source code with a syntax error
When the missing import rule runs
Then the semantic check result passed is true
And the semantic check result message contains "Skipped"
# ── BrokenReferenceRule ────────────────────────────────────────
Scenario: BrokenReferenceRule passes for well-defined code
Given Python source with all names defined
When the broken reference rule runs
Then the semantic check result passed is true
Scenario: BrokenReferenceRule flags undefined names
Given Python source with undefined references
When the broken reference rule runs
Then the semantic check result passed is false
And the semantic check result message contains "broken reference"
# ── DependencyCycleRule ────────────────────────────────────────
Scenario: DependencyCycleRule passes for no duplicate relative imports
Given Python source with unique relative imports
When the dependency cycle rule runs
Then the semantic check result passed is true
Scenario: DependencyCycleRule flags duplicate relative imports
Given Python source with duplicate relative imports
When the dependency cycle rule runs
Then the semantic check result passed is false
And the semantic check result message contains "Duplicate relative imports"
# ── APIMisuseRule ──────────────────────────────────────────────
Scenario: APIMisuseRule passes for clean code
Given Python source without API misuse
When the API misuse rule runs
Then the semantic check result passed is true
Scenario: APIMisuseRule flags eval usage
Given Python source using eval
When the API misuse rule runs
Then the semantic check result passed is false
And the semantic check result message contains "API misuse"
# ── MissingSymbolRule ──────────────────────────────────────────
Scenario: MissingSymbolRule passes for well-scoped functions
Given Python source with well-scoped functions
When the missing symbol rule runs
Then the semantic check result passed is true
Scenario: MissingSymbolRule flags undefined symbols in functions
Given Python source with undefined symbols in functions
When the missing symbol rule runs
Then the semantic check result passed is false
And the semantic check result message contains "missing symbol"
# ── SemanticRuleRegistry ───────────────────────────────────────
Scenario: Default registry has all six built-in rules
Given a default semantic rule registry
Then the registry has 6 rules
And the registry contains rule "syntax_error"
And the registry contains rule "missing_import"
And the registry contains rule "broken_reference"
And the registry contains rule "duplicate_import"
And the registry contains rule "api_misuse"
And the registry contains rule "missing_symbol"
Scenario: Registry lookup returns None for unknown rule
Given a default semantic rule registry
Then the registry returns None for rule "nonexistent"
Scenario: Registry remove returns true for existing rule
Given a default semantic rule registry
When the rule "syntax_error" is removed from the registry
Then the registry has 5 rules
And the registry returns None for rule "syntax_error"
Scenario: Registry remove returns false for unknown rule
Given a default semantic rule registry
Then removing rule "nonexistent" returns false
# ── SemanticValidationCache ────────────────────────────────────
Scenario: Cache returns None for uncached entries
Given a semantic validation cache
Then the cache returns None for rule "check" and hash "abc123"
Scenario: Cache stores and retrieves results by rule and hash
Given a semantic validation cache
And a cached semantic result for rule "check" and hash "abc123"
Then the cache returns the result for rule "check" and hash "abc123"
Scenario: Cache invalidation removes entries
Given a semantic validation cache
And a cached semantic result for rule "check" and hash "abc123"
When the cache entry for rule "check" and hash "abc123" is invalidated
Then the cache returns None for rule "check" and hash "abc123"
Scenario: Cache clear removes all entries
Given a semantic validation cache
And a cached semantic result for rule "a" and hash "h1"
And a cached semantic result for rule "b" and hash "h2"
When the semantic cache is cleared
Then the semantic cache has 0 entries
Scenario: Cache compute_hash returns consistent SHA-256
Then the semantic cache hash for "hello" is consistent
# ── Severity mapping ───────────────────────────────────────────
Scenario: ERROR severity maps to REQUIRED mode
Then semantic severity "error" maps to validation mode "required"
Scenario: WARN severity maps to INFORMATIONAL mode
Then semantic severity "warn" maps to validation mode "informational"
Scenario: INFO severity maps to INFORMATIONAL mode
Then semantic severity "info" maps to validation mode "informational"
Scenario: resolve_severity uses default mapping
Then resolve_severity for "syntax_error" returns "error"
And resolve_severity for "missing_import" returns "warn"
And resolve_severity for "duplicate_import" returns "error"
And resolve_severity for "missing_symbol" returns "info"
Scenario: resolve_severity uses custom mapping
Given a custom severity mapping with "syntax_error" as "warn"
Then resolve_severity with custom mapping for "syntax_error" returns "warn"
Scenario: resolve_severity falls back to info for unknown rule
Then resolve_severity for "unknown_rule" returns "info"
# ── SemanticValidationService ──────────────────────────────────
Scenario: Service check_file runs all rules on valid Python
Given a semantic validation service
And valid Python source code
When the service checks the file
Then all semantic check results passed
Scenario: Service check_file detects syntax error
Given a semantic validation service
And Python source code with a syntax error
When the service checks the file
Then at least one semantic check result failed
And a failing result has message containing "Syntax error"
Scenario: Service check_file with specific rules
Given a semantic validation service
And valid Python source code
When the service checks the file with rules "syntax_error"
Then only 1 semantic check result is returned
Scenario: Service returns empty when disabled
Given a semantic validation service with semantic disabled
And valid Python source code
When the service checks the file
Then 0 semantic check results are returned
Scenario: Service returns empty for Python when python disabled
Given a semantic validation service with python disabled
And valid Python source code
When the service checks the python file
Then 0 semantic check results are returned
Scenario: Service uses cache for repeated checks
Given a semantic validation service
And valid Python source code
When the service checks the file twice
Then the cache has entries
# ── Pipeline integration ───────────────────────────────────────
Scenario: as_pipeline_results returns normalised dicts
Given a semantic validation service
And valid Python source code
When the service returns pipeline results
Then each pipeline result has keys "passed" and "message" and "data"
And each pipeline result has key "severity"
And each pipeline result has key "mode"
Scenario: normalise_output returns standard schema
Given a semantic check result with passed true and message "OK"
When the result is normalised
Then the normalised output has keys "passed" and "message" and "data"
# ── Config keys ────────────────────────────────────────────────
Scenario: Config key constants are defined
Then config key "validation.semantic.enabled" exists
And config key "validation.semantic.python.enabled" exists
And config key "validation.semantic.severity_mapping" exists
# ── Required vs informational attachment modes ─────────────────
Scenario: Error severity findings integrate as required in pipeline
Given a semantic validation service
And Python source code with a syntax error
When the service returns pipeline results
Then a pipeline result with severity "error" has mode "required"
Scenario: Warn severity findings integrate as informational in pipeline
Given a semantic validation service
And Python source using eval
When the service returns pipeline results
Then a pipeline result with severity "warn" has mode "informational"
# ── BrokenReferenceRule scope-awareness ─────────────────────────
Scenario: BrokenReferenceRule passes for function-local variables
Given Python source with function-local variables
When the broken reference rule runs
Then the semantic check result passed is true
Scenario: BrokenReferenceRule passes for annotated assignments
Given Python source with annotated assignment
When the broken reference rule runs
Then the semantic check result passed is true
Scenario: BrokenReferenceRule passes for with-statement variables
Given Python source with with-statement variable
When the broken reference rule runs
Then the semantic check result passed is true
# ── MissingSymbolRule class/nested function support ────────────
Scenario: MissingSymbolRule detects undefined symbols in class methods
Given Python source with a class method using undefined symbol
When the missing symbol rule runs
Then the semantic check result passed is false
And the semantic check result message contains "missing symbol"
Scenario: MissingSymbolRule passes for nested function calls
Given Python source with nested functions
When the missing symbol rule runs
Then the semantic check result passed is true
# ── MissingSymbolRule comprehension scope ───────────────────────
Scenario: MissingSymbolRule does not flag comprehension variables
Given Python source with comprehension variables
When the missing symbol rule runs
Then the semantic check result passed is true
# ── DuplicateImportRule alias ──────────────────────────────────
Scenario: DuplicateImportRule passes for no duplicate relative imports
Given Python source with unique relative imports
When the duplicate import rule runs
Then the semantic check result passed is true
Scenario: DuplicateImportRule flags duplicate relative imports
Given Python source with duplicate relative imports
When the duplicate import rule runs
Then the semantic check result passed is false
And the semantic check result message contains "Duplicate relative imports"
# ── APIMisuseRule AST-based detection ──────────────────────────
Scenario: APIMisuseRule flags exec usage
Given Python source using exec
When the API misuse rule runs
Then the semantic check result passed is false
And the semantic check result message contains "API misuse"
Scenario: APIMisuseRule flags os.popen usage
Given Python source using os.popen
When the API misuse rule runs
Then the semantic check result passed is false
And the semantic check result message contains "API misuse"
Scenario: APIMisuseRule flags subprocess.run usage
Given Python source using subprocess.run
When the API misuse rule runs
Then the semantic check result passed is false
And the semantic check result message contains "API misuse"
Scenario: APIMisuseRule ignores eval in string literals
Given Python source with eval in a string literal
When the API misuse rule runs
Then the semantic check result passed is true
# ── Non-Python file handling ───────────────────────────────────
Scenario: Service returns empty for non-Python files
Given a semantic validation service
And valid Python source code
And a non-Python filename
When the service checks the non-python file
Then 0 semantic check results are returned
# ── Empty source ───────────────────────────────────────────────
Scenario: Service handles empty source code
Given a semantic validation service
And empty Python source code
When the service checks the file
Then all semantic check results passed
# ── Cache with max size ────────────────────────────────────────
Scenario: Cache respects max size with LRU eviction
Given a semantic validation cache with max size 2
And a cached semantic result for rule "a" and hash "h1"
And a cached semantic result for rule "b" and hash "h2"
And a cached semantic result for rule "c" and hash "h3"
Then the semantic cache has 2 entries
# ── Safe initialisation and cleanup ────────────────────────────
Scenario: Cache rejects invalid max size on construction
Then creating a semantic validation cache with max size 0 raises ValueError
Scenario: Cache put updates existing entry without growing
Given a semantic validation cache
And a cached semantic result for rule "r1" and hash "h1"
When the cache entry for rule "r1" and hash "h1" is overwritten with message "updated"
Then the cache returns the updated result for rule "r1" and hash "h1" with message "updated"
And the semantic cache has 1 entries
Scenario: Registry list_rules returns sorted names
Given a default semantic rule registry
Then the registry list_rules returns sorted rule names
Scenario: Service exposes registry property for inspection
Given a semantic validation service
Then the service registry property returns a registry with 6 rules
Scenario: resolve_severity falls back to info for invalid severity value
Given a custom severity mapping with "syntax_error" as "bogus_value"
Then resolve_severity with custom mapping for "syntax_error" returns "info"
Scenario: Service functions with non-dict severity mapping config
Given a semantic validation service with non-dict severity mapping
And valid Python source code
When the service checks the file
Then all semantic check results passed
# ── MissingImportRule from-import form ─────────────────────────
Scenario: MissingImportRule flags suspicious from-import with private module
Given Python source with from-import of private module
When the missing import rule runs
Then the semantic check result passed is false
And the semantic check result message contains "suspicious import"
# ── BrokenReferenceRule advanced scope patterns ────────────────
Scenario: BrokenReferenceRule passes for code with varargs and kwargs
Given Python source with varargs and kwargs function
When the broken reference rule runs
Then the semantic check result passed is true
Scenario: BrokenReferenceRule passes for from-import bindings
Given Python source with from-import bindings
When the broken reference rule runs
Then the semantic check result passed is true
Scenario: BrokenReferenceRule passes for except handler variable
Given Python source with except handler variable
When the broken reference rule runs
Then the semantic check result passed is true
Scenario: BrokenReferenceRule passes for tuple and starred unpacking
Given Python source with tuple and starred unpacking
When the broken reference rule runs
Then the semantic check result passed is true
Scenario: BrokenReferenceRule passes for with-statement without as clause
Given Python source with with-statement without as clause
When the broken reference rule runs
Then the semantic check result passed is true
# ── APIMisuseRule safe attribute on flagged module ──────────────
Scenario: APIMisuseRule passes for safe method call on flagged module
Given Python source calling a safe method on a flagged module
When the API misuse rule runs
Then the semantic check result passed is true
# ── MissingSymbolRule function-local binding patterns ──────────
Scenario: MissingSymbolRule passes for function with all parameter kinds
Given Python source with function using varargs kwonly and kwargs
When the missing symbol rule runs
Then the semantic check result passed is true
Scenario: MissingSymbolRule passes for function with for-loop variable
Given Python source with function containing for-loop variable
When the missing symbol rule runs
Then the semantic check result passed is true
Scenario: MissingSymbolRule passes for function with with-statement variable
Given Python source with function containing with-statement variable
When the missing symbol rule runs
Then the semantic check result passed is true
Scenario: MissingSymbolRule passes for function with except handler variable
Given Python source with function containing except handler variable
When the missing symbol rule runs
Then the semantic check result passed is true
Scenario: MissingSymbolRule passes for function with local import
Given Python source with function containing local import
When the missing symbol rule runs
Then the semantic check result passed is true
Scenario: MissingSymbolRule passes for function with local from-import
Given Python source with function containing local from-import
When the missing symbol rule runs
Then the semantic check result passed is true
features/steps/semantic_validation_steps.py
+696
View File
@@ -0,0 +1,696 @@
"""Step definitions for semantic validation scenarios.
Covers SemanticValidationService, built-in rules, rule registry,
cache, severity mapping, config keys, and pipeline integration.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from behave import given, then, when
if TYPE_CHECKING:
from behave.runner import Context
from cleveragents.application.services.semantic_validation_rules import (
APIMisuseRule,
BrokenReferenceRule,
DependencyCycleRule,
DuplicateImportRule,
MissingImportRule,
MissingSymbolRule,
SemanticCheckResult,
SemanticValidationSeverity,
SyntaxCheckRule,
)
from cleveragents.application.services.semantic_validation_service import (
CONFIG_KEY_ENABLED,
CONFIG_KEY_PYTHON_ENABLED,
CONFIG_KEY_SEVERITY_MAPPING,
DEFAULT_CONFIG,
SemanticValidationCache,
SemanticValidationService,
create_default_registry,
map_severity_to_mode,
resolve_severity,
)
# -- Fixtures ----------------------------------------------------------------
VALID_PY = "import os\n\nx = 10\n\ndef hello():\n return x\n"
SYNTAX_ERR_PY = "def broken(\n return 42\n"
SUSPICIOUS_IMPORT_PY = "import _nonexistent_private_module\n"
STD_IMPORT_PY = "import os\nimport sys\nimport json\n"
ALL_DEFINED_PY = "import os\n\nx = 10\n\ndef foo():\n return x\n"
UNDEF_REF_PY = "def foo():\n return bar_undefined_xyz\n"
UNIQUE_REL_PY = "from . import alpha\nfrom . import beta\n"
DUP_REL_PY = "from . import alpha\nfrom . import alpha\n"
CLEAN_PY = "import json\ndata = json.loads('{}')\n"
EVAL_PY = "result = eval('1 + 2')\n"
SCOPED_PY = "import os\n\ndef greet(name):\n msg = f'Hi {name}'\n return msg\n"
UNDEF_SYM_PY = "def compute(x):\n return x + undefined_thing_xyz\n"
FUNC_LOCAL_PY = "def process(items):\n result = []\n for item in items:\n result.append(item)\n return result\n"
CLASS_METHOD_PY = (
"class MyClass:\n def method(self):\n return undefined_xyz\n"
)
NESTED_FUNC_PY = (
"def outer():\n def inner():\n return 1\n return inner()\n"
)
ANNASSIGN_PY = "x: int = 10\n\ndef foo():\n return x\n"
WITH_STMT_PY = "import io\ndef read_file():\n with io.open('test.txt') as f:\n return f.read()\n"
EXEC_PY = "exec('print(1)')\n"
OS_POPEN_PY = "import os\nos.popen('ls')\n"
SUBPROCESS_RUN_PY = "import subprocess\nsubprocess.run(['echo', 'hello'])\n"
COMPREHENSION_PY = "def summarise(items):\n return [x * 2 for x in items]\n"
NON_PY_FILENAME = "config.yaml"
EMPTY_PY = ""
# -- Fixtures for coverage improvement ----------------------------------------
FROM_PRIVATE_IMPORT_PY = "from _nonexistent_private_mod import something\n"
VARARGS_KWARGS_PY = (
"def func(*args, **kwargs):\n return args, kwargs\n\nfunc(1, key=2)\n"
)
FROM_IMPORT_BINDINGS_PY = "from os.path import join\n\nresult = join('a', 'b')\n"
EXCEPT_HANDLER_PY = "try:\n x = 1\nexcept Exception as exc:\n y = exc\n"
TUPLE_STARRED_PY = "a, b = 1, 2\nc, *d = [3, 4, 5]\nresult = a + b + c\n"
WITH_NO_AS_PY = "import warnings\nwith warnings.catch_warnings():\n x = 1\n"
SAFE_FLAGGED_MODULE_PY = "import os\nresult = os.getcwd()\n"
FUNC_ALL_PARAMS_PY = (
"def func(a, /, b, *args, c=1, **kwargs):\n"
" return a + b + c + len(args) + len(kwargs)\n"
"\nfunc(1, 2, 3, c=4, d=5)\n"
)
FUNC_FOR_LOOP_PY = (
"def total(items):\n"
" s = 0\n"
" for item in items:\n"
" s += item\n"
" return s\n"
)
FUNC_WITH_STMT_PY = (
"def reader(path):\n with open(path) as fh:\n return fh.read()\n"
)
FUNC_EXCEPT_PY = (
"def safe_div(a, b):\n"
" try:\n"
" return a / b\n"
" except ZeroDivisionError as err:\n"
" return err\n"
)
FUNC_LOCAL_IMPORT_PY = "def get_cwd():\n import os\n return os.getcwd()\n"
FUNC_LOCAL_FROM_IMPORT_PY = (
"def get_path():\n from os.path import join\n return join('a', 'b')\n"
)
# -- Background --------------------------------------------------------------
@given("a semantic validation test environment")
def step_env(context: Context) -> None:
context.sv_result = None
context.sv_results = None
context.sv_source = ""
context.sv_filename = "test.py"
context.sv_service = None
context.sv_cache = None
context.sv_registry = None
context.sv_pipeline_results = None
context.sv_normalised = None
context.sv_custom_mapping = None
# -- Severity enum -----------------------------------------------------------
@then('the semantic severity enum has values "info", "warn", "error"')
def step_sev_vals(context: Context) -> None:
assert SemanticValidationSeverity.INFO == "info"
assert SemanticValidationSeverity.WARN == "warn"
assert SemanticValidationSeverity.ERROR == "error"
# -- SemanticCheckResult model -----------------------------------------------
@given('a semantic check result with passed true and message "{msg}"')
def step_cr_pass(context: Context, msg: str) -> None:
context.sv_result = SemanticCheckResult(passed=True, message=msg)
@given('a semantic check result with passed false and message "{msg}" and data')
def step_cr_fail(context: Context, msg: str) -> None:
context.sv_result = SemanticCheckResult(
passed=False,
message=msg,
data={"detail": "x"},
severity=SemanticValidationSeverity.ERROR,
)
@then("the semantic check result passed is true")
def step_r_true(context: Context) -> None:
assert context.sv_result.passed is True
@then("the semantic check result passed is false")
def step_r_false(context: Context) -> None:
assert context.sv_result.passed is False
@then('the semantic check result message is "{msg}"')
def step_r_msg(context: Context, msg: str) -> None:
assert context.sv_result.message == msg
@then("the semantic check result data is none")
def step_r_none(context: Context) -> None:
assert context.sv_result.data is None
@then('the semantic check result has data key "{key}"')
def step_r_dk(context: Context, key: str) -> None:
assert context.sv_result.data is not None and key in context.sv_result.data
@then('the semantic check result message contains "{text}"')
def step_r_mc(context: Context, text: str) -> None:
assert text in context.sv_result.message
@then('the semantic check result severity is "{sev}"')
def step_r_sev(context: Context, sev: str) -> None:
assert context.sv_result.severity.value == sev
# -- Source code given steps -------------------------------------------------
@given("valid Python source code")
def step_src_valid(context: Context) -> None:
context.sv_source = VALID_PY
@given("Python source code with a syntax error")
def step_src_syn(context: Context) -> None:
context.sv_source = SYNTAX_ERR_PY
@given("Python source with standard imports")
def step_src_std(context: Context) -> None:
context.sv_source = STD_IMPORT_PY
@given("Python source with suspicious private imports")
def step_src_sus(context: Context) -> None:
context.sv_source = SUSPICIOUS_IMPORT_PY
@given("Python source with all names defined")
def step_src_def(context: Context) -> None:
context.sv_source = ALL_DEFINED_PY
@given("Python source with undefined references")
def step_src_uref(context: Context) -> None:
context.sv_source = UNDEF_REF_PY
@given("Python source with unique relative imports")
def step_src_urel(context: Context) -> None:
context.sv_source = UNIQUE_REL_PY
@given("Python source with duplicate relative imports")
def step_src_drel(context: Context) -> None:
context.sv_source = DUP_REL_PY
@given("Python source without API misuse")
def step_src_clean(context: Context) -> None:
context.sv_source = CLEAN_PY
@given("Python source using eval")
def step_src_eval(context: Context) -> None:
context.sv_source = EVAL_PY
@given("Python source with well-scoped functions")
def step_src_scoped(context: Context) -> None:
context.sv_source = SCOPED_PY
@given("Python source with undefined symbols in functions")
def step_src_usym(context: Context) -> None:
context.sv_source = UNDEF_SYM_PY
@given("Python source with function-local variables")
def step_src_func_local(context: Context) -> None:
context.sv_source = FUNC_LOCAL_PY
@given("Python source with a class method using undefined symbol")
def step_src_cls_method(context: Context) -> None:
context.sv_source = CLASS_METHOD_PY
@given("Python source with nested functions")
def step_src_nested(context: Context) -> None:
context.sv_source = NESTED_FUNC_PY
@given("Python source with annotated assignment")
def step_src_annassign(context: Context) -> None:
context.sv_source = ANNASSIGN_PY
@given("Python source with with-statement variable")
def step_src_with_stmt(context: Context) -> None:
context.sv_source = WITH_STMT_PY
@given("Python source using exec")
def step_src_exec(context: Context) -> None:
context.sv_source = EXEC_PY
@given("Python source using os.popen")
def step_src_os_popen(context: Context) -> None:
context.sv_source = OS_POPEN_PY
@given("Python source using subprocess.run")
def step_src_subprocess_run(context: Context) -> None:
context.sv_source = SUBPROCESS_RUN_PY
@given("Python source with comprehension variables")
def step_src_comprehension(context: Context) -> None:
context.sv_source = COMPREHENSION_PY
@given("empty Python source code")
def step_src_empty(context: Context) -> None:
context.sv_source = EMPTY_PY
@given("Python source with eval in a string literal")
def step_src_eval_str(context: Context) -> None:
context.sv_source = 'description = "Do not use eval() in production"\n'
@given("a non-Python filename")
def step_non_py_filename(context: Context) -> None:
context.sv_filename = NON_PY_FILENAME
# -- Rule execution ----------------------------------------------------------
@when("the syntax check rule runs")
def step_run_syn(context: Context) -> None:
context.sv_result = SyntaxCheckRule().check(context.sv_source, context.sv_filename)
@when("the missing import rule runs")
def step_run_imp(context: Context) -> None:
context.sv_result = MissingImportRule().check(
context.sv_source, context.sv_filename
)
@when("the broken reference rule runs")
def step_run_ref(context: Context) -> None:
context.sv_result = BrokenReferenceRule().check(
context.sv_source, context.sv_filename
)
@when("the dependency cycle rule runs")
def step_run_cyc(context: Context) -> None:
context.sv_result = DependencyCycleRule().check(
context.sv_source, context.sv_filename
)
@when("the API misuse rule runs")
def step_run_api(context: Context) -> None:
context.sv_result = APIMisuseRule().check(context.sv_source, context.sv_filename)
@when("the missing symbol rule runs")
def step_run_sym(context: Context) -> None:
context.sv_result = MissingSymbolRule().check(
context.sv_source, context.sv_filename
)
@when("the duplicate import rule runs")
def step_run_dup(context: Context) -> None:
context.sv_result = DuplicateImportRule().check(
context.sv_source, context.sv_filename
)
# -- Registry ----------------------------------------------------------------
@given("a default semantic rule registry")
def step_reg(context: Context) -> None:
context.sv_registry = create_default_registry()
@then("the registry has {count:d} rules")
def step_reg_cnt(context: Context, count: int) -> None:
assert len(context.sv_registry) == count
@then('the registry contains rule "{name}"')
def step_reg_has(context: Context, name: str) -> None:
assert context.sv_registry.get(name) is not None
@then('the registry returns None for rule "{name}"')
def step_reg_none(context: Context, name: str) -> None:
assert context.sv_registry.get(name) is None
@when('the rule "{name}" is removed from the registry')
def step_reg_rm(context: Context, name: str) -> None:
context.sv_registry.remove(name)
@then('removing rule "{name}" returns false')
def step_reg_rm_f(context: Context, name: str) -> None:
assert context.sv_registry.remove(name) is False
# -- Cache -------------------------------------------------------------------
@given("a semantic validation cache")
def step_cache(context: Context) -> None:
context.sv_cache = SemanticValidationCache()
@given("a semantic validation cache with max size {n:d}")
def step_cache_max(context: Context, n: int) -> None:
context.sv_cache = SemanticValidationCache(max_size=n)
@given('a cached semantic result for rule "{rule}" and hash "{h}"')
def step_cache_put(context: Context, rule: str, h: str) -> None:
context.sv_cache.put(rule, h, SemanticCheckResult(passed=True, message="cached"))
@then('the cache returns None for rule "{rule}" and hash "{h}"')
def step_cache_miss(context: Context, rule: str, h: str) -> None:
assert context.sv_cache.get(rule, h) is None
@then('the cache returns the result for rule "{rule}" and hash "{h}"')
def step_cache_hit(context: Context, rule: str, h: str) -> None:
r = context.sv_cache.get(rule, h)
assert r is not None and r.message == "cached"
@when('the cache entry for rule "{rule}" and hash "{h}" is invalidated')
def step_cache_inv(context: Context, rule: str, h: str) -> None:
context.sv_cache.invalidate(rule, h)
@when("the semantic cache is cleared")
def step_cache_clr(context: Context) -> None:
context.sv_cache.clear()
@then("the semantic cache has {count:d} entries")
def step_cache_cnt(context: Context, count: int) -> None:
assert len(context.sv_cache) == count
@then('the semantic cache hash for "{text}" is consistent')
def step_cache_hash(context: Context, text: str) -> None:
h1 = SemanticValidationCache.compute_hash(text)
assert h1 == SemanticValidationCache.compute_hash(text) and len(h1) == 64
# -- Severity mapping --------------------------------------------------------
@then('semantic severity "{sev}" maps to validation mode "{mode}"')
def step_sev_map(context: Context, sev: str, mode: str) -> None:
assert map_severity_to_mode(SemanticValidationSeverity(sev)).value == mode
@then('resolve_severity for "{rule}" returns "{expected}"')
def step_resolve(context: Context, rule: str, expected: str) -> None:
assert resolve_severity(rule).value == expected
@given('a custom severity mapping with "{rule}" as "{sev}"')
def step_cust_map(context: Context, rule: str, sev: str) -> None:
context.sv_custom_mapping = {rule: sev}
@then('resolve_severity with custom mapping for "{rule}" returns "{expected}"')
def step_resolve_c(context: Context, rule: str, expected: str) -> None:
assert resolve_severity(rule, context.sv_custom_mapping).value == expected
# -- Service -----------------------------------------------------------------
@given("a semantic validation service")
def step_svc(context: Context) -> None:
context.sv_service = SemanticValidationService()
@given("a semantic validation service with semantic disabled")
def step_svc_off(context: Context) -> None:
context.sv_service = SemanticValidationService(config={CONFIG_KEY_ENABLED: False})
@given("a semantic validation service with python disabled")
def step_svc_pyoff(context: Context) -> None:
context.sv_service = SemanticValidationService(
config={CONFIG_KEY_PYTHON_ENABLED: False},
)
@when("the service checks the file")
def step_svc_chk(context: Context) -> None:
context.sv_results = context.sv_service.check_file(
context.sv_source,
context.sv_filename,
)
@when("the service checks the python file")
def step_svc_chk_py(context: Context) -> None:
context.sv_results = context.sv_service.check_file(context.sv_source, "module.py")
@when('the service checks the file with rules "{rules}"')
def step_svc_chk_r(context: Context, rules: str) -> None:
context.sv_results = context.sv_service.check_file(
context.sv_source,
context.sv_filename,
rule_names=[r.strip() for r in rules.split(",")],
)
@when("the service checks the file twice")
def step_svc_chk2(context: Context) -> None:
context.sv_service.check_file(context.sv_source, context.sv_filename)
context.sv_results = context.sv_service.check_file(
context.sv_source,
context.sv_filename,
)
@when("the service checks the non-python file")
def step_svc_chk_nonpy(context: Context) -> None:
context.sv_filename = NON_PY_FILENAME
context.sv_results = context.sv_service.check_file(
context.sv_source,
context.sv_filename,
)
@then("all semantic check results passed")
def step_all_pass(context: Context) -> None:
assert all(r.passed for r in context.sv_results)
@then("at least one semantic check result failed")
def step_some_fail(context: Context) -> None:
assert any(not r.passed for r in context.sv_results)
@then('a failing result has message containing "{text}"')
def step_fail_msg(context: Context, text: str) -> None:
assert any(text in r.message for r in context.sv_results if not r.passed)
@then("only {count:d} semantic check result is returned")
def step_cnt1(context: Context, count: int) -> None:
assert len(context.sv_results) == count
@then("{count:d} semantic check results are returned")
def step_cnt(context: Context, count: int) -> None:
assert len(context.sv_results) == count
@then("the cache has entries")
def step_cache_has(context: Context) -> None:
assert len(context.sv_service.cache) > 0
# -- Pipeline integration ---------------------------------------------------
@when("the service returns pipeline results")
def step_pipe(context: Context) -> None:
context.sv_pipeline_results = context.sv_service.as_pipeline_results(
context.sv_source,
context.sv_filename,
)
@then('each pipeline result has keys "passed" and "message" and "data"')
def step_pipe_k(context: Context) -> None:
for pr in context.sv_pipeline_results:
assert "passed" in pr and "message" in pr and "data" in pr
@then('each pipeline result has key "{key}"')
def step_pipe_k1(context: Context, key: str) -> None:
for pr in context.sv_pipeline_results:
assert key in pr
@when("the result is normalised")
def step_norm(context: Context) -> None:
context.sv_normalised = SemanticValidationService().normalise_output(
context.sv_result
)
@then('the normalised output has keys "passed" and "message" and "data"')
def step_norm_k(context: Context) -> None:
n = context.sv_normalised
assert "passed" in n and "message" in n and "data" in n
# -- Config keys -------------------------------------------------------------
@then('config key "{key}" exists')
def step_cfg(context: Context, key: str) -> None:
assert key in DEFAULT_CONFIG
# -- Required vs informational -----------------------------------------------
@then('a pipeline result with severity "{sev}" has mode "{mode}"')
def step_pipe_mode(context: Context, sev: str, mode: str) -> None:
matching = [r for r in context.sv_pipeline_results if r["severity"] == sev]
assert len(matching) > 0
assert all(r["mode"] == mode for r in matching)
# -- Safe initialisation and cleanup -----------------------------------------
@then("creating a semantic validation cache with max size 0 raises ValueError")
def step_cache_invalid_size(context: Context) -> None:
raised = False
try:
SemanticValidationCache(max_size=0)
except ValueError:
raised = True
assert raised, "Expected ValueError for max_size=0"
@when(
'the cache entry for rule "{rule}" and hash "{h}" is overwritten '
'with message "{msg}"'
)
def step_cache_overwrite(context: Context, rule: str, h: str, msg: str) -> None:
context.sv_cache.put(rule, h, SemanticCheckResult(passed=True, message=msg))
@then(
'the cache returns the updated result for rule "{rule}" and hash "{h}" '
'with message "{msg}"'
)
def step_cache_updated(context: Context, rule: str, h: str, msg: str) -> None:
r = context.sv_cache.get(rule, h)
assert r is not None and r.message == msg
@then("the registry list_rules returns sorted rule names")
def step_reg_list(context: Context) -> None:
names = context.sv_registry.list_rules()
assert names == sorted(names)
assert len(names) == len(context.sv_registry)
@then("the service registry property returns a registry with {count:d} rules")
def step_svc_reg(context: Context, count: int) -> None:
assert len(context.sv_service.registry) == count
@given("a semantic validation service with non-dict severity mapping")
def step_svc_nondict_sev(context: Context) -> None:
context.sv_service = SemanticValidationService(
config={CONFIG_KEY_SEVERITY_MAPPING: "not-a-dict"},
)
# -- MissingImportRule from-import form --------------------------------------
@given("Python source with from-import of private module")
def step_src_from_priv(context: Context) -> None:
context.sv_source = FROM_PRIVATE_IMPORT_PY
# -- BrokenReferenceRule advanced scope patterns -----------------------------
@given("Python source with varargs and kwargs function")
def step_src_varargs(context: Context) -> None:
context.sv_source = VARARGS_KWARGS_PY
@given("Python source with from-import bindings")
def step_src_from_import(context: Context) -> None:
context.sv_source = FROM_IMPORT_BINDINGS_PY
@given("Python source with except handler variable")
def step_src_except(context: Context) -> None:
context.sv_source = EXCEPT_HANDLER_PY
@given("Python source with tuple and starred unpacking")
def step_src_tuple_star(context: Context) -> None:
context.sv_source = TUPLE_STARRED_PY
@given("Python source with with-statement without as clause")
def step_src_with_no_as(context: Context) -> None:
context.sv_source = WITH_NO_AS_PY
# -- APIMisuseRule safe attribute on flagged module --------------------------
@given("Python source calling a safe method on a flagged module")
def step_src_safe_flagged(context: Context) -> None:
context.sv_source = SAFE_FLAGGED_MODULE_PY
# -- MissingSymbolRule function-local binding patterns -----------------------
@given("Python source with function using varargs kwonly and kwargs")
def step_src_func_all_params(context: Context) -> None:
context.sv_source = FUNC_ALL_PARAMS_PY
@given("Python source with function containing for-loop variable")
def step_src_func_for(context: Context) -> None:
context.sv_source = FUNC_FOR_LOOP_PY
@given("Python source with function containing with-statement variable")
def step_src_func_with(context: Context) -> None:
context.sv_source = FUNC_WITH_STMT_PY
@given("Python source with function containing except handler variable")
def step_src_func_except(context: Context) -> None:
context.sv_source = FUNC_EXCEPT_PY
@given("Python source with function containing local import")
def step_src_func_import(context: Context) -> None:
context.sv_source = FUNC_LOCAL_IMPORT_PY
@given("Python source with function containing local from-import")
def step_src_func_from_import(context: Context) -> None:
context.sv_source = FUNC_LOCAL_FROM_IMPORT_PY
robot/helper_semantic_validation.py
+286
View File
@@ -0,0 +1,286 @@
"""Helper script for Robot Framework semantic validation tests.
Self-contained Python helper that exercises the SemanticValidationService
and related components, printing sentinel strings on success and exiting
with code 1 on failure.
"""
from __future__ import annotations
import sys
from collections.abc import Callable
from pathlib import Path
# Ensure src is importable when run from workspace root
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from cleveragents.application.services.semantic_validation_rules import (
APIMisuseRule,
BrokenReferenceRule,
DuplicateImportRule,
MissingSymbolRule,
SemanticCheckResult,
SemanticValidationSeverity,
SyntaxCheckRule,
)
from cleveragents.application.services.semantic_validation_service import (
CONFIG_KEY_ENABLED,
SemanticValidationCache,
SemanticValidationService,
create_default_registry,
map_severity_to_mode,
)
from cleveragents.domain.models.core.tool import ValidationMode
# ---------------------------------------------------------------------------
# Source code fixtures
# ---------------------------------------------------------------------------
VALID_PYTHON = """\
import os
x = 10
def hello():
return x
"""
SYNTAX_ERROR_PYTHON = """\
def broken(
return 42
"""
EVAL_PYTHON = """\
result = eval("1 + 2")
"""
FUNC_LOCAL_PYTHON = """\
def process(items):
result = []
for item in items:
result.append(item)
return result
"""
CLASS_METHOD_PYTHON = """\
class MyClass:
def method(self):
return undefined_xyz
"""
# ---------------------------------------------------------------------------
# Sub-commands
# ---------------------------------------------------------------------------
def test_models() -> None:
"""Verify SemanticCheckResult and SemanticValidationSeverity."""
result = SemanticCheckResult(
passed=True,
message="All good",
severity=SemanticValidationSeverity.INFO,
)
assert result.passed is True
assert result.message == "All good"
assert result.severity == SemanticValidationSeverity.INFO
assert result.data is None
assert SemanticValidationSeverity.INFO == "info"
assert SemanticValidationSeverity.WARN == "warn"
assert SemanticValidationSeverity.ERROR == "error"
print("models-ok")
def test_syntax_pass() -> None:
"""SyntaxCheckRule passes for valid Python."""
rule = SyntaxCheckRule()
result = rule.check(VALID_PYTHON, "test.py")
assert result.passed is True
print("syntax-pass-ok")
def test_syntax_fail() -> None:
"""SyntaxCheckRule fails for invalid Python."""
rule = SyntaxCheckRule()
result = rule.check(SYNTAX_ERROR_PYTHON, "test.py")
assert result.passed is False
assert "Syntax error" in result.message
print("syntax-fail-ok")
def test_registry() -> None:
"""Default registry has all six built-in rules."""
registry = create_default_registry()
assert len(registry) == 6
names = registry.list_rules()
assert "syntax_error" in names
assert "missing_import" in names
assert "broken_reference" in names
assert "duplicate_import" in names
assert "api_misuse" in names
assert "missing_symbol" in names
print("registry-ok")
def test_cache() -> None:
"""Cache stores and retrieves correctly."""
cache = SemanticValidationCache()
assert cache.get("rule", "hash1") is None
result = SemanticCheckResult(passed=True, message="cached")
cache.put("rule", "hash1", result)
cached = cache.get("rule", "hash1")
assert cached is not None
assert cached.message == "cached"
cache.invalidate("rule", "hash1")
assert cache.get("rule", "hash1") is None
h1 = SemanticValidationCache.compute_hash("hello")
h2 = SemanticValidationCache.compute_hash("hello")
assert h1 == h2
assert len(h1) == 64
print("cache-ok")
def test_service_check() -> None:
"""Service runs all rules."""
service = SemanticValidationService()
results = service.check_file(VALID_PYTHON, "test.py")
assert len(results) > 0
assert all(r.passed for r in results)
print("service-check-ok")
def test_severity_mapping() -> None:
"""Severity maps to correct validation mode."""
assert (
map_severity_to_mode(SemanticValidationSeverity.ERROR)
== ValidationMode.REQUIRED
)
assert (
map_severity_to_mode(SemanticValidationSeverity.WARN)
== ValidationMode.INFORMATIONAL
)
assert (
map_severity_to_mode(SemanticValidationSeverity.INFO)
== ValidationMode.INFORMATIONAL
)
print("severity-mapping-ok")
def test_pipeline_integration() -> None:
"""as_pipeline_results returns dicts with expected keys."""
service = SemanticValidationService()
results = service.as_pipeline_results(VALID_PYTHON, "test.py")
assert len(results) > 0
for r in results:
assert "passed" in r
assert "message" in r
assert "data" in r
assert "severity" in r
assert "mode" in r
print("pipeline-integration-ok")
def test_config_disabled() -> None:
"""Service returns empty when disabled."""
service = SemanticValidationService(config={CONFIG_KEY_ENABLED: False})
results = service.check_file(VALID_PYTHON, "test.py")
assert len(results) == 0
print("config-disabled-ok")
def test_api_misuse() -> None:
"""APIMisuseRule detects eval usage via AST analysis."""
rule = APIMisuseRule()
result = rule.check(EVAL_PYTHON, "test.py")
assert result.passed is False
assert "API misuse" in result.message
print("api-misuse-ok")
def test_broken_ref_scope() -> None:
"""BrokenReferenceRule does not false-positive on function locals."""
rule = BrokenReferenceRule()
result = rule.check(FUNC_LOCAL_PYTHON, "test.py")
assert result.passed is True, f"Expected pass, got: {result.message}"
print("broken-ref-scope-ok")
def test_missing_symbol_class() -> None:
"""MissingSymbolRule detects undefined symbols in class methods."""
rule = MissingSymbolRule()
result = rule.check(CLASS_METHOD_PYTHON, "test.py")
assert result.passed is False, f"Expected fail, got: {result.message}"
assert "missing symbol" in result.message
print("missing-symbol-class-ok")
def test_cache_lru() -> None:
"""Cache evicts LRU entries when max_size is exceeded."""
cache = SemanticValidationCache(max_size=2)
r = SemanticCheckResult(passed=True, message="ok")
cache.put("a", "h1", r)
cache.put("b", "h2", r)
cache.put("c", "h3", r)
assert len(cache) == 2
assert cache.get("a", "h1") is None # evicted
assert cache.get("b", "h2") is not None
assert cache.get("c", "h3") is not None
print("cache-lru-ok")
def test_non_python_skip() -> None:
"""Service returns empty for non-Python files."""
service = SemanticValidationService()
results = service.check_file(VALID_PYTHON, "config.yaml")
assert len(results) == 0
print("non-python-skip-ok")
def test_duplicate_import_alias() -> None:
"""DuplicateImportRule is accessible via DependencyCycleRule alias."""
rule = DuplicateImportRule()
assert rule.name == "duplicate_import"
print("duplicate-import-alias-ok")
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
_COMMANDS: dict[str, Callable[[], None]] = {
"models": test_models,
"syntax_pass": test_syntax_pass,
"syntax_fail": test_syntax_fail,
"registry": test_registry,
"cache": test_cache,
"service_check": test_service_check,
"severity_mapping": test_severity_mapping,
"pipeline_integration": test_pipeline_integration,
"config_disabled": test_config_disabled,
"api_misuse": test_api_misuse,
"broken_ref_scope": test_broken_ref_scope,
"missing_symbol_class": test_missing_symbol_class,
"cache_lru": test_cache_lru,
"non_python_skip": test_non_python_skip,
"duplicate_import_alias": test_duplicate_import_alias,
}
if __name__ == "__main__":
if len(sys.argv) < 2:
print("Usage: helper_semantic_validation.py <command>", file=sys.stderr)
sys.exit(1)
cmd = sys.argv[1]
if cmd not in _COMMANDS:
print(f"Unknown command: {cmd}", file=sys.stderr)
sys.exit(1)
try:
_COMMANDS[cmd]()
except Exception as exc:
print(f"FAILED: {exc}", file=sys.stderr)
sys.exit(1)
robot/semantic_validation.robot
+107
View File
@@ -0,0 +1,107 @@
*** Settings ***
Documentation Integration tests for the semantic validation service
Library Process
Library OperatingSystem
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER_SCRIPT} robot/helper_semantic_validation.py
*** Test Cases ***
Semantic Validation Models Are Valid
[Documentation] Verify SemanticCheckResult and SemanticValidationSeverity models
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} models cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} models-ok
Semantic Validation Syntax Check Passes For Valid Code
[Documentation] SyntaxCheckRule passes for valid Python source
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} syntax_pass cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} syntax-pass-ok
Semantic Validation Syntax Check Fails For Invalid Code
[Documentation] SyntaxCheckRule fails for code with syntax errors
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} syntax_fail cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} syntax-fail-ok
Semantic Validation Registry Has Built-in Rules
[Documentation] Default registry contains all six built-in rules
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} registry cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} registry-ok
Semantic Validation Cache Works Correctly
[Documentation] Cache stores and retrieves results by rule name and hash
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} cache cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} cache-ok
Semantic Validation Service Runs All Rules
[Documentation] SemanticValidationService.check_file runs all registered rules
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} service_check cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} service-check-ok
Semantic Validation Severity Mapping
[Documentation] Severity levels map correctly to validation modes
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} severity_mapping cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} severity-mapping-ok
Semantic Validation Pipeline Integration
[Documentation] as_pipeline_results returns normalised dicts with expected keys
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} pipeline_integration cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} pipeline-integration-ok
Semantic Validation Config Disabled
[Documentation] Service returns empty results when disabled via config
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} config_disabled cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} config-disabled-ok
Semantic Validation API Misuse Detection
[Documentation] APIMisuseRule detects eval() usage
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} api_misuse cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} api-misuse-ok
Semantic Validation Broken Reference Scope Aware
[Documentation] BrokenReferenceRule does not false-positive on function-local variables
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} broken_ref_scope cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} broken-ref-scope-ok
Semantic Validation Missing Symbol Detects Class Methods
[Documentation] MissingSymbolRule detects undefined symbols inside class methods
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} missing_symbol_class cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} missing-symbol-class-ok
Semantic Validation Cache LRU Eviction
[Documentation] Cache evicts least-recently-used entries when max_size exceeded
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} cache_lru cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} cache-lru-ok
Semantic Validation Non Python Files Skipped
[Documentation] Service returns empty results for non-Python files
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} non_python_skip cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} non-python-skip-ok
Semantic Validation Duplicate Import Rule Alias
[Documentation] DuplicateImportRule is accessible via DependencyCycleRule backward-compat alias
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} duplicate_import_alias cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} duplicate-import-alias-ok
Semantic Validation Unknown Command Returns Error
[Documentation] Running with unknown command exits with code 1
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} nonexistent_cmd cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 1
Should Contain ${result.stderr} Unknown command
src/cleveragents/application/services/__init__.py
+40
View File
@@ -23,6 +23,28 @@ from cleveragents.application.services.plan_execution_context import (
RuntimeExecuteActor,
RuntimeExecuteResult,
)
from cleveragents.application.services.semantic_validation_rules import (
APIMisuseRule,
BrokenReferenceRule,
DependencyCycleRule,
DuplicateImportRule,
MissingImportRule,
MissingSymbolRule,
SemanticCheckResult,
SemanticValidationSeverity,
SyntaxCheckRule,
)
from cleveragents.application.services.semantic_validation_service import (
NormalisedOutputDict,
PipelineResultDict,
SemanticRuleRegistry,
SemanticValidationCache,
SemanticValidationRule,
SemanticValidationService,
create_default_registry,
map_severity_to_mode,
resolve_severity,
)
from cleveragents.application.services.session_service import (
PersistentSessionService,
)
@@ -49,23 +71,38 @@ from cleveragents.application.services.validation_pipeline import (
)
__all__ = [
"APIMisuseRule",
"ApplyValidationGate",
"ApplyValidationResult",
"ApplyValidationSummary",
"AttachmentScope",
"BrokenReferenceRule",
"ConfigEntry",
"ConfigLevel",
"ConfigService",
"CorrectionService",
"DecisionService",
"DefaultValidationRunner",
"DependencyCycleRule",
"DuplicateImportRule",
"InvariantService",
"MissingImportRule",
"MissingSymbolRule",
"NormalisedOutputDict",
"PersistentSessionService",
"PipelineResultDict",
"PlanExecutionContext",
"ResolvedValue",
"RuntimeExecuteActor",
"RuntimeExecuteResult",
"SemanticCheckResult",
"SemanticRuleRegistry",
"SemanticValidationCache",
"SemanticValidationRule",
"SemanticValidationService",
"SemanticValidationSeverity",
"SkillRegistryService",
"SyntaxCheckRule",
"ToolRegistryService",
"ValidationAttachment",
"ValidationCommand",
@@ -73,4 +110,7 @@ __all__ = [
"ValidationResult",
"ValidationRunner",
"ValidationSummary",
"create_default_registry",
"map_severity_to_mode",
"resolve_severity",
]
src/cleveragents/application/services/semantic_validation_rules.py
+610
View File
@@ -0,0 +1,610 @@
"""Built-in semantic validation rules for Python projects.
Provides pluggable rule implementations for syntax errors, missing
imports, broken references, duplicate relative imports, API misuse,
and missing symbols. Each rule conforms to the ``SemanticValidationRule``
protocol.
Rules are designed to be lightweight, AST-based heuristics that
avoid executing user code.
Based on ``docs/specification.md`` and ADR-013 (Validation Abstraction).
"""
from __future__ import annotations
import ast
import builtins as _builtins_mod
import sys
from enum import StrEnum
from typing import ClassVar
from pydantic import BaseModel, ConfigDict, Field
# ---------------------------------------------------------------------------
# Shared severity enum + result model (canonical definitions)
# ---------------------------------------------------------------------------
class SemanticValidationSeverity(StrEnum):
"""Severity level for semantic validation findings."""
INFO = "info"
WARN = "warn"
ERROR = "error"
class SemanticCheckResult(BaseModel):
"""Normalised result from a single semantic check.
Attributes:
passed: Whether the check passed (no issues found).
message: Human-readable description of the result.
data: Optional structured data with finding details.
severity: Severity level of the finding.
"""
passed: bool = Field(..., description="Whether the check passed")
message: str = Field(..., description="Human-readable result description")
data: dict[str, object] | None = Field(
default=None, description="Optional structured finding data"
)
severity: SemanticValidationSeverity = Field(
default=SemanticValidationSeverity.INFO,
description="Severity of the finding",
)
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
)
# ---------------------------------------------------------------------------
# Built-in rules
# ---------------------------------------------------------------------------
_BUILTINS_SET: frozenset[str] = frozenset(dir(_builtins_mod))
# Use the authoritative stdlib module list (Python 3.10+) instead of a
# hand-maintained subset so that internal modules like ``_thread``,
# ``_abc``, etc. are recognised and do not trigger false positives.
_STDLIB_MODULE_NAMES: frozenset[str] = frozenset(sys.stdlib_module_names)
class SyntaxCheckRule:
"""Check Python source for syntax errors via ``ast.parse``."""
@property
def name(self) -> str:
return "syntax_error"
def check(self, source: str, filename: str) -> SemanticCheckResult:
try:
ast.parse(source, filename=filename)
except SyntaxError as exc:
lineno = exc.lineno or 0
return SemanticCheckResult(
passed=False,
message=f"Syntax error at {filename}:{lineno}: {exc.msg}",
data={"filename": filename, "line": lineno, "detail": exc.msg},
severity=SemanticValidationSeverity.ERROR,
)
return SemanticCheckResult(
passed=True,
message=f"No syntax errors in {filename}",
severity=SemanticValidationSeverity.INFO,
)
class MissingImportRule:
"""Detect import statements that reference suspicious modules.
Uses ``ast`` to find import names that look like private
internal modules (leading underscore) that are not in the
standard library. Uses ``sys.stdlib_module_names`` for an
authoritative list of stdlib top-level modules.
"""
@property
def name(self) -> str:
return "missing_import"
def check(self, source: str, filename: str) -> SemanticCheckResult:
try:
tree = ast.parse(source, filename=filename)
except SyntaxError:
return SemanticCheckResult(
passed=True,
message="Skipped import check (syntax error)",
severity=SemanticValidationSeverity.INFO,
)
suspicious: list[dict[str, object]] = []
for node in ast.walk(tree):
if isinstance(node, ast.Import):
for alias in node.names:
top = alias.name.split(".")[0]
if top.startswith("_") and top not in _STDLIB_MODULE_NAMES:
suspicious.append({"module": alias.name, "line": node.lineno})
elif (
isinstance(node, ast.ImportFrom)
and node.module
and node.module.startswith("_")
):
top = node.module.split(".")[0]
if top not in _STDLIB_MODULE_NAMES:
suspicious.append({"module": node.module, "line": node.lineno})
if suspicious:
return SemanticCheckResult(
passed=False,
message=(f"Found {len(suspicious)} suspicious import(s) in {filename}"),
data={"filename": filename, "imports": suspicious},
severity=SemanticValidationSeverity.WARN,
)
return SemanticCheckResult(
passed=True,
message=f"No missing imports detected in {filename}",
severity=SemanticValidationSeverity.INFO,
)
# ---------------------------------------------------------------------------
# Scope-aware name collection helpers
# ---------------------------------------------------------------------------
def _walk_and_collect(tree: ast.Module, *, deep: bool) -> set[str]:
"""Shared name-collection walker.
When *deep* is ``False`` only direct children of *tree* are
inspected (module-scope bindings). When *deep* is ``True``
the entire AST is walked to capture bindings at every scope
level, including ``ast.Name`` nodes in ``Store`` context
(e.g. comprehension targets).
"""
names: set[str] = set()
iterator = ast.walk(tree) if deep else ast.iter_child_nodes(tree)
for node in iterator:
_collect_binding_names(node, names)
if deep and isinstance(node, ast.Name) and isinstance(node.ctx, ast.Store):
names.add(node.id)
return names
def _collect_defined_names(tree: ast.Module) -> set[str]:
"""Collect names defined at module scope.
Handles function/class definitions, imports, assignments
(including annotated and augmented), for-loop targets,
with-statement variables, and exception handler names.
Uses :func:`_walk_and_collect` with ``deep=False`` to restrict
collection to top-level bindings only.
"""
return _walk_and_collect(tree, deep=False)
def _collect_binding_names(node: ast.AST, out: set[str]) -> None:
"""Add names bound by *node* to *out*.
Recognised binding forms: ``FunctionDef``/``AsyncFunctionDef``
(name **and** parameters), ``ClassDef``, ``Import``,
``ImportFrom``, ``Assign``, ``AnnAssign``, ``AugAssign``,
``For``/``AsyncFor`` targets, ``With``/``AsyncWith`` optional
vars, and ``ExceptHandler`` names.
"""
if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef):
out.add(node.name)
# Collect parameter names (positional, keyword-only, vararg, kwarg)
for arg in node.args.args + node.args.posonlyargs + node.args.kwonlyargs:
out.add(arg.arg)
if node.args.vararg:
out.add(node.args.vararg.arg)
if node.args.kwarg:
out.add(node.args.kwarg.arg)
elif isinstance(node, ast.ClassDef):
out.add(node.name)
elif isinstance(node, ast.Assign):
for target in node.targets:
_collect_target_names(target, out)
elif isinstance(node, (ast.AnnAssign, ast.AugAssign)) and node.target:
_collect_target_names(node.target, out)
elif isinstance(node, ast.Import):
for alias in node.names:
n = alias.asname if alias.asname else alias.name.split(".")[0]
out.add(n)
elif isinstance(node, ast.ImportFrom):
for alias in node.names:
n = alias.asname if alias.asname else alias.name
out.add(n)
elif isinstance(node, ast.For | ast.AsyncFor):
_collect_target_names(node.target, out)
elif isinstance(node, ast.With | ast.AsyncWith):
for item in node.items:
if item.optional_vars:
_collect_target_names(item.optional_vars, out)
elif isinstance(node, ast.ExceptHandler) and node.name:
out.add(node.name)
def _collect_target_names(target: ast.AST, out: set[str]) -> None:
"""Recursively extract names from an assignment target."""
if isinstance(target, ast.Name):
out.add(target.id)
elif isinstance(target, ast.Tuple | ast.List):
for elt in target.elts:
_collect_target_names(elt, out)
elif isinstance(target, ast.Starred):
_collect_target_names(target.value, out)
def _collect_all_scope_names(tree: ast.Module) -> set[str]:
"""Collect every name defined anywhere in the module.
Walks the entire AST to find all binding forms at every scope
level. Used by ``BrokenReferenceRule`` so that names defined
inside functions, classes, loops, etc. are not falsely flagged
as broken references.
Uses :func:`_walk_and_collect` with ``deep=True`` to recurse
through all nested scopes.
"""
return _walk_and_collect(tree, deep=True)
class BrokenReferenceRule:
"""Detect references to names not defined anywhere in the module.
Uses scope-aware collection that walks the entire AST so that
names defined inside functions, classes, loops, comprehensions,
and exception handlers are properly recognised.
"""
@property
def name(self) -> str:
return "broken_reference"
def check(self, source: str, filename: str) -> SemanticCheckResult:
try:
tree = ast.parse(source, filename=filename)
except SyntaxError:
return SemanticCheckResult(
passed=True,
message="Skipped reference check (syntax error)",
severity=SemanticValidationSeverity.INFO,
)
defined = _collect_all_scope_names(tree)
defined.update(_BUILTINS_SET)
used: set[str] = set()
for node in ast.walk(tree):
if isinstance(node, ast.Name) and isinstance(node.ctx, ast.Load):
used.add(node.id)
broken = used - defined
broken -= {
"__name__",
"__file__",
"__doc__",
"__all__",
"__annotations__",
"__spec__",
"__loader__",
"__package__",
"__builtins__",
"__cached__",
"__path__",
}
if broken:
limited = sorted(broken)[:10]
return SemanticCheckResult(
passed=False,
message=(
f"Found {len(broken)} potentially broken reference(s) "
f"in {filename}: {', '.join(limited)}"
),
data={"filename": filename, "broken": limited},
severity=SemanticValidationSeverity.WARN,
)
return SemanticCheckResult(
passed=True,
message=f"No broken references in {filename}",
severity=SemanticValidationSeverity.INFO,
)
class DuplicateImportRule:
"""Detect duplicate relative imports that hint at dependency issues.
Scans a single file for repeated relative import statements.
Duplicate relative imports within one module are a code smell
that *may* indicate copy-paste errors or structural problems.
.. note::
This rule does **not** perform cross-file cycle detection.
It only flags duplicate relative imports within a single
file.
"""
@property
def name(self) -> str:
return "duplicate_import"
def check(self, source: str, filename: str) -> SemanticCheckResult:
try:
tree = ast.parse(source, filename=filename)
except SyntaxError:
return SemanticCheckResult(
passed=True,
message="Skipped duplicate import check (syntax error)",
severity=SemanticValidationSeverity.INFO,
)
relative_imports: list[str] = []
for node in ast.walk(tree):
if isinstance(node, ast.ImportFrom) and node.level and node.level > 0:
base = node.module or ""
names = ",".join(a.name for a in (node.names or []))
relative_imports.append(f"{base}:{names}")
if len(relative_imports) != len(set(relative_imports)):
duplicates = [
m for m in set(relative_imports) if relative_imports.count(m) > 1
]
return SemanticCheckResult(
passed=False,
message=(
f"Duplicate relative imports in {filename} "
f"(potential cycle): {', '.join(duplicates)}"
),
data={"filename": filename, "duplicates": duplicates},
severity=SemanticValidationSeverity.ERROR,
)
return SemanticCheckResult(
passed=True,
message=f"No duplicate relative imports in {filename}",
severity=SemanticValidationSeverity.INFO,
)
# Keep the old name as a public alias for backwards compatibility.
DependencyCycleRule = DuplicateImportRule
class APIMisuseRule:
"""Detect common API misuse patterns in Python code.
Uses AST analysis to detect calls to dangerous functions such as
``eval()``, ``exec()``, ``os.system()``, ``os.popen()``,
``pickle.load()``, ``pickle.loads()``, ``marshal.loads()``,
``compile()``, and ``__import__()``. AST-based detection avoids
false positives from string literals and comments.
"""
# Dangerous bare function names (resolved via ast.Name nodes).
_DANGEROUS_NAMES: ClassVar[frozenset[str]] = frozenset(
{"eval", "exec", "compile", "__import__"}
)
_DANGEROUS_NAME_DESCRIPTIONS: ClassVar[dict[str, str]] = {
"eval": "Use of eval() is discouraged",
"exec": "Use of exec() is discouraged",
"compile": "Use of compile() can execute arbitrary code",
"__import__": "Use importlib instead of __import__()",
}
# Dangerous qualified calls (module.func patterns).
_DANGEROUS_ATTRS: ClassVar[dict[str, dict[str, str]]] = {
"os": {
"system": "Use subprocess instead of os.system()",
"popen": "Use subprocess instead of os.popen()",
},
"subprocess": {
"call": "subprocess.call can execute arbitrary commands",
"run": "subprocess.run can execute arbitrary commands",
"Popen": "subprocess.Popen can execute arbitrary commands",
},
"pickle": {
"load": "pickle.load is unsafe with untrusted data",
"loads": "pickle.loads is unsafe with untrusted data",
},
"marshal": {
"loads": "marshal.loads is unsafe with untrusted data",
},
}
@property
def name(self) -> str:
return "api_misuse"
def check(self, source: str, filename: str) -> SemanticCheckResult:
try:
tree = ast.parse(source, filename=filename)
except SyntaxError:
return SemanticCheckResult(
passed=True,
message="Skipped API misuse check (syntax error)",
severity=SemanticValidationSeverity.INFO,
)
findings: list[dict[str, object]] = []
for node in ast.walk(tree):
if not isinstance(node, ast.Call):
continue
func = node.func
# Bare name calls: eval(...), exec(...), compile(...), __import__(...)
if isinstance(func, ast.Name) and func.id in self._DANGEROUS_NAMES:
findings.append(
{
"line": node.lineno,
"call": func.id,
"description": self._DANGEROUS_NAME_DESCRIPTIONS[func.id],
}
)
# Attribute calls: os.system(...), pickle.loads(...)
elif (
isinstance(func, ast.Attribute)
and isinstance(func.value, ast.Name)
and func.value.id in self._DANGEROUS_ATTRS
):
attr_map = self._DANGEROUS_ATTRS[func.value.id]
if func.attr in attr_map:
findings.append(
{
"line": node.lineno,
"call": f"{func.value.id}.{func.attr}",
"description": attr_map[func.attr],
}
)
if findings:
return SemanticCheckResult(
passed=False,
message=(f"Found {len(findings)} API misuse pattern(s) in {filename}"),
data={"filename": filename, "findings": findings},
severity=SemanticValidationSeverity.WARN,
)
return SemanticCheckResult(
passed=True,
message=f"No API misuse patterns in {filename}",
severity=SemanticValidationSeverity.INFO,
)
# ---------------------------------------------------------------------------
# MissingSymbolRule helpers
# ---------------------------------------------------------------------------
def _collect_function_local_names(
node: ast.FunctionDef | ast.AsyncFunctionDef,
) -> set[str]:
"""Collect all names locally bound inside a function body.
Includes parameters, assignments, for-loop targets,
with-statement variables, exception handler names, nested
function/class definitions, and comprehension variables.
"""
local_names: set[str] = set()
# Parameters
for arg in node.args.args:
local_names.add(arg.arg)
for arg in node.args.posonlyargs:
local_names.add(arg.arg)
for arg in node.args.kwonlyargs:
local_names.add(arg.arg)
if node.args.vararg:
local_names.add(node.args.vararg.arg)
if node.args.kwarg:
local_names.add(node.args.kwarg.arg)
# Walk the body for all binding forms
for child in ast.walk(node):
if child is node:
continue
# Nested function/class definitions bind their name
if isinstance(child, ast.FunctionDef | ast.AsyncFunctionDef | ast.ClassDef):
local_names.add(child.name)
# Store context names (assignments, comprehension targets)
elif isinstance(child, ast.Name) and isinstance(child.ctx, ast.Store):
local_names.add(child.id)
# For-loop targets
elif isinstance(child, ast.For | ast.AsyncFor):
_collect_target_names(child.target, local_names)
# With-statement variables
elif isinstance(child, ast.With | ast.AsyncWith):
for item in child.items:
if item.optional_vars:
_collect_target_names(item.optional_vars, local_names)
# Exception handler names
elif isinstance(child, ast.ExceptHandler) and child.name:
local_names.add(child.name)
# Import inside function
elif isinstance(child, ast.Import):
for alias in child.names:
n = alias.asname if alias.asname else alias.name.split(".")[0]
local_names.add(n)
elif isinstance(child, ast.ImportFrom):
for alias in child.names:
n = alias.asname if alias.asname else alias.name
local_names.add(n)
return local_names
def _iter_functions(
tree: ast.AST,
) -> list[ast.FunctionDef | ast.AsyncFunctionDef]:
"""Yield all function/method definitions at any nesting depth."""
results: list[ast.FunctionDef | ast.AsyncFunctionDef] = []
for node in ast.walk(tree):
if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef):
results.append(node)
return results
class MissingSymbolRule:
"""Detect undefined name usage in function and method bodies.
Inspects all functions and methods at any nesting depth
(including class methods and nested functions) for names that
are used but not defined locally, at module scope, or in
builtins.
"""
@property
def name(self) -> str:
return "missing_symbol"
def check(self, source: str, filename: str) -> SemanticCheckResult:
try:
tree = ast.parse(source, filename=filename)
except SyntaxError:
return SemanticCheckResult(
passed=True,
message="Skipped symbol check (syntax error)",
severity=SemanticValidationSeverity.INFO,
)
module_names = _collect_defined_names(tree)
functions = _iter_functions(tree)
missing: list[dict[str, object]] = []
for func_node in functions:
local_names = _collect_function_local_names(func_node)
for child in ast.walk(func_node):
if isinstance(child, ast.Name) and isinstance(child.ctx, ast.Load):
nm = child.id
if (
nm not in local_names
and nm not in module_names
and nm not in _BUILTINS_SET
):
missing.append(
{
"function": func_node.name,
"symbol": nm,
"line": child.lineno,
}
)
if missing:
limited = missing[:10]
return SemanticCheckResult(
passed=False,
message=f"Found {len(missing)} missing symbol(s) in {filename}",
data={"filename": filename, "missing": limited},
severity=SemanticValidationSeverity.INFO,
)
return SemanticCheckResult(
passed=True,
message=f"No missing symbols in {filename}",
severity=SemanticValidationSeverity.INFO,
)
src/cleveragents/application/services/semantic_validation_service.py
+445
View File
@@ -0,0 +1,445 @@
"""Semantic validation service for CleverAgents.
Provides semantic analysis checks for Python projects during the
strategize/execute phases: syntax errors, missing imports, broken
references, duplicate relative imports, API misuse, and missing
symbols.
Semantic checks are exposed as Validation tools attachable per
resource via the Tool Registry. Results integrate into the
``ValidationPipeline`` as informational by default.
Architecture:
- ``SemanticValidationSeverity``: INFO / WARN / ERROR severity levels
- ``SemanticCheckResult``: normalised result with passed/message/data/severity
- ``SemanticValidationRule``: protocol for pluggable rule implementations
- Built-in rules in ``semantic_validation_rules`` companion module
- ``SemanticRuleRegistry``: register/lookup/list rules by name
- ``SemanticValidationCache``: file-hash keyed LRU cache to skip unchanged files
- ``SemanticValidationService``: orchestrator integrating with the pipeline
- Config keys: ``validation.semantic.enabled``,
``validation.semantic.python.enabled``,
``validation.semantic.severity_mapping``
Based on ``docs/specification.md`` and ADR-013 (Validation Abstraction).
"""
from __future__ import annotations
import hashlib
import threading
from collections import OrderedDict
from collections.abc import Mapping
from typing import Protocol, TypedDict, runtime_checkable
import structlog
from cleveragents.application.services.semantic_validation_rules import (
APIMisuseRule,
BrokenReferenceRule,
DuplicateImportRule,
MissingImportRule,
MissingSymbolRule,
SemanticCheckResult,
SemanticValidationSeverity,
SyntaxCheckRule,
)
from cleveragents.domain.models.core.tool import ValidationMode
logger = structlog.get_logger(__name__)
# ---------------------------------------------------------------------------
# Typed dicts for public return types
# ---------------------------------------------------------------------------
class PipelineResultDict(TypedDict):
"""Typed dict for pipeline-compatible result dictionaries."""
passed: bool
message: str
data: dict[str, object] | None
severity: str
mode: str
class NormalisedOutputDict(TypedDict):
"""Typed dict for normalised output dictionaries."""
passed: bool
message: str
data: dict[str, object] | None
# ---------------------------------------------------------------------------
# Config key constants
# ---------------------------------------------------------------------------
CONFIG_KEY_ENABLED = "validation.semantic.enabled"
CONFIG_KEY_PYTHON_ENABLED = "validation.semantic.python.enabled"
CONFIG_KEY_SEVERITY_MAPPING = "validation.semantic.severity_mapping"
_DEFAULT_SEVERITY_MAPPING: dict[str, str] = {
"syntax_error": "error",
"missing_import": "warn",
"broken_reference": "warn",
"duplicate_import": "error",
"api_misuse": "warn",
"missing_symbol": "info",
}
DEFAULT_CONFIG: dict[str, object] = {
CONFIG_KEY_ENABLED: True,
CONFIG_KEY_PYTHON_ENABLED: True,
CONFIG_KEY_SEVERITY_MAPPING: _DEFAULT_SEVERITY_MAPPING,
}
# ---------------------------------------------------------------------------
# Rule protocol
# ---------------------------------------------------------------------------
@runtime_checkable
class SemanticValidationRule(Protocol):
"""Protocol for pluggable semantic validation rules.
Each rule has a unique ``name`` and a ``check`` method that
inspects source code and returns a normalised result.
"""
@property
def name(self) -> str:
"""Unique name of this validation rule."""
... # pragma: no cover
def check(self, source: str, filename: str) -> SemanticCheckResult:
"""Run the semantic check on source code.
Args:
source: Python source code to check.
filename: Filename (for error messages).
Returns:
SemanticCheckResult with pass/fail and details.
"""
... # pragma: no cover
# ---------------------------------------------------------------------------
# Rule registry
# ---------------------------------------------------------------------------
class SemanticRuleRegistry:
"""Registry for semantic validation rules.
Supports registration, lookup by name, and listing of all
registered rules.
"""
def __init__(self) -> None:
self._rules: dict[str, SemanticValidationRule] = {}
def register(self, rule: SemanticValidationRule) -> None:
"""Register a rule. Overwrites if name already exists."""
self._rules[rule.name] = rule
logger.debug("Registered semantic rule: %s", rule.name)
def get(self, name: str) -> SemanticValidationRule | None:
"""Look up a rule by name."""
return self._rules.get(name)
def list_rules(self) -> list[str]:
"""Return sorted list of registered rule names."""
return sorted(self._rules.keys())
def all_rules(self) -> list[SemanticValidationRule]:
"""Return all registered rules sorted by name."""
return [self._rules[n] for n in sorted(self._rules.keys())]
def remove(self, name: str) -> bool:
"""Remove a rule by name. Returns True if found."""
if name in self._rules:
del self._rules[name]
return True
return False
def __len__(self) -> int:
return len(self._rules)
def create_default_registry() -> SemanticRuleRegistry:
"""Create a registry pre-loaded with all built-in Python rules."""
registry = SemanticRuleRegistry()
registry.register(SyntaxCheckRule())
registry.register(MissingImportRule())
registry.register(BrokenReferenceRule())
registry.register(DuplicateImportRule())
registry.register(APIMisuseRule())
registry.register(MissingSymbolRule())
return registry
# ---------------------------------------------------------------------------
# Cache
# ---------------------------------------------------------------------------
_DEFAULT_CACHE_MAX_SIZE = 4096
class SemanticValidationCache:
"""File-hash keyed LRU cache for semantic check results.
Avoids re-running checks on unchanged files by caching results
keyed by ``(rule_name, file_hash)``. Uses a bounded
``OrderedDict`` with LRU eviction to prevent unbounded memory
growth. All public methods are thread-safe.
Args:
max_size: Maximum number of entries before LRU eviction
kicks in. Defaults to ``4096``.
"""
def __init__(self, max_size: int = _DEFAULT_CACHE_MAX_SIZE) -> None:
if max_size < 1:
raise ValueError("max_size must be >= 1")
self._max_size = max_size
self._cache: OrderedDict[str, SemanticCheckResult] = OrderedDict()
self._lock = threading.Lock()
@staticmethod
def compute_hash(content: str) -> str:
"""Compute SHA-256 hex digest of source content."""
return hashlib.sha256(content.encode("utf-8")).hexdigest()
def _key(self, rule_name: str, file_hash: str) -> str:
return f"{rule_name}:{file_hash}"
def get(self, rule_name: str, file_hash: str) -> SemanticCheckResult | None:
"""Retrieve a cached result, or None if not cached."""
key = self._key(rule_name, file_hash)
with self._lock:
if key in self._cache:
self._cache.move_to_end(key)
return self._cache[key]
return None
def put(self, rule_name: str, file_hash: str, result: SemanticCheckResult) -> None:
"""Store a result in the cache, evicting LRU entries if needed."""
key = self._key(rule_name, file_hash)
with self._lock:
if key in self._cache:
self._cache.move_to_end(key)
self._cache[key] = result
while len(self._cache) > self._max_size:
self._cache.popitem(last=False)
def invalidate(self, rule_name: str, file_hash: str) -> None:
"""Remove a cached entry."""
key = self._key(rule_name, file_hash)
with self._lock:
self._cache.pop(key, None)
def clear(self) -> None:
"""Clear all cached results."""
with self._lock:
self._cache.clear()
def __len__(self) -> int:
with self._lock:
return len(self._cache)
# ---------------------------------------------------------------------------
# Severity mapping
# ---------------------------------------------------------------------------
def map_severity_to_mode(
severity: SemanticValidationSeverity,
) -> ValidationMode:
"""Map a semantic severity to a ValidationPipeline mode.
- ERROR -> REQUIRED (failure blocks)
- WARN / INFO -> INFORMATIONAL (failure reported only)
"""
if severity == SemanticValidationSeverity.ERROR:
return ValidationMode.REQUIRED
return ValidationMode.INFORMATIONAL
def resolve_severity(
rule_name: str,
severity_mapping: dict[str, str] | None = None,
) -> SemanticValidationSeverity:
"""Resolve the configured severity for a rule.
Falls back to the default mapping if no override is provided.
"""
mapping = severity_mapping or _DEFAULT_SEVERITY_MAPPING
raw = mapping.get(rule_name, "info")
try:
return SemanticValidationSeverity(raw)
except ValueError:
return SemanticValidationSeverity.INFO
# ---------------------------------------------------------------------------
# Service
# ---------------------------------------------------------------------------
_PYTHON_EXTENSIONS: frozenset[str] = frozenset({".py", ".pyi"})
class SemanticValidationService:
"""Orchestrator for semantic validation checks.
Integrates with the ``ValidationPipeline`` by producing
normalised ``SemanticCheckResult`` objects and converting them
to pipeline-compatible dicts. Uses a rule registry and cache
for extensibility and efficiency.
Args:
registry: Rule registry to use (default: built-in rules).
cache: Cache instance (default: new empty cache).
config: Configuration overrides (default: DEFAULT_CONFIG).
"""
def __init__(
self,
registry: SemanticRuleRegistry | None = None,
cache: SemanticValidationCache | None = None,
config: Mapping[str, object] | None = None,
) -> None:
self._registry = registry or create_default_registry()
self._cache = cache or SemanticValidationCache()
self._config = {**DEFAULT_CONFIG, **(config or {})}
@property
def registry(self) -> SemanticRuleRegistry:
"""The rule registry used by this service."""
return self._registry
@property
def cache(self) -> SemanticValidationCache:
"""The result cache used by this service."""
return self._cache
@property
def enabled(self) -> bool:
"""Whether semantic validation is globally enabled."""
return bool(self._config.get(CONFIG_KEY_ENABLED, True))
@property
def python_enabled(self) -> bool:
"""Whether Python-specific semantic checks are enabled."""
return bool(self._config.get(CONFIG_KEY_PYTHON_ENABLED, True))
@staticmethod
def _is_python_file(filename: str) -> bool:
"""Return True if *filename* has a Python extension."""
return filename.endswith((".py", ".pyi"))
def check_file(
self,
source: str,
filename: str,
rule_names: list[str] | None = None,
) -> list[SemanticCheckResult]:
"""Run semantic checks on a single file.
Args:
source: Python source code.
filename: Filename for error messages.
rule_names: Optional list of specific rules to run.
If None, all registered rules are run.
Returns:
List of SemanticCheckResult objects.
"""
if not self.enabled:
return []
# All built-in rules are Python-specific. Skip non-Python
# files entirely when Python checks are disabled, and also
# skip non-Python files unconditionally since the rules use
# ``ast.parse`` which only handles Python.
if not self._is_python_file(filename):
return []
if not self.python_enabled:
return []
file_hash = self._cache.compute_hash(source)
rules = (
[r for n in rule_names if (r := self._registry.get(n)) is not None]
if rule_names
else self._registry.all_rules()
)
results: list[SemanticCheckResult] = []
raw_mapping = self._config.get(CONFIG_KEY_SEVERITY_MAPPING)
severity_mapping: dict[str, str] | None = None
if isinstance(raw_mapping, dict):
severity_mapping = {str(k): str(v) for k, v in raw_mapping.items()}
for rule in rules:
cached = self._cache.get(rule.name, file_hash)
if cached is not None:
results.append(cached)
continue
result = rule.check(source, filename)
# Apply configured severity override
configured = resolve_severity(rule.name, severity_mapping)
if result.severity != configured:
result = SemanticCheckResult(
passed=result.passed,
message=result.message,
data=result.data,
severity=configured,
)
self._cache.put(rule.name, file_hash, result)
results.append(result)
return results
def as_pipeline_results(
self,
source: str,
filename: str,
rule_names: list[str] | None = None,
) -> list[PipelineResultDict]:
"""Run checks and return pipeline-compatible result dicts.
Each dict has ``passed``, ``message``, ``data``, ``severity``,
and ``mode`` keys for integration with ``ValidationPipeline``.
"""
checks = self.check_file(source, filename, rule_names)
pipeline_results: list[PipelineResultDict] = []
for result in checks:
mode = map_severity_to_mode(result.severity)
pipeline_results.append(
PipelineResultDict(
passed=result.passed,
message=result.message,
data=result.data,
severity=result.severity.value,
mode=mode.value,
)
)
return pipeline_results
def normalise_output(self, result: SemanticCheckResult) -> NormalisedOutputDict:
"""Normalise a check result to the standard output schema.
Returns a dict with ``passed``, ``message``, and ``data`` keys.
"""
return NormalisedOutputDict(
passed=result.passed,
message=result.message,
data=result.data,
)
vulture_whitelist.py
+43
View File
@@ -351,3 +351,46 @@ _decision_seq # noqa: B018, F821
mock_providers # noqa: B018, F821
validate_provider_availability # noqa: B018, F821
resolve_provider_by_name # noqa: B018, F821
# Semantic validation service — public API (M6.semantic)
SemanticValidationSeverity # noqa: B018, F821
SemanticCheckResult # noqa: B018, F821
SemanticValidationRule # noqa: B018, F821
SemanticRuleRegistry # noqa: B018, F821
SemanticValidationCache # noqa: B018, F821
SemanticValidationService # noqa: B018, F821
SyntaxCheckRule # noqa: B018, F821
MissingImportRule # noqa: B018, F821
BrokenReferenceRule # noqa: B018, F821
DependencyCycleRule # noqa: B018, F821
DuplicateImportRule # noqa: B018, F821
APIMisuseRule # noqa: B018, F821
MissingSymbolRule # noqa: B018, F821
create_default_registry # noqa: B018, F821
map_severity_to_mode # noqa: B018, F821
resolve_severity # noqa: B018, F821
CONFIG_KEY_ENABLED # noqa: B018, F821
CONFIG_KEY_PYTHON_ENABLED # noqa: B018, F821
CONFIG_KEY_SEVERITY_MAPPING # noqa: B018, F821
DEFAULT_CONFIG # noqa: B018, F821
normalise_output # noqa: B018, F821
as_pipeline_results # noqa: B018, F821
check_file # noqa: B018, F821
python_enabled # noqa: B018, F821
list_rules # noqa: B018, F821
all_rules # noqa: B018, F821
compute_hash # noqa: B018, F821
invalidate # noqa: B018, F821
_BUILTINS_SET # noqa: B018, F821
_STDLIB_MODULE_NAMES # noqa: B018, F821
_DANGEROUS_NAMES # noqa: B018, F821
_DANGEROUS_NAME_DESCRIPTIONS # noqa: B018, F821
_DANGEROUS_ATTRS # noqa: B018, F821
_PYTHON_EXTENSIONS # noqa: B018, F821
_DEFAULT_CACHE_MAX_SIZE # noqa: B018, F821
_collect_binding_names # noqa: B018, F821
_collect_target_names # noqa: B018, F821
_collect_all_scope_names # noqa: B018, F821
_collect_function_local_names # noqa: B018, F821
_iter_functions # noqa: B018, F821
_is_python_file # noqa: B018, F821