Spaces:
Sleeping
Sleeping
File size: 14,962 Bytes
a3934b1 7bbd836 a3934b1 2e0e95f a3934b1 |
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 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 |
# test_properties_persistence.py
"""
Property-based tests for verification data persistence.
Tests that verification records and sessions persist correctly.
"""
import pytest
from hypothesis import given, strategies as st, settings, HealthCheck
from datetime import datetime
from src.core.verification_models import (
VerificationRecord,
VerificationSession,
)
from src.core.verification_store import JSONVerificationStore
# Strategies for generating test data
def valid_id_strategy():
"""Generate valid IDs for use as filenames."""
return st.text(
alphabet="abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-",
min_size=1,
max_size=20,
)
def verification_record_strategy():
"""Generate random verification records with consistent is_correct field."""
# Generate classifier_decision and ground_truth_label together to ensure is_correct is consistent
@st.composite
def build_record(draw):
message_id = draw(valid_id_strategy())
original_message = draw(st.text(min_size=1, max_size=500))
classifier_decision = draw(st.sampled_from(["green", "yellow", "red"]))
classifier_confidence = draw(st.floats(min_value=0.0, max_value=1.0))
classifier_indicators = draw(st.lists(st.text(min_size=1, max_size=50), max_size=5))
verifier_notes = draw(st.text(max_size=200))
# Decide if this should be correct or incorrect
is_correct = draw(st.booleans())
# Set ground_truth_label based on is_correct
if is_correct:
ground_truth_label = classifier_decision
else:
# Pick a different label
other_labels = [l for l in ["green", "yellow", "red"] if l != classifier_decision]
ground_truth_label = draw(st.sampled_from(other_labels))
return VerificationRecord(
message_id=message_id,
original_message=original_message,
classifier_decision=classifier_decision,
classifier_confidence=classifier_confidence,
classifier_indicators=classifier_indicators,
ground_truth_label=ground_truth_label,
verifier_notes=verifier_notes,
is_correct=is_correct,
timestamp=datetime.now(),
)
return build_record()
def verification_session_strategy():
"""Generate random verification sessions."""
return st.builds(
VerificationSession,
session_id=valid_id_strategy(),
verifier_name=st.text(min_size=1, max_size=50),
dataset_id=valid_id_strategy(),
dataset_name=st.text(min_size=1, max_size=100),
created_at=st.just(datetime.now()),
completed_at=st.none(),
total_messages=st.integers(min_value=1, max_value=100),
verified_count=st.integers(min_value=0, max_value=100),
correct_count=st.integers(min_value=0, max_value=100),
incorrect_count=st.integers(min_value=0, max_value=100),
verifications=st.just([]),
is_complete=st.booleans(),
)
class TestVerificationRecordPersistence:
"""
**Feature: verification-mode, Property 1: Feedback Saves Correctly**
Tests that verification records save and load correctly with all fields intact.
"""
@given(verification_record_strategy())
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture])
def test_record_saves_and_loads_correctly(self, verification_store, record):
"""
**Feature: verification-mode, Property 1: Feedback Saves Correctly**
**Validates: Requirements 3.2, 3.5, 8.1**
For any verification record, when saved to storage and then loaded,
all fields should be preserved exactly.
"""
# Create a session to hold the record
session = VerificationSession(
session_id="test_session",
verifier_name="Test Verifier",
dataset_id="test_dataset",
dataset_name="Test Dataset",
total_messages=1,
)
verification_store.save_session(session)
# Save the verification record
verification_store.save_verification("test_session", record)
# Load the session and verify the record
loaded_session = verification_store.load_session("test_session")
assert loaded_session is not None
assert len(loaded_session.verifications) == 1
loaded_record = loaded_session.verifications[0]
# Verify all fields are preserved
assert loaded_record.message_id == record.message_id
assert loaded_record.original_message == record.original_message
assert loaded_record.classifier_decision == record.classifier_decision
assert loaded_record.classifier_confidence == record.classifier_confidence
assert loaded_record.classifier_indicators == record.classifier_indicators
assert loaded_record.ground_truth_label == record.ground_truth_label
assert loaded_record.verifier_notes == record.verifier_notes
assert loaded_record.is_correct == record.is_correct
@given(verification_record_strategy())
def test_record_to_dict_and_back(self, record):
"""
**Feature: verification-mode, Property 1: Feedback Saves Correctly**
**Validates: Requirements 3.2, 3.5, 8.1**
For any verification record, converting to dict and back should
preserve all fields.
"""
# Convert to dict and back
record_dict = record.to_dict()
restored_record = VerificationRecord.from_dict(record_dict)
# Verify all fields match
assert restored_record.message_id == record.message_id
assert restored_record.original_message == record.original_message
assert restored_record.classifier_decision == record.classifier_decision
assert restored_record.classifier_confidence == record.classifier_confidence
assert restored_record.classifier_indicators == record.classifier_indicators
assert restored_record.ground_truth_label == record.ground_truth_label
assert restored_record.verifier_notes == record.verifier_notes
assert restored_record.is_correct == record.is_correct
class TestSessionStatePersistence:
"""
**Feature: verification-mode, Property 3: Session State Persists**
Tests that verification sessions persist and can be resumed with state intact.
"""
@given(verification_session_strategy())
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture])
def test_session_saves_and_loads_correctly(self, verification_store, session):
"""
**Feature: verification-mode, Property 3: Session State Persists**
**Validates: Requirements 8.2, 8.3**
For any verification session, when saved and then loaded,
all session state should be preserved exactly.
"""
# Save the session
verification_store.save_session(session)
# Load the session
loaded_session = verification_store.load_session(session.session_id)
# Verify all fields are preserved
assert loaded_session is not None
assert loaded_session.session_id == session.session_id
assert loaded_session.verifier_name == session.verifier_name
assert loaded_session.dataset_id == session.dataset_id
assert loaded_session.dataset_name == session.dataset_name
assert loaded_session.total_messages == session.total_messages
assert loaded_session.verified_count == session.verified_count
assert loaded_session.correct_count == session.correct_count
assert loaded_session.incorrect_count == session.incorrect_count
assert loaded_session.is_complete == session.is_complete
@given(verification_session_strategy())
def test_session_to_dict_and_back(self, session):
"""
**Feature: verification-mode, Property 3: Session State Persists**
**Validates: Requirements 8.2, 8.3**
For any verification session, converting to dict and back should
preserve all session state.
"""
# Convert to dict and back
session_dict = session.to_dict()
restored_session = VerificationSession.from_dict(session_dict)
# Verify all fields match
assert restored_session.session_id == session.session_id
assert restored_session.verifier_name == session.verifier_name
assert restored_session.dataset_id == session.dataset_id
assert restored_session.dataset_name == session.dataset_name
assert restored_session.total_messages == session.total_messages
assert restored_session.verified_count == session.verified_count
assert restored_session.correct_count == session.correct_count
assert restored_session.incorrect_count == session.incorrect_count
assert restored_session.is_complete == session.is_complete
@given(verification_session_strategy())
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
def test_session_with_multiple_records_persists(
self, verification_store, session
):
"""
**Feature: verification-mode, Property 3: Session State Persists**
**Validates: Requirements 8.2, 8.3**
For any session with multiple verification records, when saved and loaded,
all records and session state should be preserved.
"""
# Ensure session is not already marked complete
session.is_complete = False
session.completed_at = None
# Generate records with unique message IDs
records = []
for i in range(5):
record = VerificationRecord(
message_id=f"msg_{i}",
original_message=f"Test message {i}",
classifier_decision="green",
classifier_confidence=0.9,
classifier_indicators=["test"],
ground_truth_label="green",
verifier_notes="",
is_correct=True,
timestamp=datetime.now(),
)
records.append(record)
# Save the session
verification_store.save_session(session)
# Add records to the session
for record in records:
verification_store.save_verification(session.session_id, record)
# Load the session
loaded_session = verification_store.load_session(session.session_id)
# Verify session state
assert loaded_session is not None
assert loaded_session.session_id == session.session_id
assert len(loaded_session.verifications) == len(records)
# Verify all records are preserved
for i, original_record in enumerate(records):
loaded_record = loaded_session.verifications[i]
assert loaded_record.message_id == original_record.message_id
assert loaded_record.original_message == original_record.original_message
assert (
loaded_record.classifier_decision
== original_record.classifier_decision
)
class TestCompletedSessionImmutability:
"""
**Feature: verification-mode, Property 13: Completed Sessions Cannot be Modified**
Tests that completed sessions cannot be modified after completion.
"""
@given(verification_session_strategy(), verification_record_strategy())
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture])
def test_completed_session_cannot_be_modified(
self, verification_store, session, record
):
"""
**Feature: verification-mode, Property 13: Completed Sessions Cannot be Modified**
**Validates: Requirements 8.4**
For any completed verification session, attempting to add new verifications
should raise an error and the session should remain unchanged.
"""
# Save the session
verification_store.save_session(session)
# Mark session as complete
verification_store.mark_session_complete(session.session_id)
# Verify session is marked complete
loaded_session = verification_store.load_session(session.session_id)
assert loaded_session.is_complete is True
assert loaded_session.completed_at is not None
# Attempt to add a verification record to completed session
with pytest.raises(ValueError, match="Cannot modify completed session"):
verification_store.save_verification(session.session_id, record)
@given(verification_session_strategy())
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture])
def test_can_modify_session_returns_false_for_completed(
self, verification_store, session
):
"""
**Feature: verification-mode, Property 13: Completed Sessions Cannot be Modified**
**Validates: Requirements 8.4**
For any completed session, can_modify_session should return False.
"""
# Ensure session is not already marked complete
session.is_complete = False
session.completed_at = None
# Save the session
verification_store.save_session(session)
# Initially should be modifiable
assert verification_store.can_modify_session(session.session_id) is True
# Mark session as complete
verification_store.mark_session_complete(session.session_id)
# Now should not be modifiable
assert verification_store.can_modify_session(session.session_id) is False
@given(verification_session_strategy())
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture])
def test_completed_session_persists_completion_state(
self, verification_store, session
):
"""
**Feature: verification-mode, Property 13: Completed Sessions Cannot be Modified**
**Validates: Requirements 8.4**
For any completed session, when saved and reloaded, the completion state
should be preserved.
"""
# Save the session
verification_store.save_session(session)
# Mark session as complete
verification_store.mark_session_complete(session.session_id)
# Load the session
loaded_session = verification_store.load_session(session.session_id)
# Verify completion state is preserved
assert loaded_session.is_complete is True
assert loaded_session.completed_at is not None
# Verify it still cannot be modified
assert verification_store.can_modify_session(session.session_id) is False
|