Coverage for tests/ui/test_pipeline_execution

1import json

2import os

3import pytest

4import time

5from unittest.mock import Mock, patch, MagicMock, call

6from PyQt6.QtWidgets import QMessageBox

7from PyQt6.QtCore import QProcess, QTimer, QCoreApplication

9from main.ui.pipeline_execution_page import PipelineExecutionPage

12@pytest.fixture

13def mock_get_bin_path():

14 """Mock for get_bin_path."""

15 with patch("main.ui.pipeline_execution_page.get_bin_path") as mock:

16 mock.return_value = "/fake/path/pipeline_runner"

17 yield mock

20@pytest.fixture

21def pipeline_config_exec(temp_workspace):

22 """Create a config for pipeline execution."""

23 config = {

24 "sub-01": {

25 "mri": os.path.join(temp_workspace, "sub-01", "anat", "T1w.nii"),

26 "pet": os.path.join(temp_workspace, "sub-01", "pet", "pet.nii"),

27 },

28 "sub-02": {

29 "mri": "path/to/mri2.nii",

30 }

31 }

33 pipeline_dir = os.path.join(temp_workspace, "pipeline")

34 os.makedirs(pipeline_dir, exist_ok=True)

35 config_path = os.path.join(pipeline_dir, "01_config.json")

37 with open(config_path, "w", encoding="utf-8") as f:

38 json.dump(config, f, indent=4)

40 return config_path, config

43@pytest.fixture

44def mock_tree_view():

45 """Mock for tree_view."""

46 tree_view = Mock()

47 tree_view._open_in_explorer = Mock()

48 return tree_view

51@pytest.fixture

52def mock_context_exec(temp_workspace, signal_emitter, mock_tree_view):

53 """Context for PipelineExecutionPage."""

54 context = {

55 "language_changed": signal_emitter.language_changed,

56 "workspace_path": temp_workspace,

57 "update_main_buttons": Mock(),

58 "return_to_import": Mock(),

59 "tree_view": mock_tree_view,

60 "history": []

61 }

62 return context

65class TestPipelineExecutionPageInitialization:

66 """Tests for page initialization."""

