diff --git a/src/dispatch/case/flows.py b/src/dispatch/case/flows.py index 38381bd26e4c..585479cfb346 100644 --- a/src/dispatch/case/flows.py +++ b/src/dispatch/case/flows.py @@ -42,7 +42,7 @@ send_case_rating_feedback_message, send_case_update_notifications, send_event_paging_message, - send_event_update_prompt_reminder + send_event_update_prompt_reminder, ) from .models import Case from .service import get @@ -768,6 +768,43 @@ def map_case_roles_to_incident_roles( return list(incident_roles) or None +def copy_case_events_to_incident( + case: Case, + incident: Incident, + db_session: Session, +): + """Copies all timeline events from a case to an incident.""" + # Get all events from the case + case_events = event_service.get_by_case_id(db_session=db_session, case_id=case.id).all() + + if not case_events: + log.info(f"No events to copy from case {case.id} to incident {incident.id}") + return + + log.info(f"Copying {len(case_events)} events from case {case.id} to incident {incident.id}") + + for case_event in case_events: + # Create a new event for the incident with the same data + copied_source = f"{case_event.source} (copied from {case.name})" + event_service.log_incident_event( + db_session=db_session, + source=copied_source, + description=case_event.description, + incident_id=incident.id, + individual_id=case_event.individual_id, + started_at=case_event.started_at, + ended_at=case_event.ended_at, + details=case_event.details, + type=case_event.type, + owner=case_event.owner, + pinned=case_event.pinned, + ) + + log.info( + f"Successfully copied {len(case_events)} events from case {case.id} to incident {incident.id}" + ) + + def common_escalate_flow( case: Case, incident: Incident, @@ -793,6 +830,9 @@ def common_escalate_flow( db_session.add(case) db_session.commit() + # Copy timeline events from case to incident + copy_case_events_to_incident(case=case, incident=incident, db_session=db_session) + event_service.log_case_event( db_session=db_session, source="Dispatch Core App", @@ -862,7 +902,10 @@ def common_escalate_flow( event_service.log_case_event( db_session=db_session, source="Dispatch Core App", - description=f"The members of the incident's tactical group {incident.tactical_group.email} have been given permission to access the case's storage folder", + description=( + f"The members of the incident's tactical group {incident.tactical_group.email} " + f"have been given permission to access the case's storage folder" + ), case_id=case.id, ) diff --git a/tests/case/test_case_service.py b/tests/case/test_case_service.py index f3c0054d8cbc..34b001c7a08e 100644 --- a/tests/case/test_case_service.py +++ b/tests/case/test_case_service.py @@ -231,3 +231,54 @@ def test_delete(session, case: Case): t_case = case_get(db_session=session, case_id=case_id) assert not t_case + + +def test_copy_case_events_to_incident(session, case: Case, incident): + """Test that case events are properly copied to incident when escalating.""" + from dispatch.case.flows import copy_case_events_to_incident + from dispatch.event.service import get_by_case_id, get_by_incident_id, log_case_event + + # Create some test events for the case + log_case_event( + db_session=session, + source="Test Source", + description="Test case event 1", + case_id=case.id, + ) + + log_case_event( + db_session=session, + source="Test Source 2", + description="Test case event 2", + case_id=case.id, + ) + + # Verify case has events + case_events = get_by_case_id(db_session=session, case_id=case.id).all() + initial_case_event_count = len(case_events) + assert initial_case_event_count == 2 + + # Verify incident has no events initially + incident_events = get_by_incident_id(db_session=session, incident_id=incident.id).all() + initial_incident_event_count = len(incident_events) + assert initial_incident_event_count == 0 + + # Copy events from case to incident + copy_case_events_to_incident(case=case, incident=incident, db_session=session) + + # Verify incident now has the copied events + incident_events_after = get_by_incident_id(db_session=session, incident_id=incident.id).all() + final_incident_event_count = len(incident_events_after) + assert final_incident_event_count == initial_case_event_count + + # Verify case events are unchanged + case_events_after = get_by_case_id(db_session=session, case_id=case.id).all() + final_case_event_count = len(case_events_after) + assert final_case_event_count == initial_case_event_count + + # Verify the copied events have the correct source + for event in incident_events_after: + assert ( + event.source == f"Test Source (copied from {case.name})" + or event.source == f"Test Source 2 (copied from {case.name})" + )