tests: AutomationProgressPanel test

2026-05-01 16:20:58 +02:00
parent b99635b22e
commit f1b7cf35d9
1 changed files with 106 additions and 0 deletions
@@ -80,3 +80,109 @@ def test_parse_automation_progress_finished_state():
    assert progress.steps[0].step == WorkflowStateKind.FINAL
    assert progress.steps[0].status == StepStatus.SUCCESS
    assert progress.steps[0].message == "Automation complete"
+
+
+def test_process_automation_progress_buffer_waits_for_complete_event():
+    worker = DAQWorker(base_url=None, token="test-token")
+    emitted: list = []
+
+    worker.automation_progress.connect(emitted.append)
+
+    worker._automation_progress_buffer = (
+        'data: {"seq":1,"progress":{"current_step":"Center","steps":[{"step":"mount",'
+        '"status":"success","message":"Mount complete"}],"finished":false,"success":null}}'
+    )
+
+    worker._process_automation_progress_buffer()
+
+    assert emitted == []
+    assert worker._automation_progress_buffer.startswith("data:")
+
+    worker.cleanup()
+
+
+def test_process_automation_progress_buffer_handles_split_sse_event():
+    worker = DAQWorker(base_url=None, token="test-token")
+    emitted: list = []
+
+    worker.automation_progress.connect(emitted.append)
+
+    first_chunk = (
+        'data: {"seq":1,"progress":{"current_step":"Center","steps":[{"step":"mount",'
+    )
+    second_chunk = (
+        '"status":"success","message":"Mount complete"},{"step":"loop_centre",'
+        '"status":"running","message":"Centering sample"}],'
+        '"finished":false,"success":null}}\n\n'
+    )
+
+    worker._automation_progress_buffer += first_chunk
+    worker._process_automation_progress_buffer()
+
+    assert emitted == []
+
+    worker._automation_progress_buffer += second_chunk
+    worker._process_automation_progress_buffer()
+
+    assert len(emitted) == 1
+
+    progress = emitted[0]
+    assert progress.current_step == "Center"
+    assert progress.finished is False
+    assert progress.success is None
+    assert len(progress.steps) == 2
+    assert progress.steps[0].step == WorkflowStateKind.MOUNT
+    assert progress.steps[0].status == StepStatus.SUCCESS
+    assert progress.steps[1].step == WorkflowStateKind.LOOP_CENTRE
+    assert progress.steps[1].status == StepStatus.RUNNING
+    assert worker._automation_progress_buffer == ""
+
+    worker.cleanup()
+
+
+def test_process_automation_progress_buffer_handles_multiple_events():
+    worker = DAQWorker(base_url=None, token="test-token")
+    emitted: list = []
+
+    worker.automation_progress.connect(emitted.append)
+
+    worker._automation_progress_buffer = (
+        'data: {"seq":1,"progress":{"current_step":"Mount","steps":[{"step":"mount",'
+        '"status":"running","message":"Mounting sample"}],"finished":false,"success":null}}\n\n'
+        'data: {"seq":2,"progress":{"current_step":"Paused/Finished","steps":[{"step":"final",'
+        '"status":"success","message":"Automation complete"}],"finished":true,"success":true}}\n\n'
+    )
+
+    worker._process_automation_progress_buffer()
+
+    assert len(emitted) == 2
+
+    first = emitted[0]
+    assert first.current_step == "Mount"
+    assert first.finished is False
+    assert first.steps[0].step == WorkflowStateKind.MOUNT
+    assert first.steps[0].status == StepStatus.RUNNING
+
+    second = emitted[1]
+    assert second.current_step == "Paused/Finished"
+    assert second.finished is True
+    assert second.success is True
+    assert second.steps[0].step == WorkflowStateKind.FINAL
+    assert second.steps[0].status == StepStatus.SUCCESS
+
+    assert worker._automation_progress_buffer == ""
+
+    worker.cleanup()
+
+
+def test_handle_automation_progress_event_ignores_null_progress():
+    worker = DAQWorker(base_url=None, token="test-token")
+    emitted: list = []
+
+    worker.automation_progress.connect(emitted.append)
+
+    worker._handle_automation_progress_event('{"seq":3,"progress":null}')
+
+    assert emitted == []
+
+    worker.cleanup()