Spaces:
Sleeping
Sleeping
File size: 6,956 Bytes
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 |
# test_properties_queue_advancement.py
"""
Property-based tests for message queue advancement.
Tests that the message queue advances correctly after verification.
"""
import pytest
from hypothesis import given, strategies as st, settings, HealthCheck
from datetime import datetime
from src.core.verification_models import (
VerificationRecord,
VerificationSession,
TestMessage,
)
from src.core.message_queue_manager import MessageQueueManager
def message_strategy():
"""Generate random test messages with unique IDs."""
return st.builds(
TestMessage,
message_id=st.uuids().map(str),
text=st.text(min_size=1, max_size=500),
pre_classified_label=st.sampled_from(["green", "yellow", "red"]),
)
class TestQueueAdvancement:
"""
**Feature: verification-mode, Property 2: Queue Advances After Verification**
Tests that the message queue advances correctly after verification.
"""
@given(st.lists(message_strategy(), min_size=1, max_size=20))
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture])
def test_queue_advances_after_verification(self, messages):
"""
**Feature: verification-mode, Property 2: Queue Advances After Verification**
**Validates: Requirements 3.2, 3.5, 4.2**
For any message queue, when a verifier submits feedback on a message,
the next message in the queue should be displayed, and the verified
message should no longer be in the active queue.
"""
# Create a session and initialize queue
session = VerificationSession(
session_id="test_session",
verifier_name="Test Verifier",
dataset_id="test_dataset",
dataset_name="Test Dataset",
)
queue_manager = MessageQueueManager(session)
queue_manager.initialize_queue(messages)
# Get initial state
initial_message_id = queue_manager.get_current_message_id()
initial_position = queue_manager.get_queue_position()
# Verify initial state
assert initial_message_id is not None
assert initial_position == (1, len(messages))
# Advance the queue
advanced = queue_manager.advance_queue()
# Verify advancement
if len(messages) > 1:
assert advanced is True
next_message_id = queue_manager.get_current_message_id()
next_position = queue_manager.get_queue_position()
# Next message should be different from initial
assert next_message_id != initial_message_id
# Position should have incremented
assert next_position[0] == initial_position[0] + 1
# Verified message should be in verified list
assert initial_message_id in session.verified_message_ids
else:
# Single message queue should be complete after advance
assert advanced is False
assert queue_manager.is_queue_complete()
@given(st.lists(message_strategy(), min_size=2, max_size=20))
def test_queue_advances_multiple_times(self, messages):
"""
**Feature: verification-mode, Property 2: Queue Advances After Verification**
**Validates: Requirements 3.2, 3.5, 4.2**
For any message queue with multiple messages, advancing through all
messages should result in queue completion.
"""
session = VerificationSession(
session_id="test_session",
verifier_name="Test Verifier",
dataset_id="test_dataset",
dataset_name="Test Dataset",
)
queue_manager = MessageQueueManager(session)
queue_manager.initialize_queue(messages)
# Advance through all messages
message_count = len(messages)
for i in range(message_count):
if i < message_count - 1:
# Should be able to advance
assert queue_manager.advance_queue() is True
else:
# Last advance should complete the queue
assert queue_manager.advance_queue() is False
# Queue should be complete
assert queue_manager.is_queue_complete()
# All messages should be verified
assert len(session.verified_message_ids) == message_count
@given(st.lists(message_strategy(), min_size=1, max_size=20))
def test_verified_messages_not_in_active_queue(self, messages):
"""
**Feature: verification-mode, Property 2: Queue Advances After Verification**
**Validates: Requirements 3.2, 3.5, 4.2**
For any message queue, verified messages should not be in the active
queue position after advancement.
"""
session = VerificationSession(
session_id="test_session",
verifier_name="Test Verifier",
dataset_id="test_dataset",
dataset_name="Test Dataset",
)
queue_manager = MessageQueueManager(session)
queue_manager.initialize_queue(messages)
verified_ids = []
# Verify first message and advance
if len(messages) > 0:
first_msg_id = queue_manager.get_current_message_id()
verified_ids.append(first_msg_id)
queue_manager.advance_queue()
# Current message should not be in verified list
current_msg_id = queue_manager.get_current_message_id()
if current_msg_id:
assert current_msg_id not in verified_ids
# Verified message should be in verified list
assert first_msg_id in session.verified_message_ids
@given(st.lists(message_strategy(), min_size=1, max_size=20))
def test_queue_position_tracking(self, messages):
"""
**Feature: verification-mode, Property 2: Queue Advances After Verification**
**Validates: Requirements 3.2, 3.5, 4.2**
For any message queue, the queue position should accurately track
progress through the queue.
"""
session = VerificationSession(
session_id="test_session",
verifier_name="Test Verifier",
dataset_id="test_dataset",
dataset_name="Test Dataset",
)
queue_manager = MessageQueueManager(session)
queue_manager.initialize_queue(messages)
# Check initial position
pos, total = queue_manager.get_queue_position()
assert pos == 1
assert total == len(messages)
# Advance and check position increments
if len(messages) > 1:
queue_manager.advance_queue()
pos, total = queue_manager.get_queue_position()
assert pos == 2
assert total == len(messages)
|