385 lines
13 KiB
Python
385 lines
13 KiB
Python
"""
|
|
Unit tests for timestamp standardization across all models.
|
|
Ensures all timestamp fields use float UTC timestamps consistently.
|
|
"""
|
|
|
|
import pytest
|
|
from datetime import datetime, timedelta
|
|
import time
|
|
|
|
from modules.interfaces.interfaceAppModel import UserConnection, Session, AuthEvent, Token
|
|
from modules.interfaces.interfaceChatModel import TaskAction, ChatLog, ChatMessage, ChatWorkflow, TaskItem, TaskHandover
|
|
from modules.interfaces.interfaceComponentModel import FileItem
|
|
from modules.shared.timezoneUtils import get_utc_timestamp, create_expiration_timestamp
|
|
|
|
|
|
class TestTimestampModelConsistency:
|
|
"""Test that all models use float UTC timestamps consistently."""
|
|
|
|
def test_user_connection_timestamps(self):
|
|
"""Test UserConnection model timestamp fields."""
|
|
current_time = get_utc_timestamp()
|
|
expires_at = create_expiration_timestamp(3600) # 1 hour from now
|
|
|
|
connection = UserConnection(
|
|
userId="user123",
|
|
authority="msft",
|
|
externalId="ext123",
|
|
externalUsername="testuser",
|
|
connectedAt=current_time,
|
|
lastChecked=current_time,
|
|
expiresAt=expires_at
|
|
)
|
|
|
|
# Verify types
|
|
assert isinstance(connection.connectedAt, float)
|
|
assert isinstance(connection.lastChecked, float)
|
|
assert isinstance(connection.expiresAt, float)
|
|
|
|
# Verify values are reasonable UTC timestamps
|
|
assert connection.connectedAt > 1600000000 # After 2020
|
|
assert connection.lastChecked > 1600000000
|
|
assert connection.expiresAt > connection.connectedAt
|
|
|
|
# Test to_dict() method
|
|
connection_dict = connection.to_dict()
|
|
# Note: to_dict() converts timestamps to ISO strings, so we check for string type
|
|
assert isinstance(connection_dict["connectedAt"], str)
|
|
assert isinstance(connection_dict["lastChecked"], str)
|
|
assert isinstance(connection_dict["expiresAt"], str)
|
|
|
|
def test_session_timestamps(self):
|
|
"""Test Session model timestamp fields."""
|
|
current_time = get_utc_timestamp()
|
|
expires_at = create_expiration_timestamp(7200) # 2 hours from now
|
|
|
|
session = Session(
|
|
id="session123",
|
|
userId="user123",
|
|
tokenId="token123",
|
|
lastActivity=current_time,
|
|
expiresAt=expires_at
|
|
)
|
|
|
|
# Verify types
|
|
assert isinstance(session.lastActivity, float)
|
|
assert isinstance(session.expiresAt, float)
|
|
|
|
# Verify values
|
|
assert session.lastActivity > 1600000000
|
|
assert session.expiresAt > session.lastActivity
|
|
|
|
# Test to_dict() method
|
|
session_dict = session.to_dict()
|
|
# Note: to_dict() converts timestamps to ISO strings, so we check for string type
|
|
assert isinstance(session_dict["lastActivity"], str)
|
|
assert isinstance(session_dict["expiresAt"], str)
|
|
|
|
def test_auth_event_timestamps(self):
|
|
"""Test AuthEvent model timestamp fields."""
|
|
current_time = get_utc_timestamp()
|
|
|
|
auth_event = AuthEvent(
|
|
id="event123",
|
|
userId="user123",
|
|
eventType="login",
|
|
details={"action": "login", "success": True},
|
|
timestamp=current_time
|
|
)
|
|
|
|
# Verify types
|
|
assert isinstance(auth_event.timestamp, float)
|
|
|
|
# Verify values
|
|
assert auth_event.timestamp > 1600000000
|
|
|
|
# Test to_dict() method
|
|
event_dict = auth_event.to_dict()
|
|
# Note: to_dict() converts timestamps to ISO strings, so we check for string type
|
|
assert isinstance(event_dict["timestamp"], str)
|
|
|
|
def test_token_timestamps(self):
|
|
"""Test Token model timestamp fields."""
|
|
current_time = get_utc_timestamp()
|
|
expires_at = create_expiration_timestamp(3600)
|
|
|
|
token = Token(
|
|
userId="user123",
|
|
authority="msft",
|
|
tokenAccess="access_token",
|
|
expiresAt=expires_at,
|
|
createdAt=current_time
|
|
)
|
|
|
|
# Verify types
|
|
assert isinstance(token.expiresAt, float)
|
|
assert isinstance(token.createdAt, float)
|
|
|
|
# Verify values
|
|
assert token.expiresAt > 1600000000
|
|
assert token.createdAt > 1600000000
|
|
assert token.expiresAt > token.createdAt
|
|
|
|
# Test to_dict() method
|
|
token_dict = token.to_dict()
|
|
# Note: to_dict() converts timestamps to ISO strings, so we check for string type
|
|
assert isinstance(token_dict["expiresAt"], str)
|
|
assert isinstance(token_dict["createdAt"], str)
|
|
|
|
def test_task_action_timestamps(self):
|
|
"""Test TaskAction model timestamp fields."""
|
|
current_time = get_utc_timestamp()
|
|
|
|
task_action = TaskAction(
|
|
id="action123",
|
|
execMethod="test.method",
|
|
execAction="test_action",
|
|
timestamp=current_time
|
|
)
|
|
|
|
# Verify types
|
|
assert isinstance(task_action.timestamp, float)
|
|
|
|
# Verify values
|
|
assert task_action.timestamp > 1600000000
|
|
|
|
# Test default factory
|
|
task_action_default = TaskAction(
|
|
id="action124",
|
|
execMethod="test.method",
|
|
execAction="test_action"
|
|
)
|
|
assert isinstance(task_action_default.timestamp, float)
|
|
assert task_action_default.timestamp > 1600000000
|
|
|
|
def test_chat_log_timestamps(self):
|
|
"""Test ChatLog model timestamp fields."""
|
|
current_time = get_utc_timestamp()
|
|
|
|
chat_log = ChatLog(
|
|
id="log123",
|
|
workflowId="workflow123",
|
|
message="Test message",
|
|
type="info",
|
|
timestamp=current_time
|
|
)
|
|
|
|
# Verify types
|
|
assert isinstance(chat_log.timestamp, float)
|
|
|
|
# Verify values
|
|
assert chat_log.timestamp > 1600000000
|
|
|
|
def test_chat_message_timestamps(self):
|
|
"""Test ChatMessage model timestamp fields."""
|
|
current_time = get_utc_timestamp()
|
|
|
|
chat_message = ChatMessage(
|
|
id="msg123",
|
|
workflowId="workflow123",
|
|
role="user",
|
|
status="first",
|
|
sequenceNr=1,
|
|
message="Test message",
|
|
publishedAt=current_time
|
|
)
|
|
|
|
# Verify types
|
|
assert isinstance(chat_message.publishedAt, float)
|
|
|
|
# Verify values
|
|
assert chat_message.publishedAt > 1600000000
|
|
|
|
def test_chat_workflow_timestamps(self):
|
|
"""Test ChatWorkflow model timestamp fields."""
|
|
current_time = get_utc_timestamp()
|
|
|
|
workflow = ChatWorkflow(
|
|
id="workflow123",
|
|
mandateId="mandate123",
|
|
status="active",
|
|
currentRound=1,
|
|
startedAt=current_time,
|
|
lastActivity=current_time
|
|
)
|
|
|
|
# Verify types
|
|
assert isinstance(workflow.startedAt, float)
|
|
assert isinstance(workflow.lastActivity, float)
|
|
|
|
# Verify values
|
|
assert workflow.startedAt > 1600000000
|
|
assert workflow.lastActivity > 1600000000
|
|
|
|
def test_task_item_timestamps(self):
|
|
"""Test TaskItem model timestamp fields."""
|
|
current_time = get_utc_timestamp()
|
|
finished_time = current_time + 300 # 5 minutes later
|
|
|
|
task_item = TaskItem(
|
|
id="task123",
|
|
workflowId="workflow123",
|
|
userInput="Test user input",
|
|
startedAt=current_time,
|
|
finishedAt=finished_time
|
|
)
|
|
|
|
# Verify types
|
|
assert isinstance(task_item.startedAt, float)
|
|
assert isinstance(task_item.finishedAt, float)
|
|
|
|
# Verify values
|
|
assert task_item.startedAt > 1600000000
|
|
assert task_item.finishedAt > task_item.startedAt
|
|
|
|
def test_task_handover_timestamps(self):
|
|
"""Test TaskHandover model timestamp fields."""
|
|
current_time = get_utc_timestamp()
|
|
|
|
handover = TaskHandover(
|
|
taskId="task123",
|
|
timestamp=current_time
|
|
)
|
|
|
|
# Verify types
|
|
assert isinstance(handover.timestamp, float)
|
|
|
|
# Verify values
|
|
assert handover.timestamp > 1600000000
|
|
|
|
# Test default factory
|
|
handover_default = TaskHandover(
|
|
taskId="task124"
|
|
)
|
|
assert isinstance(handover_default.timestamp, float)
|
|
assert handover_default.timestamp > 1600000000
|
|
|
|
def test_file_item_timestamps(self):
|
|
"""Test FileItem model timestamp fields."""
|
|
current_time = get_utc_timestamp()
|
|
|
|
file_item = FileItem(
|
|
id="file123",
|
|
mandateId="mandate123",
|
|
filename="test.txt",
|
|
mimeType="text/plain",
|
|
fileHash="abc123hash",
|
|
fileSize=1024,
|
|
creationDate=current_time
|
|
)
|
|
|
|
# Verify types
|
|
assert isinstance(file_item.creationDate, float)
|
|
|
|
# Verify values
|
|
assert file_item.creationDate > 1600000000
|
|
|
|
# Test default factory
|
|
file_item_default = FileItem(
|
|
id="file124",
|
|
mandateId="mandate123",
|
|
filename="test.txt",
|
|
mimeType="text/plain",
|
|
fileHash="def456hash",
|
|
fileSize=2048
|
|
)
|
|
assert isinstance(file_item_default.creationDate, float)
|
|
assert file_item_default.creationDate > 1600000000
|
|
|
|
# Test to_dict() method
|
|
file_dict = file_item.to_dict()
|
|
# Note: to_dict() converts timestamps to ISO strings, so we check for string type
|
|
assert isinstance(file_dict["creationDate"], str)
|
|
|
|
|
|
class TestTimestampGenerationFunctions:
|
|
"""Test timestamp generation utility functions."""
|
|
|
|
def test_get_utc_timestamp(self):
|
|
"""Test get_utc_timestamp function."""
|
|
timestamp = get_utc_timestamp()
|
|
|
|
# Verify type
|
|
assert isinstance(timestamp, float)
|
|
|
|
# Verify value is reasonable
|
|
assert timestamp > 1600000000 # After 2020
|
|
assert timestamp < 4102444800 # Before 2100
|
|
|
|
# Verify it's close to current time (within 2 seconds to account for execution time)
|
|
current_time = time.time()
|
|
assert abs(timestamp - current_time) < 2
|
|
|
|
def test_create_expiration_timestamp(self):
|
|
"""Test create_expiration_timestamp function."""
|
|
current_time = get_utc_timestamp()
|
|
expires_in = 3600 # 1 hour
|
|
|
|
expiration_timestamp = create_expiration_timestamp(expires_in)
|
|
|
|
# Verify type
|
|
assert isinstance(expiration_timestamp, float)
|
|
|
|
# Verify value
|
|
assert expiration_timestamp > current_time
|
|
# Check if it's close to current_time + expires_in (within 1 second to account for execution time)
|
|
assert abs(expiration_timestamp - (current_time + expires_in)) < 1
|
|
|
|
# Verify it's reasonable
|
|
assert expiration_timestamp > 1600000000
|
|
assert expiration_timestamp < 4102444800
|
|
|
|
|
|
class TestModelValidation:
|
|
"""Test model validation and constraints."""
|
|
|
|
def test_timestamp_field_descriptions(self):
|
|
"""Test that all timestamp fields have proper descriptions mentioning UTC."""
|
|
# Test UserConnection
|
|
connection = UserConnection(
|
|
userId="user123",
|
|
authority="msft",
|
|
externalId="ext123",
|
|
externalUsername="testuser"
|
|
)
|
|
|
|
# Check field descriptions contain UTC timestamp info
|
|
# Note: This test depends on the actual field descriptions in the model
|
|
# For now, we'll just verify the fields exist
|
|
# Handle both Pydantic v1 and v2
|
|
if hasattr(connection, 'model_fields'):
|
|
fields = connection.model_fields
|
|
else:
|
|
fields = connection.__fields__
|
|
|
|
assert "connectedAt" in fields
|
|
assert "lastChecked" in fields
|
|
assert "expiresAt" in fields
|
|
|
|
def test_optional_timestamp_fields(self):
|
|
"""Test that optional timestamp fields work correctly."""
|
|
# Test Token with optional createdAt
|
|
token = Token(
|
|
userId="user123",
|
|
authority="msft",
|
|
tokenAccess="access_token",
|
|
expiresAt=create_expiration_timestamp(3600)
|
|
)
|
|
|
|
# createdAt should be None by default
|
|
assert token.createdAt is None
|
|
|
|
# Test UserConnection with optional expiresAt
|
|
connection = UserConnection(
|
|
userId="user123",
|
|
authority="msft",
|
|
externalId="ext123",
|
|
externalUsername="testuser"
|
|
)
|
|
|
|
# expiresAt should be None by default
|
|
assert connection.expiresAt is None
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__])
|