Spaces:
Sleeping
Sleeping
File size: 4,172 Bytes
0304d75 | 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 | """
tests/test_temporal_model.py
============================
Unit tests for intervention/temporal_model.py — SecureTemporalModel.
"""
import pytest
from intervention.temporal_model import SecureTemporalModel
from security.auth import decrypt_data
class TestSecureTemporalModel:
def test_first_score_no_history(self):
"""First score with no prior history should work."""
model = SecureTemporalModel()
analysis, encrypted = model.process(score=0.5)
assert analysis.score == 0.5
assert analysis.score_count == 1
assert isinstance(encrypted, str)
assert len(encrypted) > 0
def test_history_encrypted_at_rest(self):
"""The returned history should be encrypted (not plaintext)."""
model = SecureTemporalModel()
_, encrypted = model.process(score=0.5)
# Should not contain plaintext score
assert "0.5" not in encrypted
# But should be decryptable
decrypted = decrypt_data(encrypted)
assert isinstance(decrypted, list)
assert len(decrypted) == 1
def test_chained_scores(self):
"""Multiple scores should accumulate in the encrypted history."""
model = SecureTemporalModel()
_, enc1 = model.process(score=0.3)
_, enc2 = model.process(score=0.5, encrypted_history=enc1)
analysis, enc3 = model.process(score=0.7, encrypted_history=enc2)
assert analysis.score_count == 3
decrypted = decrypt_data(enc3)
assert len(decrypted) == 3
def test_velocity_computed_after_multiple_scores(self):
"""Velocity should be computed after sufficient history."""
model = SecureTemporalModel()
encrypted = None
for score in [0.2, 0.4, 0.6, 0.8]:
analysis, encrypted = model.process(
score=score, encrypted_history=encrypted
)
assert analysis.stress_velocity is not None
assert analysis.stress_velocity > 0
def test_volatility_detection(self):
"""Alternating scores should trigger volatility."""
model = SecureTemporalModel(volatility_threshold=0.2)
encrypted = None
for score in [0.1, 0.9, 0.1, 0.9, 0.1]:
analysis, encrypted = model.process(
score=score, encrypted_history=encrypted
)
assert analysis.is_volatile is True
def test_max_history_respected(self):
"""History should not exceed max_history."""
model = SecureTemporalModel(max_history=5)
encrypted = None
for i in range(20):
_, encrypted = model.process(
score=(i % 10) / 10, encrypted_history=encrypted
)
decrypted = decrypt_data(encrypted)
assert len(decrypted) == 5
def test_corrupted_history_falls_back_to_empty(self):
"""Corrupted/invalid encrypted history should not crash — starts fresh."""
model = SecureTemporalModel()
analysis, encrypted = model.process(
score=0.6, encrypted_history="not-valid-ciphertext"
)
assert analysis.score == 0.6
assert analysis.score_count == 1
assert isinstance(encrypted, str)
def test_wrong_key_history_falls_back_to_empty(self):
"""History encrypted with a different Fernet key should not crash."""
from cryptography.fernet import Fernet
import json
other_key = Fernet.generate_key()
other_fernet = Fernet(other_key)
foreign_encrypted = other_fernet.encrypt(
json.dumps([[1000.0, 0.5]]).encode()
).decode()
model = SecureTemporalModel()
analysis, encrypted = model.process(
score=0.7, encrypted_history=foreign_encrypted
)
assert analysis.score == 0.7
assert analysis.score_count == 1
def test_custom_timestamp(self):
"""Custom timestamps should be stored correctly."""
model = SecureTemporalModel()
analysis, encrypted = model.process(score=0.5, timestamp=1000.0)
decrypted = decrypt_data(encrypted)
assert decrypted[0][0] == 1000.0
assert decrypted[0][1] == 0.5
|