150 lines
4.4 KiB
Python
150 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Test script for Phase 1 implementation."""
|
|
import sys
|
|
import os
|
|
|
|
# Add current directory to path
|
|
sys.path.insert(0, os.path.dirname(__file__))
|
|
|
|
def test_imports():
|
|
"""Test that all modules can be imported."""
|
|
print("Testing Phase 1 imports...")
|
|
|
|
try:
|
|
from config.error_codes import ErrorCode
|
|
print("✅ config.error_codes imported")
|
|
print(f" Error codes: {[e.value for e in ErrorCode][:3]}...")
|
|
except Exception as e:
|
|
print(f"❌ config.error_codes failed: {e}")
|
|
return False
|
|
|
|
try:
|
|
from config.settings import settings
|
|
print("✅ config.settings imported")
|
|
print(f" Service token configured: {bool(settings.HEICODE_SERVICE_TOKEN)}")
|
|
except Exception as e:
|
|
print(f"❌ config.settings failed: {e}")
|
|
return False
|
|
|
|
try:
|
|
from api.agnet.models import HealthCheckResponse, HealthCheckData, ErrorResponse
|
|
print("✅ api.agnet.models imported")
|
|
except Exception as e:
|
|
print(f"❌ api.agnet.models failed: {e}")
|
|
return False
|
|
|
|
try:
|
|
from api.agnet.validators import scan_for_sensitive_fields, validate_no_sensitive_fields
|
|
print("✅ api.agnet.validators imported")
|
|
except Exception as e:
|
|
print(f"❌ api.agnet.validators failed: {e}")
|
|
return False
|
|
|
|
try:
|
|
from api.agnet.auth import verify_service_token, extract_headers
|
|
print("✅ api.agnet.auth imported")
|
|
except Exception as e:
|
|
print(f"❌ api.agnet.auth failed: {e}")
|
|
return False
|
|
|
|
try:
|
|
from api.agnet.idempotency import idempotency_cache
|
|
print("✅ api.agnet.idempotency imported")
|
|
print(f" Redis available: {idempotency_cache.redis_client is not None}")
|
|
except Exception as e:
|
|
print(f"❌ api.agnet.idempotency failed: {e}")
|
|
return False
|
|
|
|
try:
|
|
from api.agnet.router import router
|
|
print("✅ api.agnet.router imported")
|
|
print(f" Router prefix: {router.prefix}")
|
|
except Exception as e:
|
|
print(f"❌ api.agnet.router failed: {e}")
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def test_validators():
|
|
"""Test sensitive field scanner."""
|
|
print("\nTesting validators...")
|
|
|
|
from api.agnet.validators import scan_for_sensitive_fields
|
|
|
|
# Test case 1: Simple sensitive field
|
|
result = scan_for_sensitive_fields({"password": "test123"})
|
|
assert result == ["password"], f"Expected ['password'], got {result}"
|
|
print("✅ Simple sensitive field detected")
|
|
|
|
# Test case 2: Nested sensitive field
|
|
result = scan_for_sensitive_fields({"user": {"token": "abc"}})
|
|
assert result == ["user.token"], f"Expected ['user.token'], got {result}"
|
|
print("✅ Nested sensitive field detected")
|
|
|
|
# Test case 3: Multiple sensitive fields
|
|
result = scan_for_sensitive_fields({
|
|
"name": "john",
|
|
"password": "test",
|
|
"config": {"api_key": "secret"}
|
|
})
|
|
assert "password" in result and "config.api_key" in result
|
|
print("✅ Multiple sensitive fields detected")
|
|
|
|
# Test case 4: No sensitive fields
|
|
result = scan_for_sensitive_fields({"name": "john", "age": 30})
|
|
assert result == [], f"Expected [], got {result}"
|
|
print("✅ Clean payload passed")
|
|
|
|
return True
|
|
|
|
|
|
def test_app_integration():
|
|
"""Test that app.py includes the new router."""
|
|
print("\nTesting app.py integration...")
|
|
|
|
try:
|
|
from app import app
|
|
|
|
# Check if agnet router is registered
|
|
agnet_routes = [route for route in app.routes if '/agnet' in str(route.path)]
|
|
|
|
if agnet_routes:
|
|
print(f"✅ Agnet router registered with {len(agnet_routes)} routes")
|
|
for route in agnet_routes[:3]:
|
|
print(f" - {route.path}")
|
|
else:
|
|
print("❌ Agnet router not found in app")
|
|
return False
|
|
|
|
return True
|
|
except Exception as e:
|
|
print(f"❌ App integration test failed: {e}")
|
|
return False
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("=" * 60)
|
|
print("Phase 1 Implementation Test")
|
|
print("=" * 60)
|
|
|
|
success = True
|
|
|
|
if not test_imports():
|
|
success = False
|
|
|
|
if not test_validators():
|
|
success = False
|
|
|
|
if not test_app_integration():
|
|
success = False
|
|
|
|
print("\n" + "=" * 60)
|
|
if success:
|
|
print("✅ All Phase 1 tests passed!")
|
|
else:
|
|
print("❌ Some tests failed")
|
|
print("=" * 60)
|
|
|
|
sys.exit(0 if success else 1)
|