396 lines
15 KiB
Python
396 lines
15 KiB
Python
"""Unit tests for the case_manager module."""
|
|
import unittest
|
|
import tempfile
|
|
import os
|
|
import sqlite3
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
from datetime import datetime
|
|
import time
|
|
|
|
# Add the src directory to the path
|
|
import sys
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / 'src'))
|
|
|
|
from forensictrails.core.case_manager import CaseManager
|
|
from forensictrails.db.database import create_fresh_database
|
|
|
|
|
|
class TestCaseManager(unittest.TestCase):
|
|
"""Test cases for CaseManager class."""
|
|
|
|
def setUp(self):
|
|
"""Set up test fixtures."""
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.test_db_path = os.path.join(self.temp_dir, 'test.db')
|
|
self.schema_path = Path(__file__).parent.parent / 'src' / 'forensictrails' / 'db' / 'schema.sql'
|
|
|
|
# Create a fresh database for testing
|
|
create_fresh_database(self.test_db_path, self.schema_path)
|
|
|
|
# Create case manager instance
|
|
self.case_manager = CaseManager(self.test_db_path)
|
|
|
|
def tearDown(self):
|
|
"""Clean up test fixtures."""
|
|
if hasattr(self, 'case_manager') and hasattr(self.case_manager, 'conn'):
|
|
self.case_manager.conn.close()
|
|
|
|
if os.path.exists(self.test_db_path):
|
|
os.remove(self.test_db_path)
|
|
os.rmdir(self.temp_dir)
|
|
|
|
def test_case_manager_initialization(self):
|
|
"""Test CaseManager initializes correctly."""
|
|
self.assertEqual(self.case_manager.db_path, self.test_db_path)
|
|
self.assertIsNotNone(self.case_manager.conn)
|
|
self.assertIsNotNone(self.case_manager.cursor)
|
|
|
|
def test_create_case(self):
|
|
"""Test creating a new case."""
|
|
case_id = self.case_manager.create_case(
|
|
description='Test homicide investigation',
|
|
investigator_name='Detective Smith',
|
|
investigator_role='Lead Detective'
|
|
)
|
|
|
|
self.assertIsNotNone(case_id)
|
|
self.assertIsInstance(case_id, int)
|
|
|
|
# Verify case was created
|
|
case = self.case_manager.get_case(case_id)
|
|
self.assertIsNotNone(case)
|
|
assert case is not None # Type narrowing for Pylance
|
|
self.assertEqual(case['case_id'], case_id)
|
|
self.assertEqual(case['description'], 'Test homicide investigation')
|
|
self.assertEqual(case['investigators'], 'Detective Smith')
|
|
self.assertEqual(case['status'], 'Open')
|
|
|
|
def test_create_case_minimal(self):
|
|
"""Test creating a case with minimal fields."""
|
|
case_id = self.case_manager.create_case(
|
|
description='Minimal case',
|
|
investigator_name='Detective Jones'
|
|
)
|
|
|
|
case = self.case_manager.get_case(case_id)
|
|
self.assertIsNotNone(case)
|
|
assert case is not None # Type narrowing for Pylance
|
|
self.assertEqual(case['description'], 'Minimal case')
|
|
self.assertEqual(case['investigators'], 'Detective Jones')
|
|
self.assertEqual(case['status'], 'Open')
|
|
|
|
def test_create_case_with_status(self):
|
|
"""Test creating a case with specific status."""
|
|
case_id = self.case_manager.create_case(
|
|
description='Archived case',
|
|
investigator_name='Detective Brown',
|
|
status='Archived'
|
|
)
|
|
|
|
case = self.case_manager.get_case(case_id)
|
|
assert case is not None # Type narrowing for Pylance
|
|
self.assertEqual(case['status'], 'Archived')
|
|
|
|
def test_get_case_nonexistent(self):
|
|
"""Test getting a non-existent case returns None."""
|
|
case = self.case_manager.get_case(99999)
|
|
self.assertIsNone(case)
|
|
|
|
def test_list_cases_empty(self):
|
|
"""Test listing cases when database is empty."""
|
|
cases = self.case_manager.list_cases()
|
|
self.assertEqual(cases, [])
|
|
|
|
def test_list_cases_multiple(self):
|
|
"""Test listing multiple cases."""
|
|
# Create multiple cases
|
|
case_id1 = self.case_manager.create_case('Case One', 'Detective A')
|
|
case_id2 = self.case_manager.create_case('Case Two', 'Detective B')
|
|
case_id3 = self.case_manager.create_case('Case Three', 'Detective C')
|
|
|
|
cases = self.case_manager.list_cases()
|
|
|
|
self.assertEqual(len(cases), 3)
|
|
case_ids = [c['case_id'] for c in cases]
|
|
self.assertIn(case_id1, case_ids)
|
|
self.assertIn(case_id2, case_ids)
|
|
self.assertIn(case_id3, case_ids)
|
|
|
|
def test_list_cases_filter_by_status(self):
|
|
"""Test listing cases filtered by status."""
|
|
case_id1 = self.case_manager.create_case('Active Case', 'Detective A')
|
|
case_id2 = self.case_manager.create_case('Another Active', 'Detective B')
|
|
|
|
# Close one case
|
|
self.case_manager.close_case(case_id1)
|
|
|
|
# List only Open cases
|
|
open_cases = self.case_manager.list_cases(status='Open')
|
|
self.assertEqual(len(open_cases), 1)
|
|
self.assertEqual(open_cases[0]['case_id'], case_id2)
|
|
|
|
# List only Closed cases
|
|
closed_cases = self.case_manager.list_cases(status='Closed')
|
|
self.assertEqual(len(closed_cases), 1)
|
|
self.assertEqual(closed_cases[0]['case_id'], case_id1)
|
|
|
|
def test_list_cases_search_term(self):
|
|
"""Test listing cases with search term."""
|
|
self.case_manager.create_case('Murder investigation', 'Detective Smith')
|
|
self.case_manager.create_case('Theft case', 'Detective Jones')
|
|
self.case_manager.create_case('Murder trial', 'Detective Brown')
|
|
|
|
# Search by description
|
|
results = self.case_manager.list_cases(search_term='Murder')
|
|
self.assertEqual(len(results), 2)
|
|
|
|
# Search by investigator
|
|
results = self.case_manager.list_cases(search_term='Smith')
|
|
self.assertEqual(len(results), 1)
|
|
|
|
def test_list_cases_combined_filters(self):
|
|
"""Test listing cases with combined status and search filters."""
|
|
case_id1 = self.case_manager.create_case('Active Murder', 'Detective A')
|
|
case_id2 = self.case_manager.create_case('Active Theft', 'Detective B')
|
|
case_id3 = self.case_manager.create_case('Closed Murder', 'Detective C')
|
|
|
|
self.case_manager.close_case(case_id3)
|
|
|
|
# Search for "Murder" in Open cases only
|
|
results = self.case_manager.list_cases(status='Open', search_term='Murder')
|
|
self.assertEqual(len(results), 1)
|
|
self.assertEqual(results[0]['case_id'], case_id1)
|
|
|
|
def test_update_case_description(self):
|
|
"""Test updating a case description."""
|
|
case_id = self.case_manager.create_case('Original description', 'Detective A')
|
|
|
|
result = self.case_manager.update_case(
|
|
case_id,
|
|
description='Updated description'
|
|
)
|
|
|
|
self.assertTrue(result)
|
|
|
|
case = self.case_manager.get_case(case_id)
|
|
assert case is not None # Type narrowing for Pylance
|
|
self.assertEqual(case['description'], 'Updated description')
|
|
|
|
def test_update_case_status(self):
|
|
"""Test updating a case status."""
|
|
case_id = self.case_manager.create_case('Test case', 'Detective A')
|
|
|
|
result = self.case_manager.update_case(case_id, status='Archived')
|
|
self.assertTrue(result)
|
|
|
|
case = self.case_manager.get_case(case_id)
|
|
assert case is not None # Type narrowing for Pylance
|
|
self.assertEqual(case['status'], 'Archived')
|
|
|
|
def test_update_case_both_fields(self):
|
|
"""Test updating both description and status."""
|
|
case_id = self.case_manager.create_case('Test case', 'Detective A')
|
|
|
|
result = self.case_manager.update_case(
|
|
case_id,
|
|
description='Updated description',
|
|
status='Closed'
|
|
)
|
|
self.assertTrue(result)
|
|
|
|
case = self.case_manager.get_case(case_id)
|
|
assert case is not None # Type narrowing for Pylance
|
|
self.assertEqual(case['description'], 'Updated description')
|
|
self.assertEqual(case['status'], 'Closed')
|
|
|
|
def test_update_case_no_fields(self):
|
|
"""Test updating case with no fields returns False."""
|
|
case_id = self.case_manager.create_case('Test case', 'Detective A')
|
|
|
|
result = self.case_manager.update_case(case_id)
|
|
self.assertFalse(result)
|
|
|
|
def test_update_case_sets_modified_timestamp(self):
|
|
"""Test that updating a case sets modified timestamp."""
|
|
case_id = self.case_manager.create_case('Test case', 'Detective A')
|
|
|
|
case_before = self.case_manager.get_case(case_id)
|
|
assert case_before is not None # Type narrowing for Pylance
|
|
created = case_before['created']
|
|
|
|
time.sleep(1) # Ensure timestamp difference
|
|
|
|
self.case_manager.update_case(case_id, description='Updated')
|
|
|
|
case_after = self.case_manager.get_case(case_id)
|
|
assert case_after is not None # Type narrowing for Pylance
|
|
modified = case_after['modified']
|
|
|
|
# Modified should be different from created
|
|
self.assertNotEqual(created, modified)
|
|
|
|
def test_update_nonexistent_case(self):
|
|
"""Test updating a non-existent case returns False."""
|
|
result = self.case_manager.update_case(99999, description='Updated')
|
|
self.assertFalse(result)
|
|
|
|
def test_close_case(self):
|
|
"""Test closing a case."""
|
|
case_id = self.case_manager.create_case('Test case', 'Detective A')
|
|
|
|
result = self.case_manager.close_case(case_id)
|
|
self.assertTrue(result)
|
|
|
|
case = self.case_manager.get_case(case_id)
|
|
assert case is not None # Type narrowing for Pylance
|
|
self.assertEqual(case['status'], 'Closed')
|
|
self.assertIsNotNone(case['closed'])
|
|
|
|
def test_archive_case(self):
|
|
"""Test archiving a case."""
|
|
case_id = self.case_manager.create_case('Test case', 'Detective A')
|
|
|
|
result = self.case_manager.archive_case(case_id)
|
|
self.assertTrue(result)
|
|
|
|
case = self.case_manager.get_case(case_id)
|
|
assert case is not None # Type narrowing for Pylance
|
|
self.assertEqual(case['status'], 'Archived')
|
|
|
|
def test_delete_case(self):
|
|
"""Test deleting a case."""
|
|
case_id = self.case_manager.create_case('Test case', 'Detective A')
|
|
|
|
# Verify case exists
|
|
case = self.case_manager.get_case(case_id)
|
|
self.assertIsNotNone(case)
|
|
|
|
# Delete case
|
|
result = self.case_manager.delete_case(case_id)
|
|
self.assertTrue(result)
|
|
|
|
# Verify case is gone
|
|
case = self.case_manager.get_case(case_id)
|
|
self.assertIsNone(case)
|
|
|
|
def test_delete_nonexistent_case(self):
|
|
"""Test deleting a non-existent case returns False."""
|
|
result = self.case_manager.delete_case(99999)
|
|
self.assertFalse(result)
|
|
|
|
def test_case_manager_with_default_db_path(self):
|
|
"""Test CaseManager uses config default when no path provided."""
|
|
with patch('forensictrails.core.case_manager.config') as mock_config:
|
|
mock_config.database_path = self.test_db_path
|
|
|
|
cm = CaseManager()
|
|
self.assertEqual(cm.db_path, self.test_db_path)
|
|
cm.conn.close()
|
|
|
|
def test_multiple_investigators_on_case(self):
|
|
"""Test adding multiple investigators to a case."""
|
|
case_id1 = self.case_manager.create_case('Case 1', 'Detective Smith')
|
|
case_id2 = self.case_manager.create_case('Case 2', 'Detective Smith')
|
|
|
|
# Verify investigator is reused
|
|
case1 = self.case_manager.get_case(case_id1)
|
|
case2 = self.case_manager.get_case(case_id2)
|
|
assert case1 is not None and case2 is not None # Type narrowing for Pylance
|
|
|
|
self.assertEqual(case1['investigators'], 'Detective Smith')
|
|
self.assertEqual(case2['investigators'], 'Detective Smith')
|
|
|
|
|
|
class TestCaseManagerIntegration(unittest.TestCase):
|
|
"""Integration tests for CaseManager."""
|
|
|
|
def setUp(self):
|
|
"""Set up test fixtures."""
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.test_db_path = os.path.join(self.temp_dir, 'integration.db')
|
|
self.schema_path = Path(__file__).parent.parent / 'src' / 'forensictrails' / 'db' / 'schema.sql'
|
|
|
|
create_fresh_database(self.test_db_path, self.schema_path)
|
|
self.case_manager = CaseManager(self.test_db_path)
|
|
|
|
def tearDown(self):
|
|
"""Clean up test fixtures."""
|
|
if hasattr(self, 'case_manager') and hasattr(self.case_manager, 'conn'):
|
|
self.case_manager.conn.close()
|
|
|
|
if os.path.exists(self.test_db_path):
|
|
os.remove(self.test_db_path)
|
|
os.rmdir(self.temp_dir)
|
|
|
|
def test_full_case_lifecycle(self):
|
|
"""Test complete case lifecycle: create, update, close, archive, delete."""
|
|
# Create case
|
|
case_id = self.case_manager.create_case(
|
|
'Investigation case',
|
|
'Detective Smith'
|
|
)
|
|
|
|
# Verify creation
|
|
case = self.case_manager.get_case(case_id)
|
|
assert case is not None # Type narrowing for Pylance
|
|
self.assertEqual(case['status'], 'Open')
|
|
|
|
# Update case
|
|
self.case_manager.update_case(case_id, description='Updated findings')
|
|
case = self.case_manager.get_case(case_id)
|
|
assert case is not None # Type narrowing for Pylance
|
|
self.assertEqual(case['description'], 'Updated findings')
|
|
|
|
# Close case
|
|
self.case_manager.close_case(case_id)
|
|
case = self.case_manager.get_case(case_id)
|
|
assert case is not None # Type narrowing for Pylance
|
|
self.assertEqual(case['status'], 'Closed')
|
|
|
|
# Archive case
|
|
self.case_manager.archive_case(case_id)
|
|
case = self.case_manager.get_case(case_id)
|
|
assert case is not None # Type narrowing for Pylance
|
|
self.assertEqual(case['status'], 'Archived')
|
|
|
|
# Delete case
|
|
self.case_manager.delete_case(case_id)
|
|
case = self.case_manager.get_case(case_id)
|
|
self.assertIsNone(case)
|
|
|
|
def test_multiple_cases_workflow(self):
|
|
"""Test working with multiple cases simultaneously."""
|
|
# Create multiple cases
|
|
case_ids = []
|
|
for i in range(1, 6):
|
|
case_id = self.case_manager.create_case(
|
|
f'Case {i}',
|
|
f'Detective {chr(64+i)}' # Detective A, B, C, etc.
|
|
)
|
|
case_ids.append(case_id)
|
|
|
|
# Verify all created
|
|
all_cases = self.case_manager.list_cases()
|
|
self.assertEqual(len(all_cases), 5)
|
|
|
|
# Close some cases
|
|
self.case_manager.close_case(case_ids[0])
|
|
self.case_manager.close_case(case_ids[2])
|
|
|
|
# Archive one
|
|
self.case_manager.archive_case(case_ids[4])
|
|
|
|
# Check status distribution
|
|
open_cases = self.case_manager.list_cases(status='Open')
|
|
closed_cases = self.case_manager.list_cases(status='Closed')
|
|
archived_cases = self.case_manager.list_cases(status='Archived')
|
|
|
|
self.assertEqual(len(open_cases), 2) # case_ids[1], case_ids[3]
|
|
self.assertEqual(len(closed_cases), 2) # case_ids[0], case_ids[2]
|
|
self.assertEqual(len(archived_cases), 1) # case_ids[4]
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|