diff --git a/CHANGELOG.md b/CHANGELOG.md index dca4e6b0f..1c4f2550c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/benchmarks/semantic_validation_bench.py b/benchmarks/semantic_validation_bench.py new file mode 100644 index 000000000..2a318ac68 --- /dev/null +++ b/benchmarks/semantic_validation_bench.py @@ -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") diff --git a/docs/reference/semantic_validation.md b/docs/reference/semantic_validation.md new file mode 100644 index 000000000..68d34abef --- /dev/null +++ b/docs/reference/semantic_validation.md @@ -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 | diff --git a/features/semantic_validation.feature b/features/semantic_validation.feature new file mode 100644 index 000000000..a239ef1c8 --- /dev/null +++ b/features/semantic_validation.feature @@ -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 diff --git a/features/steps/semantic_validation_steps.py b/features/steps/semantic_validation_steps.py new file mode 100644 index 000000000..1921a72b4 --- /dev/null +++ b/features/steps/semantic_validation_steps.py @@ -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 diff --git a/robot/helper_semantic_validation.py b/robot/helper_semantic_validation.py new file mode 100644 index 000000000..54ff8539d --- /dev/null +++ b/robot/helper_semantic_validation.py @@ -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 ", 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) diff --git a/robot/semantic_validation.robot b/robot/semantic_validation.robot new file mode 100644 index 000000000..4760af8d7 --- /dev/null +++ b/robot/semantic_validation.robot @@ -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 diff --git a/src/cleveragents/application/services/__init__.py b/src/cleveragents/application/services/__init__.py index 011d5debd..76ae0db62 100644 --- a/src/cleveragents/application/services/__init__.py +++ b/src/cleveragents/application/services/__init__.py @@ -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", ] diff --git a/src/cleveragents/application/services/semantic_validation_rules.py b/src/cleveragents/application/services/semantic_validation_rules.py new file mode 100644 index 000000000..19524468b --- /dev/null +++ b/src/cleveragents/application/services/semantic_validation_rules.py @@ -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, + ) diff --git a/src/cleveragents/application/services/semantic_validation_service.py b/src/cleveragents/application/services/semantic_validation_service.py new file mode 100644 index 000000000..d9aa10f3e --- /dev/null +++ b/src/cleveragents/application/services/semantic_validation_service.py @@ -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, + ) diff --git a/vulture_whitelist.py b/vulture_whitelist.py index 88c6c4377..a9259e8c0 100644 --- a/vulture_whitelist.py +++ b/vulture_whitelist.py @@ -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