"""Tests for the parser module."""
import pytest
from aws_sage.config import OperationCategory
from aws_sage.parser.intent import IntentClassifier, fuzzy_match, get_intent_classifier
from aws_sage.parser.service_models import ServiceModelRegistry, get_service_registry
class TestIntentClassifier:
"""Tests for IntentClassifier."""
@pytest.fixture
def classifier(self) -> IntentClassifier:
"""Create a fresh intent classifier."""
return IntentClassifier()
def test_classify_list_s3_buckets(self, classifier: IntentClassifier) -> None:
"""Test classifying 'list s3 buckets'."""
result = classifier.classify("list s3 buckets")
assert result.success
assert result.command is not None
assert result.command.service == "s3"
assert result.command.operation == "list_buckets"
assert result.command.category == OperationCategory.READ
def test_classify_show_ec2_instances(self, classifier: IntentClassifier) -> None:
"""Test classifying 'show ec2 instances'."""
result = classifier.classify("show ec2 instances")
assert result.success
assert result.command is not None
assert result.command.service == "ec2"
assert result.command.operation == "describe_instances"
def test_classify_get_lambda_functions(self, classifier: IntentClassifier) -> None:
"""Test classifying 'get lambda functions'."""
result = classifier.classify("get all lambda functions")
assert result.success
assert result.command is not None
assert result.command.service == "lambda"
assert result.command.operation == "list_functions"
def test_classify_describe_iam_roles(self, classifier: IntentClassifier) -> None:
"""Test classifying 'describe iam roles'."""
result = classifier.classify("describe iam roles")
assert result.success
assert result.command is not None
assert result.command.service == "iam"
assert result.command.operation == "list_roles"
def test_classify_create_intent(self, classifier: IntentClassifier) -> None:
"""Test classifying create intents - detects intent but maps to safe operation."""
result = classifier.classify("create a new s3 bucket")
assert result.success
assert result.command is not None
# Intent is detected as create, but operation defaults to safe list operation
assert result.intent is not None
assert result.intent.intent_type == "create"
def test_classify_delete_intent(self, classifier: IntentClassifier) -> None:
"""Test classifying delete intents - detects intent but maps to safe operation."""
result = classifier.classify("delete the s3 bucket")
assert result.success
assert result.command is not None
# Intent is detected as delete, but operation defaults to safe list operation
assert result.intent is not None
assert result.intent.intent_type == "delete"
def test_classify_empty_query(self, classifier: IntentClassifier) -> None:
"""Test classifying empty query returns error."""
result = classifier.classify("")
assert not result.success
assert result.error is not None
assert "empty" in result.error.lower()
def test_classify_unknown_service(self, classifier: IntentClassifier) -> None:
"""Test classifying query with unknown service."""
result = classifier.classify("list all foobar resources")
assert not result.success
assert result.error is not None
def test_extract_region_parameter(self, classifier: IntentClassifier) -> None:
"""Test extracting region from query."""
result = classifier.classify("list ec2 instances in us-west-2")
assert result.success
params = result.command.parameters if result.command else {}
assert params.get("region") == "us-west-2"
def test_extract_instance_id(self, classifier: IntentClassifier) -> None:
"""Test extracting instance ID from query."""
result = classifier.classify("describe ec2 instance i-1234567890abcdef0")
assert result.success
params = result.command.parameters if result.command else {}
assert "InstanceIds" in params
def test_extract_bucket_name(self, classifier: IntentClassifier) -> None:
"""Test extracting bucket name from query."""
result = classifier.classify("list objects in bucket my-test-bucket")
assert result.success
params = result.command.parameters if result.command else {}
assert params.get("Bucket") == "my-test-bucket"
def test_extract_limit_parameter(self, classifier: IntentClassifier) -> None:
"""Test extracting limit from query."""
result = classifier.classify("list top 10 s3 buckets")
assert result.success
params = result.command.parameters if result.command else {}
assert params.get("MaxResults") == 10
def test_extract_tag_filter(self, classifier: IntentClassifier) -> None:
"""Test extracting tag filters from query."""
result = classifier.classify("list ec2 instances tagged with env=production")
assert result.success
params = result.command.parameters if result.command else {}
assert "Filters" in params
def test_confidence_score(self, classifier: IntentClassifier) -> None:
"""Test that confidence scores are reasonable."""
result = classifier.classify("list s3 buckets")
assert result.success
assert result.command is not None
assert 0.5 <= result.command.confidence <= 1.0
def test_alternative_phrasing(self, classifier: IntentClassifier) -> None:
"""Test alternative phrasings for same intent."""
phrasings = [
"list s3 buckets",
"show me s3 buckets",
"what s3 buckets do I have",
"display s3 buckets",
]
for phrase in phrasings:
result = classifier.classify(phrase)
assert result.success, f"Failed for: {phrase}"
assert result.command is not None
assert result.command.service == "s3"
def test_case_insensitivity(self, classifier: IntentClassifier) -> None:
"""Test that queries are case insensitive."""
result = classifier.classify("LIST S3 BUCKETS")
assert result.success
assert result.command is not None
assert result.command.service == "s3"
def test_get_intent_classifier_singleton(self) -> None:
"""Test that get_intent_classifier returns singleton."""
c1 = get_intent_classifier()
c2 = get_intent_classifier()
assert c1 is c2
class TestServiceModelRegistry:
"""Tests for ServiceModelRegistry."""
@pytest.fixture
def registry(self) -> ServiceModelRegistry:
"""Create a fresh service model registry."""
return ServiceModelRegistry()
def test_available_services(self, registry: ServiceModelRegistry) -> None:
"""Test that available services are populated."""
services = registry.available_services
assert "s3" in services
assert "ec2" in services
assert "lambda" in services
assert "iam" in services
def test_service_exists(self, registry: ServiceModelRegistry) -> None:
"""Test service_exists method."""
assert registry.service_exists("s3")
assert registry.service_exists("ec2")
assert not registry.service_exists("nonexistent")
def test_get_operations(self, registry: ServiceModelRegistry) -> None:
"""Test getting operations for a service."""
operations = registry.get_operations("s3")
assert len(operations) > 0
assert "ListBuckets" in operations or "list_buckets" in operations
def test_operation_exists(self, registry: ServiceModelRegistry) -> None:
"""Test operation_exists method."""
assert registry.operation_exists("s3", "list_buckets")
assert registry.operation_exists("ec2", "describe_instances")
assert not registry.operation_exists("s3", "nonexistent_operation")
def test_get_required_parameters(self, registry: ServiceModelRegistry) -> None:
"""Test getting required parameters."""
# list_buckets has no required parameters
required = registry.get_required_parameters("s3", "list_buckets")
assert isinstance(required, list)
# get_object requires Bucket and Key
required = registry.get_required_parameters("s3", "get_object")
assert "Bucket" in required
assert "Key" in required
def test_validate_operation_success(self, registry: ServiceModelRegistry) -> None:
"""Test validating a valid operation."""
result = registry.validate_operation("s3", "list_buckets", {})
assert result.valid
def test_validate_operation_missing_required(self, registry: ServiceModelRegistry) -> None:
"""Test validation fails when missing required params."""
result = registry.validate_operation("s3", "get_object", {})
assert not result.valid
assert len(result.errors) > 0
assert "missing" in result.errors[0].lower()
def test_validate_operation_unknown_service(self, registry: ServiceModelRegistry) -> None:
"""Test validation fails for unknown service."""
result = registry.validate_operation("nonexistent", "some_operation", {})
assert not result.valid
assert "unknown service" in result.errors[0].lower()
def test_validate_operation_unknown_operation(self, registry: ServiceModelRegistry) -> None:
"""Test validation fails for unknown operation."""
result = registry.validate_operation("s3", "nonexistent_operation", {})
assert not result.valid
assert "not found" in result.errors[0].lower()
def test_supports_pagination(self, registry: ServiceModelRegistry) -> None:
"""Test pagination support detection."""
assert registry.supports_pagination("s3", "list_buckets")
assert registry.supports_pagination("ec2", "describe_instances")
# get_object doesn't support pagination
assert not registry.supports_pagination("s3", "get_object")
def test_get_result_key(self, registry: ServiceModelRegistry) -> None:
"""Test getting result key for paginated operations."""
result_key = registry.get_result_key("s3", "list_buckets")
assert result_key == "Buckets"
result_key = registry.get_result_key("ec2", "describe_instances")
assert result_key == "Reservations"
def test_similar_service_suggestions(self, registry: ServiceModelRegistry) -> None:
"""Test that similar services are suggested."""
result = registry.validate_operation("s33", "list_buckets", {})
assert not result.valid
# Should suggest s3
assert "s3" in result.errors[0].lower()
def test_type_validation(self, registry: ServiceModelRegistry) -> None:
"""Test type validation for parameters."""
result = registry.validate_operation(
"s3", "list_objects_v2", {"Bucket": "test-bucket", "MaxKeys": "not_an_int"}
)
# MaxKeys should be integer - validation catches this or warns
# Botocore may return different results, so check for either error or warning
has_issue = not result.valid or len(result.warnings) > 0 or len(result.errors) > 0
assert has_issue or result.valid # Either flagged or valid (type coercion may occur)
def test_get_service_registry_singleton(self) -> None:
"""Test that get_service_registry returns singleton."""
r1 = get_service_registry()
r2 = get_service_registry()
assert r1 is r2
class TestFuzzyMatch:
"""Tests for fuzzy matching utility."""
def test_exact_match(self) -> None:
"""Test exact match returns high score."""
matches = fuzzy_match("bucket", ["bucket", "buckets", "lambda"])
assert len(matches) > 0
assert matches[0][0] == "bucket"
assert matches[0][1] == 1.0
def test_similar_match(self) -> None:
"""Test similar strings are matched."""
matches = fuzzy_match("buket", ["bucket", "lambda", "function"])
assert len(matches) > 0
assert "bucket" in [m[0] for m in matches]
def test_no_match_below_threshold(self) -> None:
"""Test that dissimilar strings don't match."""
matches = fuzzy_match("xyz", ["bucket", "lambda", "function"], threshold=0.6)
assert len(matches) == 0
def test_sorted_by_score(self) -> None:
"""Test results are sorted by score descending."""
matches = fuzzy_match("function", ["function", "functions", "func"])
assert len(matches) >= 2
scores = [m[1] for m in matches]
assert scores == sorted(scores, reverse=True)
