Spaces:
Sleeping
Sleeping
File size: 10,045 Bytes
b7c2c9d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 | """
Unit tests for Quantum RNG implementation
"""
import pytest
import asyncio
from unittest.mock import Mock, patch
import numpy as np
# Add app to path
import sys
from pathlib import Path
sys.path.append(str(Path(__file__).parent.parent.parent))
from app.quantum.qrng import QuantumRNG, QuantumGenerationResult, EntropyAnalysis
class TestQuantumRNG:
"""Test suite for QuantumRNG class"""
@pytest.fixture
def qrng(self):
"""Create QuantumRNG instance for testing"""
return QuantumRNG(backend="qrisp_simulator")
@pytest.mark.asyncio
async def test_generate_bytes_basic(self, qrng):
"""Test basic byte generation"""
result = await qrng.generate_bytes(32, 8, "hex")
assert isinstance(result, QuantumGenerationResult)
assert result.length == 32
assert result.format == "hex"
assert result.entropy_bits == 32 * 8
assert result.qubits_used == 8
assert len(result.data) == 64 # 32 bytes = 64 hex chars
@pytest.mark.asyncio
async def test_generate_bytes_formats(self, qrng):
"""Test different output formats"""
# Test hex format
result_hex = await qrng.generate_bytes(16, 8, "hex")
assert isinstance(result_hex.data, str)
assert len(result_hex.data) == 32 # 16 bytes = 32 hex chars
# Test base64 format
result_b64 = await qrng.generate_bytes(16, 8, "base64")
assert isinstance(result_b64.data, str)
assert result_b64.data.endswith('=') or len(result_b64.data) % 4 == 0
# Test array format
result_array = await qrng.generate_bytes(16, 8, "array")
assert isinstance(result_array.data, list)
assert len(result_array.data) == 16
assert all(0 <= b <= 255 for b in result_array.data)
# Test raw format
result_raw = await qrng.generate_bytes(16, 8, "raw")
assert isinstance(result_raw.data, bytes)
assert len(result_raw.data) == 16
@pytest.mark.asyncio
async def test_generate_bytes_validation(self, qrng):
"""Test input validation"""
# Test invalid byte count
with pytest.raises(ValueError, match="at least 1"):
await qrng.generate_bytes(0, 8, "hex")
# Test invalid qubit count
with pytest.raises(ValueError, match="at least 1"):
await qrng.generate_bytes(32, 0, "hex")
# Test exceeding qubit limit
with pytest.raises(ValueError, match="exceeds limit"):
await qrng.generate_bytes(32, 100, "hex")
@pytest.mark.asyncio
async def test_generate_key(self, qrng):
"""Test cryptographic key generation"""
# Test AES key generation
result = await qrng.generate_key(256, "AES")
assert isinstance(result.data, dict)
assert result.data["algorithm"] == "AES"
assert result.data["key_size_bits"] == 256
assert len(result.data["key"]) == 64 # 256 bits = 32 bytes = 64 hex chars
# Test invalid key size
with pytest.raises(ValueError, match="Invalid key size"):
await qrng.generate_key(512, "AES")
# Test invalid algorithm
with pytest.raises(ValueError, match="Unsupported algorithm"):
await qrng.generate_key(256, "INVALID")
@pytest.mark.asyncio
async def test_generate_token(self, qrng):
"""Test session token generation"""
# Test URL-safe token
result = await qrng.generate_token(32, url_safe=True)
assert isinstance(result.data, str)
# URL-safe base64 should not contain +, /, or =
assert '+' not in result.data
assert '/' not in result.data
# Test non-URL-safe token
result = await qrng.generate_token(32, url_safe=False)
assert isinstance(result.data, str)
@pytest.mark.asyncio
async def test_generate_uuid(self, qrng):
"""Test UUID generation"""
result = await qrng.generate_uuid(4)
assert isinstance(result.data, str)
assert result.format == "uuid"
# Validate UUID format
uuid_parts = result.data.split('-')
assert len(uuid_parts) == 5
assert len(uuid_parts[0]) == 8
assert len(uuid_parts[1]) == 4
assert len(uuid_parts[2]) == 4
assert len(uuid_parts[3]) == 4
assert len(uuid_parts[4]) == 12
# Check version bits (should be 4)
assert uuid_parts[2][0] == '4'
# Test invalid version
with pytest.raises(ValueError, match="Only UUID v4"):
await qrng.generate_uuid(5)
def test_entropy_pool_management(self, qrng):
"""Test entropy pool updates"""
initial_size = len(qrng.entropy_pool)
# Add measurements to pool
for i in range(10):
qrng._update_entropy_pool(i)
assert len(qrng.entropy_pool) == initial_size + 10
assert qrng.generation_count == 10
# Test pool size limit
for i in range(2000):
qrng._update_entropy_pool(i)
assert len(qrng.entropy_pool) <= qrng.pool_size
def test_entropy_analysis(self, qrng):
"""Test entropy analysis"""
# Test with insufficient data
analysis = qrng.analyze_entropy()
assert analysis.health_status == "insufficient_data"
# Add random data to pool
np.random.seed(42) # For reproducibility
for _ in range(200):
qrng._update_entropy_pool(np.random.randint(0, 256))
# Analyze entropy
analysis = qrng.analyze_entropy()
assert isinstance(analysis, EntropyAnalysis)
assert 0 <= analysis.shannon_entropy <= 1
assert 0 <= analysis.bit_balance <= 1
assert analysis.pool_size == 200
assert isinstance(analysis.passed_tests, dict)
assert analysis.health_status in ["excellent", "good", "poor"]
def test_statistics(self, qrng):
"""Test statistics tracking"""
stats = qrng.get_statistics()
assert isinstance(stats, dict)
assert "total_bytes_generated" in stats
assert "total_generations" in stats
assert "average_generation_time_ms" in stats
assert "entropy_pool_size" in stats
assert "backend" in stats
assert "backend_status" in stats
@pytest.mark.asyncio
async def test_performance(self, qrng):
"""Test generation performance"""
import time
# Generate different sizes and measure time
sizes = [32, 128, 512, 1024]
for size in sizes:
start_time = time.time()
result = await qrng.generate_bytes(size, 8, "hex")
elapsed = (time.time() - start_time) * 1000
assert result.length == size
assert result.generation_time_ms > 0
assert result.generation_time_ms < 5000 # Should be < 5 seconds
print(f"Generated {size} bytes in {elapsed:.2f}ms")
@pytest.mark.asyncio
async def test_randomness_quality(self, qrng):
"""Test quality of generated random numbers"""
# Generate a large sample
result = await qrng.generate_bytes(1000, 8, "array")
data = np.array(result.data)
# Test uniform distribution
mean = np.mean(data)
assert 100 < mean < 155 # Should be around 127.5
# Test standard deviation
std = np.std(data)
assert 60 < std < 90 # Should be around 74
# Test uniqueness
unique_ratio = len(np.unique(data)) / len(data)
assert unique_ratio > 0.2 # At least 20% unique values
# Test no obvious patterns
diffs = np.diff(data)
assert np.std(diffs) > 50 # Differences should vary
@pytest.mark.asyncio
async def test_concurrent_generation(self, qrng):
"""Test concurrent generation requests"""
# Create multiple concurrent tasks
tasks = []
for i in range(10):
task = qrng.generate_bytes(32, 8, "hex")
tasks.append(task)
# Execute concurrently
results = await asyncio.gather(*tasks)
# Verify all results are unique
assert len(results) == 10
hex_values = [r.data for r in results]
assert len(set(hex_values)) == 10 # All should be different
def test_request_id_generation(self, qrng):
"""Test request ID uniqueness"""
ids = set()
for _ in range(100):
request_id = qrng._generate_request_id()
assert request_id not in ids # Should be unique
ids.add(request_id)
assert request_id.startswith("req_")
@pytest.mark.integration
class TestQuantumRNGIntegration:
"""Integration tests for QuantumRNG"""
@pytest.mark.asyncio
async def test_full_workflow(self):
"""Test complete workflow from initialization to analysis"""
# Initialize
qrng = QuantumRNG()
# Generate various types of random data
bytes_result = await qrng.generate_bytes(64, 8, "hex")
key_result = await qrng.generate_key(256, "AES")
token_result = await qrng.generate_token(32, True)
uuid_result = await qrng.generate_uuid(4)
# Verify all succeeded
assert bytes_result.length == 64
assert key_result.data["key_size_bits"] == 256
assert len(token_result.data) > 0
assert '-' in uuid_result.data
# Check entropy pool was updated
assert len(qrng.entropy_pool) > 0
# Get statistics
stats = qrng.get_statistics()
assert stats["total_bytes_generated"] > 0
assert stats["total_generations"] > 0
if __name__ == "__main__":
pytest.main([__file__, "-v", "-s"]) |