Khi xây dựng production system với AI models, việc nhận được response chỉ là bước đầu tiên. Điều thực sự quan trọng là đảm bảo output từ AI luôn an toàn, đúng format, và không chứa malicious content. Trong bài viết này, tôi sẽ chia sẻ kinh nghiệm thực chiến 3 năm xây dựng validation layer cho các hệ thống AI tại production, cùng với code examples có thể copy-paste ngay.
Bảng So Sánh: HolySheep vs Official API vs Relay Services
| Tiêu chí | HolySheep AI | Official OpenAI/Anthropic | Proxy/Relay Services |
|---|---|---|---|
| Chi phí GPT-4.1 | $8/1M tokens | $60/1M tokens | $15-30/1M tokens |
| Chi phí Claude Sonnet 4.5 | $15/1M tokens | $3/1M tokens | $8-12/1M tokens |
| Chi phí Gemini 2.5 Flash | $2.50/1M tokens | $0.125/1M tokens | $1.5-3/1M tokens |
| DeepSeek V3.2 | $0.42/1M tokens | Không có | $0.80/1M tokens |
| Độ trễ trung bình | <50ms | 100-300ms | 150-500ms |
| Thanh toán | WeChat/Alipay, Visa | Chỉ Visa quốc tế | Khác nhau |
| Tín dụng miễn phí | Có khi đăng ký | $5 trial | Ít khi có |
| Built-in Validation | Basic checks | Không | Không |
Tỷ giá ¥1 = $1 của HolySheep giúp tiết kiệm 85%+ so với các giải pháp khác. Với độ trễ dưới 50ms, đây là lựa chọn tối ưu cho các production systems cần real-time validation.
Tại Sao Response Validation Lại Quan Trọng?
Trong thực tế production, tôi đã gặp nhiều trường hợp AI trả về:
- JSON malformed - thiếu dấu phẩy, ngoặc không đóng đúng
- Injection attempts - user cố tình inject prompt qua input
- Content policy violations - text vi phạm guidelines
- Unexpected format - model thay đổi output pattern
- Empty responses - model không generate được gì
- Excessive length - output quá dài, gây memory issues
Không có validation, những response này sẽ crash production system hoặc gây security vulnerabilities nghiêm trọng.
Kiến Trúc Validation Layer Tổng Quát
# response_validator.py
Architecture tổng quát cho AI response validation
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any, Optional, List, Dict
from enum import Enum
import json
import re
from html import escape
class ValidationSeverity(Enum):
ERROR = "error"
WARNING = "warning"
INFO = "info"
@dataclass
class ValidationResult:
is_valid: bool
severity: ValidationSeverity
message: str
sanitized_output: Optional[Any] = None
metadata: Optional[Dict] = None
class BaseValidator(ABC):
"""Abstract base class cho tất cả validators"""
@abstractmethod
def validate(self, response: str, **kwargs) -> ValidationResult:
pass
@abstractmethod
def sanitize(self, response: str) -> str:
"""Sanitize input/output"""
pass
class ResponseValidatorPipeline:
"""
Pipeline orchestrator - chạy validators theo thứ tự
Tối ưu cho HolySheep API responses với <50ms latency
"""
def __init__(self, validators: List[BaseValidator]):
self.validators = validators
self._cache = {} # Optional: cache validation results
def validate(self, response: str) -> ValidationResult:
current_output = response
for validator in self.validators:
result = validator.validate(current_output)
if not result.is_valid:
return result
if result.sanitized_output:
current_output = result.sanitized_output
return ValidationResult(
is_valid=True,
severity=ValidationSeverity.INFO,
message="All validations passed",
sanitized_output=current_output
)
Ví dụ usage với HolySheep API
def validate_ai_response(response_text: str) -> ValidationResult:
pipeline = ResponseValidatorPipeline([
JSONStructureValidator(),
ContentSafetyValidator(),
LengthValidator(max_chars=10000),
XSSValidator(),
])
return pipeline.validate(response_text)Validator 1: JSON Structure Validation
Đây là validator phổ biến nhất mà tôi sử dụng. AI models thường generate JSON với syntax errors nhỏ mà con người khó phát hiện.
# json_validator.py
import json
import re
from typing import Optional, Dict, Any, List, Tuple
class JSONStructureValidator(BaseValidator):
"""
Validator cho JSON responses - xử lý cả JSON thuần và streaming
Hỗ trợ schema validation với JSON Schema draft-07
"""
def __init__(
self,
required_fields: Optional[List[str]] = None,
field_types: Optional[Dict[str, type]] = None,
allow_extra_fields: bool = False,
strict_mode: bool = True
):
self.required_fields = required_fields or []
self.field_types = field_types or {}
self.allow_extra_fields = allow_extra_fields
self.strict_mode = strict_mode
def validate(self, response: str, **kwargs) -> ValidationResult:
# Bước 1: Parse JSON
parsed, error = self._parse_json(response)
if error:
# Thử fix common JSON errors
fixed, was_fixed = self._attempt_fix(response)
if was_fixed:
return ValidationResult(
is_valid=True,
severity=ValidationSeverity.WARNING,
message=f"Auto-fixed JSON: {error}",
sanitized_output=fixed,
metadata={"original_error": error, "was_auto_fixed": True}
)
return ValidationResult(
is_valid=False,
severity=ValidationSeverity.ERROR,
message=f"JSON parse error: {error}",
metadata={"original_response": response[:500]}
)
# Bước 2: Validate required fields
missing_fields = self._check_required_fields(parsed)
if missing_fields:
return ValidationResult(
is_valid=False,
severity=ValidationSeverity.ERROR,
message=f"Missing required fields: {missing_fields}",
sanitized_output=response
)
# Bước 3: Validate field types
type_errors = self._check_field_types(parsed)
if type_errors:
return ValidationResult(
is_valid=False,
severity=ValidationSeverity.ERROR,
message=f"Type mismatches: {type_errors}",
sanitized_output=response
)
return ValidationResult(
is_valid=True,
severity=ValidationSeverity.INFO,
message="JSON structure valid",
sanitized_output=json.dumps(parsed, ensure_ascii=False)
)
def _parse_json(self, text: str) -> Tuple[Optional[Dict], Optional[str]]:
"""Parse JSON với error handling chi tiết"""
text = text.strip()
# Thử direct parse trước
try:
return json.loads(text), None
except json.JSONDecodeError as e:
return None, str(e)
def _attempt_fix(self, text: str) -> Tuple[str, bool]:
"""Thử auto-fix common JSON errors"""
fixed = text
# Fix 1: Trailing commas
fixed = re.sub(r',(\s*[}\]])', r'\1', fixed)
# Fix 2: Single quotes to double quotes (simple cases)
# Chỉ fix keys và simple strings
fixed = re.sub(r"'([^']*)':", r'"\1":', fixed)
fixed = re.sub(r":\s*'([^']*)'(,?\s*[}\]])", r': "\1"\2', fixed)
# Fix 3: Comments
fixed = re.sub(r'//.*?$', '', fixed, flags=re.MULTILINE)
fixed = re.sub(r'/\*.*?\*/', '', fixed, flags=re.DOTALL)
# Re-parse attempt
try:
json.loads(fixed)
return fixed, True
except:
return text, False
def _check_required_fields(self, data: Dict) -> List[str]:
missing = []
for field in self.required_fields:
if field not in data:
missing.append(field)
return missing
def _check_field_types(self, data: Dict) -> Dict[str, str]:
errors = {}
for field, expected_type in self.field_types.items():
if field in data and not isinstance(data[field], expected_type):
errors[field] = f"Expected {expected_type.__name__}, got {type(data[field]).__name__}"
return errors
def sanitize(self, response: str) -> str:
"""Sanitize JSON response"""
parsed, _ = self._parse_json(response)
if parsed:
return json.dumps(parsed, ensure_ascii=False, indent=2)
return response
Usage example với HolySheep API response
def process_holysheep_json_response(response_text: str) -> Dict[str, Any]:
validator = JSONStructureValidator(
required_fields=["id", "model", "choices"],
field_types={
"id": str,
"model": str,
"choices": list
}
)
result = validator.validate(response_text)
if not result.is_valid:
raise ValueError(f"Response validation failed: {result.message}")
return json.loads(result.sanitized_output)Validator 2: Content Safety và XSS Protection
Security là ưu tiên hàng đầu. AI có thể vô tình generate content chứa XSS payloads hoặc sensitive data.
# security_validator.py
import re
from html import escape, unescape
from typing import Set, List, Pattern
import hashlib
class SecurityValidator(BaseValidator):
"""
Comprehensive security validator cho AI responses
- XSS protection
- SQL injection prevention
- Sensitive data redaction
- Prompt injection detection
"""
# Common XSS patterns
XSS_PATTERNS: List[Pattern] = [
re.compile(r']*>.*?', re.IGNORECASE | re.DOTALL),
re.compile(r'javascript:', re.IGNORECASE),
re.compile(r'on\w+\s*=', re.IGNORECASE), # onerror=, onclick=, etc.
re.compile(r']*>.*?', re.IGNORECASE | re.DOTALL),
re.compile(r']*>', re.IGNORECASE),
re.compile(r']*>.*?', re.IGNORECASE | re.DOTALL),
re.compile(r'eval\s*\(', re.IGNORECASE),
re.compile(r'expression\s*\(', re.IGNORECASE),
]
# SQL injection patterns
SQL_INJECTION_PATTERNS: List[Pattern] = [
re.compile(r"(\bOR\b|\bAND\b).*=.*", re.IGNORECASE),
re.compile(r"(\bUNION\b|\bSELECT\b|\bINSERT\b|\bUPDATE\b|\bDELETE\b)", re.IGNORECASE),
re.compile(r"(--|#|/\*|\*/)", re.IGNORECASE),
re.compile(r";\s*(DROP|TRUNCATE|ALTER)", re.IGNORECASE),
]
# Prompt injection markers
INJECTION_MARKERS: List[Pattern] = [
re.compile(r'\[INST\]\s*$', re.IGNORECASE), # Llama jailbreak
re.compile(r'\{\{.*\}\}', re.IGNORECASE), # Template injection
re.compile(r'<\|.*\|>', re.IGNORECASE), # Special tokens
re.compile(r' Ignore.*instructions', re.IGNORECASE),
]
def __init__(
self,
redact_sensitive: bool = True,
sensitive_patterns: List[Pattern] = None,
allow_html: bool = False,
max_depth: int = 10
):
self.redact_sensitive = redact_sensitive
self.sensitive_patterns = sensitive_patterns or self._default_sensitive_patterns()
self.allow_html = allow_html
self.max_depth = max_depth
def _default_sensitive_patterns(self) -> List[Pattern]:
return [
re.compile(r'\b\d{3}-\d{2}-\d{4}\b'), # SSN
re.compile(r'\b\d{16}\b'), # Credit card
re.compile(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'), # Email
re.compile(r'api[_-]?key["\']?\s*[:=]\s*["\']?[A-Za-z0-9_-]{20,}', re.IGNORECASE),
re.compile(r'sk-[A-Za-z0-9]{20,}', re.IGNORECASE), # OpenAI keys
re.compile(r'Bearer\s+[A-Za-z0-9_-]+', re.IGNORECASE),
]
def validate(self, response: str, **kwargs) -> ValidationResult:
violations = []
sanitized = response
# Check XSS
xss_found = self._check_xss(response)
if xss_found:
if not self.allow_html:
violations.append(f"XSS patterns detected: {xss_found}")
sanitized = self._sanitize_xss(sanitized)
else:
violations.append(f"Potential XSS (allowed): {xss_found}")
# Check SQL injection
sql_found = self._check_sql_injection(response)
if sql_found:
violations.append(f"SQL injection patterns: {sql_found}")
sanitized = self._redact_sql_injection(sanitized)
# Check prompt injection
injection_found = self._check_prompt_injection(response)
if injection_found:
violations.append(f"Potential prompt injection: {injection_found}")
sanitized = self._neutralize_injection(sanitized)
# Redact sensitive data
if self.redact_sensitive:
sanitized, redacted = self._redact_sensitive_data(sanitized)
if redacted:
violations.append(f"Redacted {len(redacted)} sensitive data instances")
if violations:
return ValidationResult(
is_valid=True, # Still valid, just sanitized
severity=ValidationSeverity.WARNING,
message="; ".join(violations),
sanitized_output=sanitized,
metadata={"violations": violations, "was_sanitized": True}
)
return ValidationResult(
is_valid=True,
severity=ValidationSeverity.INFO,
message="Security validation passed",
sanitized_output=sanitized
)
def _check_xss(self, text: str) -> List[str]:
found = []
for pattern in self.XSS_PATTERNS:
match = pattern.search(text)
if match:
found.append(match.group()[:50])
return found
def _sanitize_xss(self, text: str) -> str:
"""Sanitize XSS by escaping HTML entities"""
return escape(text)
def _check_sql_injection(self, text: str) -> List[str]:
found = []
for pattern in self.SQL_INJECTION_PATTERNS:
match = pattern.search(text)
if match:
found.append(match.group()[:50])
return found
def _redact_sql_injection(self, text: str) -> str:
"""Replace potentially dangerous SQL with placeholders"""
for pattern in self.SQL_INJECTION_PATTERNS:
text = pattern.sub('[REDACTED_SQL]', text)
return text
def _check_prompt_injection(self, text: str) -> List[str]:
found = []
for pattern in self.INJECTION_MARKERS:
match = pattern.search(text)
if match:
found.append(match.group()[:50])
return found
def _neutralize_injection(self, text: str) -> str:
"""Remove or neutralize prompt injection attempts"""
for pattern in self.INJECTION_MARKERS:
text = pattern.sub('[INJECTION_BLOCKED]', text)
return text
def _redact_sensitive_data(self, text: str) -> tuple:
redacted_items = []
for pattern in self.sensitive_patterns:
matches = pattern.findall(text)
redacted_items.extend(matches)
text = pattern.sub('[REDACTED]', text)
return text, redacted_items
def sanitize(self, response: str) -> str:
"""Main sanitization method"""
_, result = self._parse_and_sanitize(response)
return result
def _parse_and_sanitize(self, text: str, depth: int = 0) -> tuple:
"""Recursively sanitize text, handling nested structures"""
if depth > self.max_depth:
return False, text
# Check and sanitize
sanitized = escape(text) if not self.allow_html else text
is_safe = not any(p.search(sanitized) for p in self.XSS_PATTERNS)
return is_safe, sanitized
Production usage với HolySheep API
def safe_ai_response(response: str, allow_html: bool = False) -> str:
validator = SecurityValidator(
redact_sensitive=True,
allow_html=allow_html
)
result = validator.validate(response)
return result.sanitized_output Validator 3: Length và Format Constraints
# length_format_validator.py
import re
from typing import Optional, Tuple, List
class LengthFormatValidator(BaseValidator):
"""
Validator cho response length và format constraints
Essential cho prevent token overflow và quota issues
"""
def __init__(
self,
min_chars: int = 0,
max_chars: int = 100000,
min_tokens: Optional[int] = None,
max_tokens: Optional[int] = 4096,
expected_formats: Optional[List[str]] = None,
line_count_range: Tuple[int, int] = (0, 10000),
word_count_range: Tuple[int, int] = (0, 50000)
):
self.min_chars = min_chars
self.max_chars = max_chars
self.min_tokens = min_tokens
self.max_tokens = max_tokens
self.expected_formats = expected_formats or []
self.line_count_range = line_count_range
self.word_count_range = word_count_range
def estimate_tokens(self, text: str) -> int:
"""Estimate token count -rough but fast (4 chars ~ 1 token)"""
return len(text) // 4
def validate(self, response: str, **kwargs) -> ValidationResult:
issues = []
char_count = len(response)
# Character length check
if char_count < self.min_chars:
issues.append(f"Response too short: {char_count} < {self.min_chars}")
if char_count > self.max_chars:
issues.append(f"Response too long: {char_count} > {self.max_chars}")
# Token estimation
estimated_tokens = self.estimate_tokens(response)
if self.min_tokens and estimated_tokens < self.min_tokens:
issues.append(f"Token estimate too low: {estimated_tokens} < {self.min_tokens}")
if self.max_tokens and estimated_tokens > self.max_tokens:
issues.append(f"Token estimate too high: {estimated_tokens} > {self.max_tokens}")
# Line count
line_count = len(response.splitlines())
if not (self.line_count_range[0] <= line_count <= self.line_count_range[1]):
issues.append(f"Line count out of range: {line_count}")
# Word count
word_count = len(response.split())
if not (self.word_count_range[0] <= word_count <= self.word_count_range[1]):
issues.append(f"Word count out of range: {word_count}")
# Format validation
if self.expected_formats:
format_valid = self._check_format(response)
if not format_valid:
issues.append(f"Format mismatch. Expected one of: {self.expected_formats}")
if issues:
return ValidationResult(
is_valid=False,
severity=ValidationSeverity.ERROR,
message="; ".join(issues),
sanitized_output=response,
metadata={
"char_count": char_count,
"estimated_tokens": estimated_tokens,
"line_count": line_count,
"word_count": word_count
}
)
return ValidationResult(
is_valid=True,
severity=ValidationSeverity.INFO,
message="Length and format validation passed",
sanitized_output=response,
metadata={
"char_count": char_count,
"estimated_tokens": estimated_tokens
}
)
def _check_format(self, text: str) -> bool:
"""Check if text matches any expected format"""
# JSON format
if "json" in self.expected_formats:
try:
import json
json.loads(text)
return True
except:
pass
# Markdown format
if "markdown" in self.expected_formats:
if text.startswith("#") or "```" in text or "**" in text:
return True
# Plain text
if "text" in self.expected_formats:
return True
return False
def sanitize(self, response: str) -> str:
"""Truncate if too long"""
if len(response) > self.max_chars:
return response[:self.max_chars] + "\n... [TRUNCATED]"
return response
Token calculation helper - quan trọng cho HolySheep billing
def calculate_cost(response: str, model: str) -> float:
"""
Calculate approximate cost dựa trên token count
HolySheep 2026 pricing
"""
tokens = len(response) // 4 # Rough estimate
pricing = {
"gpt-4.1": 8.0, # $8/1M tokens
"claude-sonnet-4.5": 15.0, # $15/1M tokens
"gemini-2.5-flash": 2.50, # $2.50/1M tokens
"deepseek-v3.2": 0.42, # $0.42/1M tokens
}
rate = pricing.get(model, 10.0) # Default fallback
return (tokens / 1_000_000) * rateTích Hợp Hoàn Chỉnh với HolySheep API
# holysheep_integration.py
import requests
import time
import logging
from typing import Optional, Dict, Any
from response_validator import (
ResponseValidatorPipeline,
JSONStructureValidator,
SecurityValidator,
LengthFormatValidator
)
from content_safety import ContentModerationValidator
logger = logging.getLogger(__name__)
class HolySheepAIClient:
"""
Production-ready client cho HolySheep AI API
Tích hợp sẵn validation pipeline
"""
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1",
timeout: int = 30,
max_retries: int = 3,
enable_validation: bool = True
):
self.api_key = api_key
self.base_url = base_url.rstrip('/')
self.timeout = timeout
self.max_retries = max_retries
self.enable_validation = enable_validation
# Initialize validation pipeline
self.validator = ResponseValidatorPipeline([
JSONStructureValidator(
required_fields=["id", "model", "choices"],
field_types={"id": str, "model": str}
),
SecurityValidator(
redact_sensitive=True,
allow_html=False
),
LengthFormatValidator(
max_chars=50000,
max_tokens=4000,
expected_formats=["json", "text"]
),
ContentModerationValidator()
])
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
})
# Metrics
self.metrics = {
"total_requests": 0,
"successful_requests": 0,
"failed_requests": 0,
"validation_failures": 0,
"avg_latency_ms": 0,
"total_cost_usd": 0.0
}
def chat_completions(
self,
messages: list,
model: str = "gpt-4.1",
temperature: float = 0.7,
max_tokens: int = 1000,
stream: bool = False,
**kwargs
) -> Dict[str, Any]:
"""
Gửi chat completion request tới HolySheep API
"""
self.metrics["total_requests"] += 1
start_time = time.time()
endpoint = f"{self.base_url}/chat/completions"
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
"stream": stream,
**kwargs
}
for attempt in range(self.max_retries):
try:
response = self.session.post(
endpoint,
json=payload,
timeout=self.timeout
)
response.raise_for_status()
result = response.json()
latency_ms = (time.time() - start_time) * 1000
# Validate response
if self.enable_validation:
validation_result = self._validate_response(result)
if not validation_result.is_valid:
logger.warning(f"Validation failed: {validation_result.message}")
self.metrics["validation_failures"] += 1
raise ValueError(f"Response validation failed: {validation_result.message}")
# Update metrics
self.metrics["successful_requests"] += 1
self._update_latency_metrics(latency_ms)
self._update_cost_metrics(result, model)
return result
except requests.exceptions.Timeout:
logger.warning(f"Request timeout (attempt {attempt + 1}/{self.max_retries})")
if attempt == self.max_retries - 1:
self.metrics["failed_requests"] += 1
raise
except requests.exceptions.RequestException as e:
logger.error(f"Request error: {e}")
self.metrics["failed_requests"] += 1
raise
except ValueError as e:
self.metrics["failed_requests"] += 1
raise
def _validate_response(self, response: Dict) -> ValidationResult:
"""Validate API response"""
# Extract content from response
content = self._extract_content(response)
return self.validator.validate(content)
def _extract_content(self, response: Dict) -> str:
"""Extract text content from various response formats"""
try:
if "choices" in response:
return response["choices"][0].get("message", {}).get("content", "")
elif "content" in response:
return response["content"]
return str(response)
except Exception:
return str(response)
def _update_latency_metrics(self, latency_ms: float):
"""Update rolling average latency"""
current_avg = self.metrics["avg_latency_ms"]
total = self.metrics["total_requests"]
self.metrics["avg_latency_ms"] = (
(current_avg * (total - 1) + latency_ms) / total
)
def _update_cost_metrics(self, response: Dict, model: str):
"""Calculate and update cost metrics"""
try:
usage = response.get("usage", {})
prompt_tokens = usage.get("prompt_tokens", 0)
completion_tokens = usage.get("completion_tokens", 0)
total_tokens = usage.get("total_tokens", prompt_tokens + completion_tokens)
pricing = {
"gpt-4.1": 8.0,
"claude-sonnet-4.5": 15.0,
"gemini-2.5-flash": 2.50,
"deepseek-v3.2": 0.42
}
rate = pricing.get(model, 10.0)
cost = (total_tokens / 1_000_000) * rate
self.metrics["total_cost_usd"] += cost
except Exception as e:
logger.warning(f"Could not calculate cost: {e}")
def get_metrics(self) -> Dict[str, Any]:
"""Return current metrics"""
return self.metrics.copy()
Usage example
def main():
# Initialize client
client = HolySheepAIClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
enable_validation=True
)
# Send request
messages = [
{"role": "system", "content": "You are a helpful assistant. Always respond with valid JSON."},
{"role": "user", "content": "Get me information about Python libraries"}
]
try:
response = client.chat_completions(
messages=messages,
model="gpt-4.1",
temperature=0.7,
max_tokens=500
)
print(f"Response: {response['choices'][0]['message']['content']}")
print(f"Latency: {client.metrics['avg_latency_ms']:.2f}ms")
print(f"Total cost: ${client.metrics['total_cost_usd']:.4f}")
except Exception as e:
print(f"Error: {e}")
if __name__ == "__main__":
main()Kiểm Thử Validators
# test_validators.py
import pytest
import json
from response_validator import (
JSONStructureValidator,
SecurityValidator,
LengthFormatValidator,
ResponseValidatorPipeline,
ValidationResult,
ValidationSeverity
)
class TestJSONValidator:
"""Test cases cho JSON validator"""
def test_valid_json(self):
validator = JSONStructureValidator()
test_data = '{"name": "test", "value": 123}'
result = validator.validate(test_data)
assert result.is_valid == True
def test_missing_required_fields(self):
validator = JSONStructureValidator(required_fields=["id", "name"])
test_data = '{"name": "test"}'
result = validator.validate(test_data)
assert result.is_valid == False
assert "Missing required fields" in result.message
def test_auto_fix_trailing_comma(self):
validator = JSONStructureValidator()
test_data = '{"name": "test", "value": 123,}'
result = validator.validate(test_data)
assert result.is_valid == True
assert result.metadata.get("was_auto_fixed") == True
def test_type_mismatch(self):
validator = JSONStructureValidator(
field_types={"count": int}
)
test_data = '{"count": "not_a_number"}'
result = validator.validate(test_data)
assert result.is_valid == False
class TestSecurityValidator:
"""Test cases cho security validator"""
def test_xss_script_tag(self):
validator = SecurityValidator(allow_html=False)
test_data = ''
result = validator.validate(test_data)
assert result.sanitized_output != test_data
def test_sensitive_data_redaction(self):
validator = SecurityValidator(redact_sensitive=True)
test_data = 'API Key: sk-1234567890abcdefghij'
result = validator.validate(test_data)
assert 'sk-1234567890' not in result.sanitized_output
assert '[REDACTED]' in result.sanitized_output
def test_sql_injection_detection(self):
validator = SecurityValidator()
test_data = "SELECT * FROM users WHERE id=1 OR 1=1"
result = validator.validate(test_data)
assert '[