File size: 10,788 Bytes
4e96b7a |
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 |
"""
Test per funzioni utility.
"""
import os
import sys
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'src')))
import pytest
import json
import tempfile
import os
from unittest.mock import Mock, patch, MagicMock
from datetime import datetime
from utils import (
validate_file_upload, export_results_json, get_confirmed_docs_count,
add_chat_message, add_crewai_result, get_system_stats
)
class TestFileValidation:
"""Test validazione file"""
def test_validate_file_upload_valid(self):
"""Test file valido"""
mock_file = Mock()
mock_file.name = "test.txt"
mock_file.size = 1024 # 1KB
assert validate_file_upload(mock_file) == True
def test_validate_file_upload_none(self):
"""Test file None"""
assert validate_file_upload(None) == False
@patch('streamlit.error')
def test_validate_file_upload_wrong_extension(self, mock_error):
"""Test estensione file sbagliata"""
mock_file = Mock()
mock_file.name = "test.pdf"
mock_file.size = 1024
result = validate_file_upload(mock_file)
assert result == False
mock_error.assert_called_once()
@patch('streamlit.error')
def test_validate_file_upload_too_large(self, mock_error):
"""Test file troppo grande"""
mock_file = Mock()
mock_file.name = "test.txt"
mock_file.size = 11 * 1024 * 1024 # 11MB
result = validate_file_upload(mock_file)
assert result == False
mock_error.assert_called_once()
class TestExportResults:
"""Test export risultati"""
def test_export_results_json_basic(self):
"""Test export JSON base"""
data = {"test": "value", "number": 123}
result = export_results_json(data, "test")
# Verifica che sia JSON valido
parsed = json.loads(result)
assert parsed["test"] == "value"
assert parsed["number"] == 123
assert "metadata" in parsed
assert "exported_at" in parsed["metadata"]
def test_export_results_json_with_datetime(self):
"""Test export con datetime"""
data = {"timestamp": datetime.now()}
result = export_results_json(data, "test")
# Non dovrebbe lanciare errori
parsed = json.loads(result)
assert "timestamp" in parsed
def test_export_results_json_metadata(self):
"""Test metadati export"""
data = {"item1": "value1", "item2": "value2"}
result = export_results_json(data, "test")
parsed = json.loads(result)
assert "metadata" in parsed
assert parsed["metadata"]["total_items"] == 2
assert "exported_at" in parsed["metadata"]
# Verifica formato ISO datetime
timestamp = parsed["metadata"]["exported_at"]
datetime.fromisoformat(timestamp.replace('Z', '+00:00'))
class TestSessionStateHelpers:
"""Test helper per session state"""
@patch('streamlit.session_state', {})
def test_get_confirmed_docs_count_empty(self):
"""Test conteggio documenti confermati vuoto"""
result = get_confirmed_docs_count()
assert result == 0
@patch('streamlit.session_state')
def test_get_confirmed_docs_count_with_docs(self, mock_session):
"""Test conteggio documenti confermati"""
mock_session.get.return_value = {
'doc1': {'confirmed': True},
'doc2': {'confirmed': False},
'doc3': {'confirmed': True}
}
result = get_confirmed_docs_count()
assert result == 2
@patch('streamlit.session_state')
def test_add_chat_message(self, mock_session):
"""Test aggiunta messaggio chat"""
mock_session.chat_history = []
add_chat_message("user", "Test message")
assert len(mock_session.chat_history) == 1
assert mock_session.chat_history[0]["role"] == "user"
assert mock_session.chat_history[0]["content"] == "Test message"
@patch('streamlit.session_state')
def test_add_crewai_result(self, mock_session):
"""Test aggiunta risultato CrewAI"""
mock_session.crewai_history = []
add_crewai_result("test query", "comprehensive", "test result", ["agent1"])
assert len(mock_session.crewai_history) == 1
result = mock_session.crewai_history[0]
assert result["query"] == "test query"
assert result["analysis_type"] == "comprehensive"
assert result["result"] == "test result"
assert result["agents_used"] == ["agent1"]
assert "timestamp" in result
class TestSystemStats:
"""Test statistiche sistema"""
@patch('streamlit.session_state')
def test_get_system_stats_empty(self, mock_session):
"""Test statistiche sistema vuoto"""
mock_session.get.return_value = {}
stats = get_system_stats()
assert stats['uploaded_files'] == 0
assert stats['anonymized_docs'] == 0
assert stats['confirmed_docs'] == 0
assert stats['processed_docs'] == 0
assert stats['chat_messages'] == 0
assert stats['crewai_analyses'] == 0
assert stats['vector_store_ready'] == False
@patch('streamlit.session_state')
def test_get_system_stats_populated(self, mock_session):
"""Test statistiche sistema con dati"""
def mock_get(key, default=None):
data = {
'uploaded_files': {'file1': {}, 'file2': {}},
'anonymized_docs': {
'file1': {'confirmed': True},
'file2': {'confirmed': False}
},
'processed_docs': {'file1': {}},
'chat_history': [{'role': 'user'}, {'role': 'assistant'}],
'crewai_history': [{'query': 'test'}],
'vector_store_built': True
}
return data.get(key, default)
mock_session.get.side_effect = mock_get
with patch('utils.get_confirmed_docs_count', return_value=1):
stats = get_system_stats()
assert stats['uploaded_files'] == 2
assert stats['anonymized_docs'] == 2
assert stats['confirmed_docs'] == 1
assert stats['processed_docs'] == 1
assert stats['chat_messages'] == 2
assert stats['crewai_analyses'] == 1
assert stats['vector_store_ready'] == True
class TestFileOperations:
"""Test operazioni file"""
def test_temp_file_creation_and_cleanup(self, temp_test_file):
"""Test creazione e cleanup file temporaneo"""
# File dovrebbe esistere
assert os.path.exists(temp_test_file)
# Contenuto dovrebbe essere corretto
with open(temp_test_file, 'r') as f:
content = f.read()
assert content == "Test content for file operations"
# Dopo il test, il file viene automaticamente rimosso dal fixture
class TestDataProcessing:
"""Test elaborazione dati"""
def test_json_serialization_complex_data(self):
"""Test serializzazione dati complessi"""
complex_data = {
"string": "test",
"number": 123,
"float": 45.67,
"boolean": True,
"null": None,
"list": [1, 2, 3],
"nested": {
"inner": "value"
},
"datetime": datetime.now()
}
result = export_results_json(complex_data, "complex")
# Dovrebbe serializzare senza errori
parsed = json.loads(result)
assert parsed["string"] == "test"
assert parsed["number"] == 123
assert parsed["float"] == 45.67
assert parsed["boolean"] == True
assert parsed["null"] is None
assert parsed["list"] == [1, 2, 3]
assert parsed["nested"]["inner"] == "value"
assert "datetime" in parsed # Convertito in stringa
class TestErrorHandling:
"""Test gestione errori"""
@patch('streamlit.session_state', side_effect=Exception("Session state error"))
def test_get_confirmed_docs_count_exception(self):
"""Test gestione eccezione in conteggio documenti"""
# Dovrebbe gestire l'eccezione e tornare 0
result = get_confirmed_docs_count()
assert result == 0
def test_export_results_with_non_serializable(self):
"""Test export con oggetti non serializzabili"""
class NonSerializable:
pass
data = {"object": NonSerializable()}
# Dovrebbe gestire oggetti non serializzabili
result = export_results_json(data, "test")
parsed = json.loads(result)
# L'oggetto dovrebbe essere convertito in stringa
assert "object" in parsed
class TestValidationHelpers:
"""Test helper di validazione"""
def test_validate_file_upload_edge_cases(self):
"""Test casi limite validazione file"""
# File con nome vuoto
mock_file = Mock()
mock_file.name = ""
mock_file.size = 1024
with patch('streamlit.error'):
result = validate_file_upload(mock_file)
assert result == False
# File esattamente al limite (10MB)
mock_file.name = "test.txt"
mock_file.size = 10 * 1024 * 1024
result = validate_file_upload(mock_file)
assert result == True
# File con estensione maiuscola
mock_file.name = "test.TXT"
mock_file.size = 1024
result = validate_file_upload(mock_file)
assert result == True
class TestIntegrationHelpers:
"""Test helper per integrazione"""
@patch('streamlit.session_state')
def test_session_state_integration(self, mock_session):
"""Test integrazione con session state"""
# Simula stato iniziale
mock_session.chat_history = []
mock_session.crewai_history = []
# Aggiungi dati
add_chat_message("user", "Hello")
add_chat_message("assistant", "Hi there")
add_crewai_result("test query", "sentiment", "positive result")
# Verifica stato finale
assert len(mock_session.chat_history) == 2
assert len(mock_session.crewai_history) == 1
# Verifica contenuti
assert mock_session.chat_history[0]["role"] == "user"
assert mock_session.chat_history[1]["role"] == "assistant"
assert mock_session.crewai_history[0]["analysis_type"] == "sentiment" |