68 def test_initialization_basic(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

69 """Test basic initialization."""

70 config_path, config = pipeline_config_exec

71 page = PipelineExecutionPage(mock_context_exec)

72 qtbot.addWidget(page)

74 assert page.context == mock_context_exec

75 assert page.workspace_path == mock_context_exec["workspace_path"]

76 assert page.previous_page is None

77 assert page.next_page is None

78 assert page.pipeline_process is None

79 assert page.pipeline_completed is False

80 assert page.pipeline_error is None

81 assert page.config_path == config_path

83 def test_initialization_with_previous_page(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

84 """Test initialization with previous page."""

85 previous = Mock()

86 page = PipelineExecutionPage(mock_context_exec, previous_page=previous)

87 qtbot.addWidget(page)

89 assert page.previous_page == previous

91 def test_pipeline_output_dir_created(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

92 """Test that the output directory is created."""

93 page = PipelineExecutionPage(mock_context_exec)

94 qtbot.addWidget(page)

96 assert os.path.exists(page.pipeline_output_dir)

97 assert "01_output" in page.pipeline_output_dir

99 def test_ui_elements_created(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

100 """Test that all UI elements are created."""

101 page = PipelineExecutionPage(mock_context_exec)

102 qtbot.addWidget(page)

103

104 assert page.header is not None

105 assert page.current_operation is not None

106 assert page.progress_bar is not None

107 assert page.log_text is not None

108 assert page.log_label is not None

109 assert page.stop_button is not None

110 assert isinstance(page.folder_cards, dict)

111

112 def test_folder_cards_created(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

113 """Test folder cards creation for each patient."""

114 page = PipelineExecutionPage(mock_context_exec)

115 qtbot.addWidget(page)

116

117 page.on_enter()

118

119 assert len(page.folder_cards) == 2

120

121 def test_get_bin_path_called(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

122 """Test that get_bin_path is called."""

123 page = PipelineExecutionPage(mock_context_exec)

124 qtbot.addWidget(page)

125

126 mock_get_bin_path.assert_called_once_with("pipeline_runner")

127 assert page.pipeline_bin_path == "/fake/path/pipeline_runner"

128

129 def test_get_bin_path_error(self, qtbot, mock_context_exec, pipeline_config_exec):

130 """Test error handling when get_bin_path fails."""

131 with patch("main.ui.pipeline_execution_page.get_bin_path") as mock:

132 mock.side_effect = FileNotFoundError("Binary not found")

133

134 with pytest.raises(RuntimeError):

135 page = PipelineExecutionPage(mock_context_exec)

136

137

138class TestConfigFileDetection:

139 """Tests for config file detection."""

140

141 def test_find_latest_config_single(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

142 """Test detection with a single config."""

143 config_path, _ = pipeline_config_exec

144 page = PipelineExecutionPage(mock_context_exec)

145 qtbot.addWidget(page)

146

147 assert page.config_path == config_path

148

149 def test_find_latest_config_multiple(self, qtbot, mock_context_exec, temp_workspace, mock_get_bin_path):

150 """Test detection with multiple configs (should take the last one)."""

151 pipeline_dir = os.path.join(temp_workspace, "pipeline")

152 os.makedirs(pipeline_dir, exist_ok=True)

153

154 # Create multiple configs

155 configs = []

156 for i in [1, 2, 5, 3]: # Non-sequential order

157 config_path = os.path.join(pipeline_dir, f"{i:02d}_config.json")

158 config = {f"sub-{i:02d}": {"mri": "path.nii"}}

159 with open(config_path, "w") as f:

160 json.dump(config, f)

161 configs.append((i, config_path))

162

163 page = PipelineExecutionPage(mock_context_exec)

164 qtbot.addWidget(page)

165

166 # Should take 05_config.json

167 assert "05_config.json" in page.config_path

168

169 def test_find_latest_config_no_files(self, qtbot, mock_context_exec, temp_workspace, mock_get_bin_path):

170 """Test when there are no config files."""

171

172 pipeline_dir = os.path.join(temp_workspace, "pipeline")

173 os.makedirs(pipeline_dir, exist_ok=True)

174

175 with patch.object(PipelineExecutionPage, "get_sub_list", return_value=[]):

176 page = PipelineExecutionPage(mock_context_exec)

177 qtbot.addWidget(page)

178

179 expected = os.path.join(pipeline_dir, "pipeline_config.json")

180 assert page.config_path == expected

181

182

183class TestGetSubList:

184 """Tests for the get_sub_list method."""

185

186 def test_get_sub_list_basic(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

187 """Test patient list extraction."""

188 config_path, config = pipeline_config_exec

189 page = PipelineExecutionPage(mock_context_exec)

190 qtbot.addWidget(page)

191

192 sub_list = page.get_sub_list(config_path)

193

194 assert len(sub_list) == 2

195 assert "sub-01" in sub_list

196 assert "sub-02" in sub_list

197

198 def test_get_sub_list_filters_non_sub(self, qtbot, mock_context_exec, temp_workspace, mock_get_bin_path):

199 """Test that it correctly filters only sub-XX."""

200 pipeline_dir = os.path.join(temp_workspace, "pipeline")

201 config_path = os.path.join(pipeline_dir, "01_config.json")

202

203 config = {

204 "sub-01": {"mri": "path1.nii"},

205 "sub-02": {"mri": "path2.nii"},

206 "metadata": {"version": "1.0"},

207 "settings": {"param": "value"}

208 }

209

210 with open(config_path, "w") as f:

211 json.dump(config, f)

212

213 page = PipelineExecutionPage(mock_context_exec)

214 qtbot.addWidget(page)

215

216 sub_list = page.get_sub_list(config_path)

217

218 assert len(sub_list) == 2

219 assert "metadata" not in sub_list

220 assert "settings" not in sub_list

221

222

223class TestPipelineExecution:

224 """Tests for pipeline execution."""

225

226 def test_start_pipeline_basic(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

227 """Test basic pipeline start."""

228 page = PipelineExecutionPage(mock_context_exec)

229 qtbot.addWidget(page)

230

231 with patch.object(QProcess, 'start') as mock_start, \

232 patch.object(QProcess, 'waitForStarted', return_value=True):

233 page._start_pipeline()

234

235 assert page.pipeline_process is not None

236 mock_start.assert_called_once()

237 assert page.stop_button.isEnabled()

238

239 def test_start_pipeline_already_running(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

240 """Test that it does not start the pipeline if already running."""

241 page = PipelineExecutionPage(mock_context_exec)

242 qtbot.addWidget(page)

243

244 # Simulate process already running

245 page.pipeline_process = Mock()

246

247 with patch.object(QProcess, 'start') as mock_start:

248 page._start_pipeline()

249

250 mock_start.assert_not_called()

251

252 def test_start_pipeline_arguments(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

253 """Test that correct parameters are passed."""

254 config_path, _ = pipeline_config_exec

255 page = PipelineExecutionPage(mock_context_exec)

256 qtbot.addWidget(page)

257

258 with patch.object(QProcess, 'start') as mock_start, \

259 patch.object(QProcess, 'waitForStarted', return_value=True):

260 page._start_pipeline()

261

262 # Verify passed arguments

263 call_args = mock_start.call_args

264 assert '--config' in call_args[0][1]

265 assert config_path in call_args[0][1]

266 assert '--work-dir' in call_args[0][1]

267 assert '--out-dir' in call_args[0][1]

268

269 def test_start_pipeline_failed_to_start(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

270 """Test process start error handling."""

271 page = PipelineExecutionPage(mock_context_exec)

272 qtbot.addWidget(page)

273

274 with patch.object(QProcess, 'start'), \

275 patch.object(QProcess, 'waitForStarted', return_value=False), \

276 patch.object(page, '_on_pipeline_error') as mock_error:

277 page._start_pipeline()

278

279 mock_error.assert_called_once()

280

281

282class TestProcessOutput:

283 """Tests for process output handling."""

284

285 def test_process_pipeline_output_log(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

286 """Test processing LOG line."""

287 page = PipelineExecutionPage(mock_context_exec)

288 qtbot.addWidget(page)

289

290 with patch.object(page, '_log_message') as mock_log, \

291 patch.object(page, '_update_current_operation') as mock_update:

292 page._process_pipeline_output("LOG: Processing started")

293

294 mock_log.assert_called_once_with("Processing started")

295 mock_update.assert_called_once_with("Processing started")

296

297 def test_process_pipeline_output_error(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

298 """Test processing ERROR line."""

299 page = PipelineExecutionPage(mock_context_exec)

300 qtbot.addWidget(page)

301

302 with patch.object(page, '_log_message') as mock_log:

303 page._process_pipeline_output("ERROR: File not found")

304

305 mock_log.assert_called_once()

306 assert "ERROR:" in mock_log.call_args[0][0]

307

308 def test_process_pipeline_output_progress(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

309 """Test processing PROGRESS line."""

310 page = PipelineExecutionPage(mock_context_exec)

311 qtbot.addWidget(page)

312

313 with patch.object(page, '_update_progress') as mock_progress:

314 page._process_pipeline_output("PROGRESS: 5/10")

315

316 mock_progress.assert_called_once_with("5/10")

317

318 def test_process_pipeline_output_finished(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

319 """Test processing FINISHED line."""

320 page = PipelineExecutionPage(mock_context_exec)

321 qtbot.addWidget(page)

322

323 with patch.object(page, '_log_message') as mock_log:

324 page._process_pipeline_output("FINISHED: All done")

325

326 mock_log.assert_called_once()

327

328 def test_process_pipeline_output_generic(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

329 """Test processing generic line."""

330 page = PipelineExecutionPage(mock_context_exec)

331 qtbot.addWidget(page)

332

333 with patch.object(page, '_log_message') as mock_log:

334 page._process_pipeline_output("Some generic output")

335

336 mock_log.assert_called_once_with("Some generic output")

337

338 def test_on_stdout_ready(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

339 """Test stdout reading."""

340 page = PipelineExecutionPage(mock_context_exec)

341 qtbot.addWidget(page)

342

343 mock_process = Mock()

344 mock_data = Mock()

345 mock_data.data.return_value.decode.return_value = "LOG: Test message\nLOG: Another message"

346 mock_process.readAllStandardOutput.return_value = mock_data

347 page.pipeline_process = mock_process

348

349 with patch.object(page, '_process_pipeline_output') as mock_process_output:

350 page._on_stdout_ready()

351

352 assert mock_process_output.call_count == 2

353

354 def test_on_stderr_ready(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

355 """Test stderr reading."""

356 page = PipelineExecutionPage(mock_context_exec)

357 qtbot.addWidget(page)

358

359 mock_process = Mock()

360 mock_data = Mock()

361 mock_data.data.return_value.decode.return_value = "Error message"

362 mock_process.readAllStandardError.return_value = mock_data

363 page.pipeline_process = mock_process

364

365 with patch.object(page, '_log_message') as mock_log:

366 page._on_stderr_ready()

367

368 mock_log.assert_called_once()

369 assert "STDERR:" in mock_log.call_args[0][0]

370

371

372class TestProgressUpdates:

373 """Tests for progress updates."""

374

375 def test_update_progress_valid(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

376 """Test valid progress update."""

377 page = PipelineExecutionPage(mock_context_exec)

378 qtbot.addWidget(page)

379

380 with patch.object(page, 'check_new_files') as mock_check:

381 page._update_progress("5/10")

382

383 assert page.progress_bar.value == 50

384 mock_check.assert_called_once()

385

386 def test_update_progress_invalid(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path,

387 mock_logger):

388 """Test progress update with invalid format."""

389 page = PipelineExecutionPage(mock_context_exec)

390 qtbot.addWidget(page)

391

392 initial_value = page.progress_bar.value

393

394 with patch("main.ui.pipeline_execution_page.log", mock_logger):

395 page._update_progress("invalid")

396 # The value should not change

397 assert page.progress_bar.value == initial_value

398

399 def test_update_current_operation_starting(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

400 """Test operation update: starting."""

401 page = PipelineExecutionPage(mock_context_exec)

402 qtbot.addWidget(page)

403

404 page._update_current_operation("Starting pipeline execution")

405

406 assert "Initializing" in page.current_operation.text()

407

408 def test_update_current_operation_patient(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

409 """Test operation update: processing patient."""

410 page = PipelineExecutionPage(mock_context_exec)

411 qtbot.addWidget(page)

412

413 page._update_current_operation("Processing patient sub-01")

414

415 assert "sub-01" in page.current_operation.text()

416

417 def test_update_current_operation_analysis(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

418 """Test operation update: analysis."""

419 page = PipelineExecutionPage(mock_context_exec)

420 qtbot.addWidget(page)

421

422 page._update_current_operation("Performing statistical analysis")

423

424 assert "analysis" in page.current_operation.text().lower()

425

426 def test_update_current_operation_saving(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

427 """Test operation update: saving."""

428 page = PipelineExecutionPage(mock_context_exec)

429 qtbot.addWidget(page)

430

431 page._update_current_operation("Saving results to CSV")

432

433 assert "Saving" in page.current_operation.text()

434

435 def test_extract_patient_id_from_log(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

436 """Test patient ID extraction from log."""

437 page = PipelineExecutionPage(mock_context_exec)

438 qtbot.addWidget(page)

439

440 patient_id = page._extract_patient_id_from_log("Processing sub-01 data")

441 assert patient_id == "sub-01"

442

443 patient_id = page._extract_patient_id_from_log("sub-123 completed")

444 assert patient_id == "sub-123"

445

446 patient_id = page._extract_patient_id_from_log("No patient here")

447 assert patient_id is None

448

449

450class TestProcessCompletion:

451 """Tests for process completion."""

452

453 def test_on_process_finished_success(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

454 """Test successful completion."""

455 page = PipelineExecutionPage(mock_context_exec)

456 qtbot.addWidget(page)

457

458 with patch.object(page, '_on_pipeline_finished') as mock_finished:

459 page._on_process_finished(0, QProcess.ExitStatus.NormalExit)

460

461 mock_finished.assert_called_once()

462

463 def test_on_process_finished_error(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

464 """Test completion with error."""

465 page = PipelineExecutionPage(mock_context_exec)

466 qtbot.addWidget(page)

467

468 with patch.object(page, '_on_pipeline_error') as mock_error:

469 page._on_process_finished(1, QProcess.ExitStatus.NormalExit)

470

471 mock_error.assert_called_once()

472

473 def test_on_pipeline_finished(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

474 """Test pipeline finished handling."""

475 page = PipelineExecutionPage(mock_context_exec)

476 qtbot.addWidget(page)

477

478 page._on_pipeline_finished()

479

480 assert page.pipeline_completed is True

481 assert page.progress_bar.value == 100

482 assert not page.stop_button.isEnabled()

483 assert page.pipeline_process is None

484 mock_context_exec["update_main_buttons"].assert_called_once()

485

486 def test_on_pipeline_error(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

487 """Test pipeline error handling."""

488 page = PipelineExecutionPage(mock_context_exec)

489 qtbot.addWidget(page)

490

491 page._on_pipeline_error("Test error message")

492

493 assert page.pipeline_error == "Test error message"

494 assert not page.stop_button.isEnabled()

495 assert page.pipeline_process is None

496 mock_context_exec["update_main_buttons"].assert_called_once()

497

498 def test_on_process_error(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

499 """Test handling of different process errors."""

500 page = PipelineExecutionPage(mock_context_exec)

501 qtbot.addWidget(page)

502

503 with patch.object(page, '_on_pipeline_error') as mock_error:

504 page._on_process_error(QProcess.ProcessError.FailedToStart)

505 mock_error.assert_called_once()

506

507 with patch.object(page, '_on_pipeline_error') as mock_error:

508 page._on_process_error(QProcess.ProcessError.Crashed)

509 mock_error.assert_called_once()

510

511

512class TestStopButton:

513 """Tests for the Stop button."""

514

515 def test_stop_button_running(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

516 """Test stop while process is running."""

517 page = PipelineExecutionPage(mock_context_exec)

518 qtbot.addWidget(page)

519

520 mock_process = Mock()

521 mock_process.state.return_value = QProcess.ProcessState.Running

522 mock_process.waitForFinished.return_value = True

523 page.pipeline_process = mock_process

524

525 page._on_stop_clicked()

526

527 mock_process.terminate.assert_called_once()

528 assert page.pipeline_process is None

529 assert not page.stop_button.isEnabled()

530 mock_context_exec["update_main_buttons"].assert_called_once()

531

532 def test_stop_button_forced_kill(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

533 """Test stop with forced kill if terminate fails."""

534 page = PipelineExecutionPage(mock_context_exec)

535 qtbot.addWidget(page)

536

537 mock_process = Mock()

538 mock_process.state.return_value = QProcess.ProcessState.Running

539 mock_process.waitForFinished.side_effect = [False, True] # First False, then True

540 page.pipeline_process = mock_process

541

542 page._on_stop_clicked()

543

544 mock_process.terminate.assert_called_once()

545 mock_process.kill.assert_called_once()

546

547 def test_stop_button_not_running(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

548 """Test stop when process is not running."""

549 page = PipelineExecutionPage(mock_context_exec)

550 qtbot.addWidget(page)

551

552 mock_process = Mock()

553 mock_process.state.return_value = QProcess.ProcessState.NotRunning

554 page.pipeline_process = mock_process

555

556 page._on_stop_clicked()

557

558 mock_process.terminate.assert_not_called()

559

560

561class TestLogMessages:

562 """Tests for logging."""

563

564 def test_log_message_basic(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

565 """Test basic log message."""

566 page = PipelineExecutionPage(mock_context_exec)

567 qtbot.addWidget(page)

568

569 initial_text = page.log_text.toPlainText()

570

571 page._log_message("Test message")

572

573 log_content = page.log_text.toPlainText()

574 assert "Test message" in log_content

575 assert "[" in log_content # Timestamp

576

577 def test_log_message_multiple(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

578 """Test multiple messages."""

579 page = PipelineExecutionPage(mock_context_exec)

580 qtbot.addWidget(page)

581

582 messages = ["First message", "Second message", "Third message"]

583 for msg in messages:

584 page._log_message(msg)

585

586 log_content = page.log_text.toPlainText()

587 for msg in messages:

588 assert msg in log_content

589

590 def test_log_message_autoscroll(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

591 """Test autoscroll to the bottom of the log."""

592 page = PipelineExecutionPage(mock_context_exec)

593 qtbot.addWidget(page)

594

595 # Add many messages to force scrolling

596 for i in range(50):

597 page._log_message(f"Message {i}")

598

599 scrollbar = page.log_text.verticalScrollBar()

600 # Should be scrolled to the bottom

601 assert scrollbar.value() == scrollbar.maximum()

602

603

604class TestNavigation:

605 """Tests for navigation."""

606

607 def test_on_enter_starts_pipeline(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

608 """Test that on_enter automatically starts the pipeline."""

609 page = PipelineExecutionPage(mock_context_exec)

610 qtbot.addWidget(page)

611

612 with patch.object(page, '_start_pipeline') as mock_start:

613 page.on_enter()

614

615 # Wait for QTimer to call the method

616 qtbot.wait(600)

617

618 mock_start.assert_called_once()

619

620 def test_on_enter_not_started_twice(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

621 """Test that on_enter does not start twice if already completed."""

622 page = PipelineExecutionPage(mock_context_exec)

623 qtbot.addWidget(page)

624

625 page.pipeline_completed = True

626

627 with patch.object(page, '_start_pipeline') as mock_start:

628 page.on_enter()

629 qtbot.wait(600)

630

631 mock_start.assert_not_called()

632

633 def test_back_calls_reset(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

634 """Test that back calls reset_page."""

635 page = PipelineExecutionPage(mock_context_exec)

636 qtbot.addWidget(page)

637

638 previous = Mock()

639 page.previous_page = previous

640

641 with patch.object(page, 'reset_page') as mock_reset:

642 result = page.back()

643

644 mock_reset.assert_called_once()

645 assert result == previous

646

647 def test_back_no_previous_page(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

648 """Test back without previous page."""

649 page = PipelineExecutionPage(mock_context_exec)

650 qtbot.addWidget(page)

651

652 result = page.back()

653

654 assert result is None

655

656 def test_next_with_confirmation(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

657 """Test next with user confirmation."""

658 page = PipelineExecutionPage(mock_context_exec)

659 qtbot.addWidget(page)

660

661 with patch.object(QMessageBox, 'exec', return_value=QMessageBox.StandardButton.Ok):

662 result = page.next(mock_context_exec)

663

664 mock_context_exec["return_to_import"].assert_called()

665

666 def test_next_cancelled(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

667 """Test next with user cancellation."""

668 page = PipelineExecutionPage(mock_context_exec)

669 qtbot.addWidget(page)

670

671 with patch.object(QMessageBox, 'exec', return_value=QMessageBox.StandardButton.Cancel):

672 result = page.next(mock_context_exec)

673

674 assert result is None

675 mock_context_exec["return_to_import"].assert_not_called()

676

677 def test_is_ready_to_go_back_completed(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

678 """Test is_ready_to_go_back when completed."""

679 page = PipelineExecutionPage(mock_context_exec)

680 qtbot.addWidget(page)

681

682 page.pipeline_completed = True

683 assert page.is_ready_to_go_back() is True

684

685 def test_is_ready_to_go_back_error(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

686 """Test is_ready_to_go_back when there is an error."""

687 page = PipelineExecutionPage(mock_context_exec)

688 qtbot.addWidget(page)

689

690 page.pipeline_error = "Some error"

691 assert page.is_ready_to_go_back() is True

692

693 def test_is_ready_to_go_back_running(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

694 """Test is_ready_to_go_back during execution."""

695 page = PipelineExecutionPage(mock_context_exec)

696 qtbot.addWidget(page)

697

698 assert page.is_ready_to_go_back() is False

699

700 def test_is_ready_to_advance(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

701 """Test is_ready_to_advance."""

702 page = PipelineExecutionPage(mock_context_exec)

703 qtbot.addWidget(page)

704

705 assert page.is_ready_to_advance() is False

706

707 page.pipeline_completed = True

708 assert page.is_ready_to_advance() is True

709

710 page.pipeline_completed = False

711 page.pipeline_error = "Error"

712 assert page.is_ready_to_advance() is True

713

714

715class TestResetPage:

716 """Tests for page reset."""

717

718 def test_reset_page_basic(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

719 """Test basic page reset."""

720 page = PipelineExecutionPage(mock_context_exec)

721 qtbot.addWidget(page)

722

723 # Modify state

724 page.progress_bar.setValue(50)

725 page.pipeline_error = "Some error"

726 page.pipeline_completed = True

727 page.pipeline_process = Mock()

728 page.log_text.setText("Some logs")

729

730 page.reset_page()

731

732 assert page.progress_bar.value == 0

733 assert page.pipeline_error is None

734 assert page.pipeline_completed is False

735 assert page.pipeline_process is None

736 assert page.log_text.toPlainText() == ""

737 assert not page.stop_button.isEnabled()

738

739 def test_reset_page_color(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

740 """Test that reset restores the progress bar color."""

741 page = PipelineExecutionPage(mock_context_exec)

742 qtbot.addWidget(page)

743

744 page.progress_bar.setColor("#FF0000")

745

746 page.reset_page()

747

748 # Verify it returned to default color

749 assert page.progress_bar.color.name() == "#3498db"

750

751 def test_reset_page_current_operation(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

752 """Test that reset restores the current operation text."""

753 page = PipelineExecutionPage(mock_context_exec)

754 qtbot.addWidget(page)

755

756 page.current_operation.setText("Processing...")

757

758 page.reset_page()

759

760 assert "Preparing to start" in page.current_operation.text()

761

762

763class TestCheckNewFiles:

764 """Tests for check_new_files."""

765

766 def test_check_new_files(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

767 """Test that check_new_files calls the method on all cards."""

768 page = PipelineExecutionPage(mock_context_exec)

769 qtbot.addWidget(page)

770

771 # Mock folder cards

772 for card in page.folder_cards.values():

773 card.check_new_files = Mock()

774

775 page.check_new_files()

776

777 # Verify it was called on all cards

778 for card in page.folder_cards.values():

779 card.check_new_files.assert_called_once()

780

781

782class TestDestructor:

783 """Tests for the destructor."""

784

785 def test_destructor_kills_process(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

786 """Test that the destructor kills the process if running."""

787 page = PipelineExecutionPage(mock_context_exec)

788 qtbot.addWidget(page)

789

790 mock_process = Mock()

791 mock_process.state.return_value = QProcess.ProcessState.Running

792 mock_process.waitForFinished.return_value = True

793 page.pipeline_process = mock_process

794

795 # Simulate destruction

796 page.__del__()

797

798 mock_process.kill.assert_called_once()

799 mock_process.waitForFinished.assert_called_once()

800

801 def test_destructor_no_process(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

802 """Test that the destructor does not crash if there is no process."""

803 page = PipelineExecutionPage(mock_context_exec)

804 qtbot.addWidget(page)

805

806 page.pipeline_process = None

807

808 # Should not crash

809 page.__del__()

810

811

812class TestTranslation:

813 """Tests for translations."""

814

815 def test_translate_ui_called_on_init(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

816 """Test that _translate_ui is called during init."""

817 with patch.object(PipelineExecutionPage, '_translate_ui') as mock_translate:

818 page = PipelineExecutionPage(mock_context_exec)

819 qtbot.addWidget(page)

820

821 mock_translate.assert_called()

822

823 def test_translate_ui_updates_labels(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

824 """Test that _translate_ui updates labels."""

825 page = PipelineExecutionPage(mock_context_exec)

826 qtbot.addWidget(page)

827

828 page._translate_ui()

829

830 assert page.header.text() is not None

831 assert page.stop_button.text() is not None

832 assert page.log_label.text() is not None

833

834 def test_language_changed_signal(self, qtbot, mock_context_exec, signal_emitter, pipeline_config_exec,

835 mock_get_bin_path):

836 """Test that the language_changed signal updates the UI."""

837 page = PipelineExecutionPage(mock_context_exec)

838 qtbot.addWidget(page)

839

840 with patch.object(page, '_translate_ui') as mock_translate:

841 mock_context_exec["language_changed"].connect(mock_translate)

842 mock_context_exec["language_changed"].emit("it")

843 mock_translate.assert_called()

844

845

846class TestFolderCards:

847 """Tests for folder cards."""

848

849 def test_folder_cards_connected_to_tree_view(self, qtbot, mock_context_exec, pipeline_config_exec,

850 mock_get_bin_path, mock_tree_view):

851 """Test that folder cards are connected to tree_view."""

852 page = PipelineExecutionPage(mock_context_exec)

853 qtbot.addWidget(page)

854

855 # Simulate request to open folder

856 for card in page.folder_cards.values():

857 # Difficult to test signal connection directly, but we can verify cards exist

858 assert card is not None

859

860 def test_folder_cards_paths(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

861 """Test that folder cards paths are correct."""

862 page = PipelineExecutionPage(mock_context_exec)

863 qtbot.addWidget(page)

864

865 page.on_enter()

866

867 for watch_dir in page.watch_dirs:

868 assert page.pipeline_output_dir in watch_dir

869 assert watch_dir in page.folder_cards

870

871

872class TestEdgeCases:

873 """Tests for edge cases."""

874

875 def test_empty_config(self, qtbot, mock_context_exec, temp_workspace, mock_get_bin_path):

876 """Test with empty config."""

877 pipeline_dir = os.path.join(temp_workspace, "pipeline")

878 os.makedirs(pipeline_dir, exist_ok=True)

879 config_path = os.path.join(pipeline_dir, "01_config.json")

880

881 with open(config_path, "w") as f:

882 json.dump({}, f)

883

884 page = PipelineExecutionPage(mock_context_exec)

885 qtbot.addWidget(page)

886

887 assert len(page.folder_cards) == 0

888

889 def test_invalid_config_id(self, qtbot, mock_context_exec, temp_workspace, mock_get_bin_path, mock_logger):

890 """Test with invalid config ID."""

891 pipeline_dir = os.path.join(temp_workspace, "pipeline")

892 os.makedirs(pipeline_dir, exist_ok=True)

893 config_path = os.path.join(pipeline_dir, "invalid_config.json")

894

895 with open(config_path, "w") as f:

896 json.dump({"sub-01": {"mri": "path.nii"}}, f)

897

898 with (patch("main.ui.pipeline_execution_page.log", mock_logger),

899 patch.object(PipelineExecutionPage, "get_sub_list", return_value=[])):

900 page = PipelineExecutionPage(mock_context_exec)

901 qtbot.addWidget(page)

902

903 # Should use default

904 assert page.pipeline_output_dir == os.path.join(temp_workspace, "pipeline", "pipeline_output")

905 mock_logger.debug.assert_called()

906

907 def test_process_output_empty_lines(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

908 """Test that empty lines are ignored."""

909 page = PipelineExecutionPage(mock_context_exec)

910 qtbot.addWidget(page)

911

912 mock_process = Mock()

913 mock_data = Mock()

914 mock_data.data.return_value.decode.return_value = "\n\n \nLOG: Valid message\n\n"

915 mock_process.readAllStandardOutput.return_value = mock_data

916 page.pipeline_process = mock_process

917

918 with patch.object(page, '_process_pipeline_output') as mock_process_output:

919 page._on_stdout_ready()

920

921 # Only the valid line should be processed

922 assert mock_process_output.call_count == 1

923 mock_process_output.assert_called_with("LOG: Valid message")

924

925 def test_multiple_progress_updates(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

926 """Test multiple progress updates."""

927 page = PipelineExecutionPage(mock_context_exec)

928 qtbot.addWidget(page)

929

930 with patch.object(page, 'check_new_files'):

931 for i in range(1, 11):

932 page._update_progress(f"{i}/10")

933 assert page.progress_bar.value == i*10

934

935 def test_unicode_in_logs(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

936 """Test with unicode characters in logs."""

937 page = PipelineExecutionPage(mock_context_exec)

938 qtbot.addWidget(page)

939

940 unicode_messages = [

941 "Processing àèéìòù",

942 "Patient données françaises",

943 "Εργασία ελληνικά",

944 "处理中文"

945 ]

946

947 for msg in unicode_messages:

948 page._log_message(msg)

949

950 log_content = page.log_text.toPlainText()

951 for msg in unicode_messages:

952 assert msg in log_content

953

954 def test_very_long_log_messages(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

955 """Test with very long log messages."""

956 page = PipelineExecutionPage(mock_context_exec)

957 qtbot.addWidget(page)

958

959 long_message = "A" * 10000

960 page._log_message(long_message)

961

962 log_content = page.log_text.toPlainText()

963 assert long_message in log_content

964

965 def test_rapid_process_output(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

966 """Test with rapid process output."""

967 page = PipelineExecutionPage(mock_context_exec)

968 qtbot.addWidget(page)

969

970 messages = [f"LOG: Message {i}" for i in range(100)]

971

972 with patch.object(page, '_log_message'):

973 for msg in messages:

974 page._process_pipeline_output(msg)

975

976 assert True

977

978 def test_concurrent_stop_and_finish(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

979 """Test concurrent stop and finish."""

980 page = PipelineExecutionPage(mock_context_exec)

981 qtbot.addWidget(page)

982

983 mock_process = Mock()

984 mock_process.state.return_value = QProcess.ProcessState.Running

985 mock_process.waitForFinished.return_value = True

986 page.pipeline_process = mock_process

987

988 # Call stop

989 page._on_stop_clicked()

990

991 # Then finish (as if the process finished just as we were stopping)

992 page._on_pipeline_finished()

993

994 # Should handle the situation without crashing

995 assert page.pipeline_completed is True

996 assert page.pipeline_process is None

997

998

999class TestReturnToImport:

1000 """Tests for _return_to_import."""

1001

1002 def test_return_to_import_not_running(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

1003 """Test return_to_import when not running."""

1004 page = PipelineExecutionPage(mock_context_exec)

1005 qtbot.addWidget(page)

1006

1007 page._return_to_import()

1008

1009 mock_context_exec["return_to_import"].assert_called_once()

1010

1011 def test_return_to_import_while_running(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

1012 """Test that return_to_import is blocked during execution."""

1013 page = PipelineExecutionPage(mock_context_exec)

1014 qtbot.addWidget(page)

1015

1016 mock_process = Mock()

1017 mock_process.state.return_value = QProcess.ProcessState.Running

1018 page.pipeline_process = mock_process

1019

1020 page._return_to_import()

1021

1022 # Should not call return_to_import

1023 mock_context_exec["return_to_import"].assert_not_called()

1024

1025

1026class TestIntegration:

1027 """Integration tests for complete flows."""

1028

1029 def test_full_pipeline_success_flow(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

1030 """Test full flow: start -> execution -> success."""

1031 page = PipelineExecutionPage(mock_context_exec)

1032 qtbot.addWidget(page)

1033

1034 # Simulate start

1035 with patch.object(QProcess, 'start'), \

1036 patch.object(QProcess, 'waitForStarted', return_value=True):

1037 page._start_pipeline()

1038 assert page.pipeline_process is not None

1039 assert page.stop_button.isEnabled()

1040

1041 # Simulate output

1042 page._process_pipeline_output("LOG: Starting pipeline")

1043 page._process_pipeline_output("PROGRESS: 5/10")

1044 page._process_pipeline_output("LOG: Processing sub-01")

1045 page._process_pipeline_output("PROGRESS: 10/10")

1046

1047 # Simulate completion

1048 page._on_pipeline_finished()

1049

1050 assert page.pipeline_completed is True

1051 assert page.progress_bar.value == 100

1052 assert not page.stop_button.isEnabled()

1053

1054 def test_full_pipeline_error_flow(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

1055 """Test full flow: start -> execution -> error."""

1056 page = PipelineExecutionPage(mock_context_exec)

1057 qtbot.addWidget(page)

1058

1059 # Simulate start

1060 with patch.object(QProcess, 'start'), \

1061 patch.object(QProcess, 'waitForStarted', return_value=True):

1062 page._start_pipeline()

1063

1064 # Simulate output

1065 page._process_pipeline_output("LOG: Starting pipeline")

1066 page._process_pipeline_output("ERROR: File not found")

1067

1068 # Simulate error

1069 page._on_pipeline_error("Fatal error occurred")

1070

1071 assert page.pipeline_error == "Fatal error occurred"

1072 assert not page.stop_button.isEnabled()

1073 assert page.pipeline_completed is False

1074

1075 def test_full_pipeline_stop_flow(self, qtbot, mock_context_exec, pipeline_config_exec, mock_get_bin_path):

1076 """Test full flow: start -> execution -> user stop."""

1077 page = PipelineExecutionPage(mock_context_exec)

1078 qtbot.addWidget(page)

1079

1080 # Simulate start

1081 with patch.object(QProcess, 'start'), \

1082 patch.object(QProcess, 'waitForStarted', return_value=True):

1083 page._start_pipeline()

1084

1085 mock_process = Mock()

1086 mock_process.state.return_value = QProcess.ProcessState.Running

1087 mock_process.waitForFinished.return_value = True

1088 page.pipeline_process = mock_process

1089

1090 # Simulate stop

1091 page._on_stop_clicked()

1092

1093 mock_process.terminate.assert_called_once()

1094 assert page.pipeline_process is None

1095 assert not page.stop_button.isEnabled()

1096 assert page.progress_bar.value == 0

Coverage for tests/ui/test_pipeline_execution_page.py: 99%

637 statements