"""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 # 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.""" self.case_manager.create_case( case_id='CASE-001', case_title='Test Case', investigator='Detective Smith', classification='Homicide', summary='Test summary' ) # Verify case was created case = self.case_manager.get_case('CASE-001') self.assertIsNotNone(case) self.assertEqual(case['case_id'], 'CASE-001') self.assertEqual(case['case_title'], 'Test Case') self.assertEqual(case['investigator'], 'Detective Smith') self.assertEqual(case['classification'], 'Homicide') self.assertEqual(case['summary'], 'Test summary') self.assertEqual(case['status'], 'active') def test_create_case_minimal_fields(self): """Test creating a case with only required fields.""" self.case_manager.create_case( case_id='CASE-002', case_title='Minimal Case', investigator='Detective Jones' ) case = self.case_manager.get_case('CASE-002') self.assertIsNotNone(case) self.assertEqual(case['case_id'], 'CASE-002') self.assertEqual(case['case_title'], 'Minimal Case') self.assertEqual(case['investigator'], 'Detective Jones') self.assertIsNone(case['classification']) self.assertIsNone(case['summary']) def test_get_case_nonexistent(self): """Test getting a non-existent case returns None.""" case = self.case_manager.get_case('NONEXISTENT') 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 self.case_manager.create_case('CASE-001', 'Case One', 'Detective A') self.case_manager.create_case('CASE-002', 'Case Two', 'Detective B') self.case_manager.create_case('CASE-003', '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-001', case_ids) self.assertIn('CASE-002', case_ids) self.assertIn('CASE-003', case_ids) def test_list_cases_filter_by_status(self): """Test listing cases filtered by status.""" self.case_manager.create_case('CASE-001', 'Active Case', 'Detective A') self.case_manager.create_case('CASE-002', 'Another Active', 'Detective B') # Close one case self.case_manager.close_case('CASE-001') # List only active cases active_cases = self.case_manager.list_cases(status='active') self.assertEqual(len(active_cases), 1) self.assertEqual(active_cases[0]['case_id'], 'CASE-002') # 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-001') def test_list_cases_search_term(self): """Test listing cases with search term.""" self.case_manager.create_case('CASE-001', 'Murder Investigation', 'Detective Smith') self.case_manager.create_case('CASE-002', 'Theft Case', 'Detective Jones') self.case_manager.create_case('CASE-003', 'Murder Trial', 'Detective Brown') # Search by title 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) self.assertEqual(results[0]['case_id'], 'CASE-001') # Search by case ID results = self.case_manager.list_cases(search_term='002') self.assertEqual(len(results), 1) self.assertEqual(results[0]['case_id'], 'CASE-002') def test_list_cases_combined_filters(self): """Test listing cases with combined status and search filters.""" self.case_manager.create_case('CASE-001', 'Active Murder', 'Detective A') self.case_manager.create_case('CASE-002', 'Active Theft', 'Detective B') self.case_manager.create_case('CASE-003', 'Closed Murder', 'Detective C') self.case_manager.close_case('CASE-003') # Search for "Murder" in active cases only results = self.case_manager.list_cases(status='active', search_term='Murder') self.assertEqual(len(results), 1) self.assertEqual(results[0]['case_id'], 'CASE-001') def test_update_case(self): """Test updating a case.""" self.case_manager.create_case('CASE-001', 'Original Title', 'Detective A') result = self.case_manager.update_case( 'CASE-001', case_title='Updated Title', classification='Homicide', summary='Updated summary' ) self.assertTrue(result) case = self.case_manager.get_case('CASE-001') self.assertEqual(case['case_title'], 'Updated Title') self.assertEqual(case['classification'], 'Homicide') self.assertEqual(case['summary'], 'Updated summary') self.assertEqual(case['investigator'], 'Detective A') # Unchanged def test_update_case_invalid_fields(self): """Test updating case with invalid fields ignores them.""" self.case_manager.create_case('CASE-001', 'Test Case', 'Detective A') result = self.case_manager.update_case( 'CASE-001', case_title='Updated Title', invalid_field='Should be ignored' ) # Should still work, just ignoring invalid field self.assertTrue(result) case = self.case_manager.get_case('CASE-001') self.assertEqual(case['case_title'], 'Updated Title') self.assertNotIn('invalid_field', case) def test_update_case_no_valid_fields(self): """Test updating case with no valid fields.""" self.case_manager.create_case('CASE-001', 'Test Case', 'Detective A') result = self.case_manager.update_case( 'CASE-001', invalid_field='Should be ignored' ) # Should return None since no valid fields self.assertIsNone(result) def test_update_case_sets_modified_at(self): """Test that updating a case sets modified_at timestamp.""" self.case_manager.create_case('CASE-001', 'Test Case', 'Detective A') case_before = self.case_manager.get_case('CASE-001') created_at = case_before['created_at'] # Small delay to ensure different timestamp import time time.sleep(0.01) self.case_manager.update_case('CASE-001', case_title='Updated') case_after = self.case_manager.get_case('CASE-001') modified_at = case_after['modified_at'] # modified_at should be different from created_at self.assertNotEqual(created_at, modified_at) def test_update_nonexistent_case(self): """Test updating a non-existent case returns False.""" result = self.case_manager.update_case( 'NONEXISTENT', case_title='Updated Title' ) self.assertFalse(result) def test_close_case(self): """Test closing a case.""" self.case_manager.create_case('CASE-001', 'Test Case', 'Detective A') result = self.case_manager.close_case('CASE-001') self.assertTrue(result) case = self.case_manager.get_case('CASE-001') self.assertEqual(case['status'], 'closed') def test_archive_case(self): """Test archiving a case.""" self.case_manager.create_case('CASE-001', 'Test Case', 'Detective A') result = self.case_manager.archive_case('CASE-001') self.assertTrue(result) case = self.case_manager.get_case('CASE-001') self.assertEqual(case['status'], 'archived') def test_delete_case(self): """Test deleting a case.""" self.case_manager.create_case('CASE-001', 'Test Case', 'Detective A') # Verify case exists case = self.case_manager.get_case('CASE-001') self.assertIsNotNone(case) # Delete case result = self.case_manager.delete_case('CASE-001') self.assertTrue(result) # Verify case is gone case = self.case_manager.get_case('CASE-001') self.assertIsNone(case) def test_delete_nonexistent_case(self): """Test deleting a non-existent case returns False.""" result = self.case_manager.delete_case('NONEXISTENT') 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() 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 self.case_manager.create_case( 'CASE-001', 'Investigation Case', 'Detective Smith', 'Robbery', 'Initial summary' ) # Verify creation case = self.case_manager.get_case('CASE-001') self.assertEqual(case['status'], 'active') # Update case self.case_manager.update_case( 'CASE-001', summary='Updated with new findings' ) case = self.case_manager.get_case('CASE-001') self.assertEqual(case['summary'], 'Updated with new findings') # Close case self.case_manager.close_case('CASE-001') case = self.case_manager.get_case('CASE-001') self.assertEqual(case['status'], 'closed') # Archive case self.case_manager.archive_case('CASE-001') case = self.case_manager.get_case('CASE-001') self.assertEqual(case['status'], 'archived') # Delete case self.case_manager.delete_case('CASE-001') case = self.case_manager.get_case('CASE-001') self.assertIsNone(case) def test_multiple_cases_workflow(self): """Test working with multiple cases simultaneously.""" # Create multiple cases for i in range(1, 6): self.case_manager.create_case( f'CASE-{i:03d}', f'Case {i}', f'Detective {chr(64+i)}' # Detective A, B, C, etc. ) # 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-001') self.case_manager.close_case('CASE-003') # Archive one self.case_manager.archive_case('CASE-005') # Check status distribution active_cases = self.case_manager.list_cases(status='active') closed_cases = self.case_manager.list_cases(status='closed') archived_cases = self.case_manager.list_cases(status='archived') self.assertEqual(len(active_cases), 2) # 002, 004 self.assertEqual(len(closed_cases), 2) # 001, 003 self.assertEqual(len(archived_cases), 1) # 005 if __name__ == '__main__': unittest.main()