|
1 | 1 | """Tests to verify the service functionality of the application module.""" |
2 | 2 |
|
3 | 3 | from datetime import UTC, datetime, timedelta |
| 4 | +from pathlib import Path |
4 | 5 | from unittest.mock import MagicMock, patch |
5 | 6 |
|
| 7 | +import pydicom |
6 | 8 | import pytest |
7 | 9 | from typer.testing import CliRunner |
8 | 10 |
|
9 | 11 | from aignostics.application import Service as ApplicationService |
10 | 12 | from aignostics.application._utils import validate_due_date |
11 | 13 | from aignostics.platform import NotFoundException, RunData, RunOutput |
12 | 14 | from tests.constants_test import ( |
13 | | - HETA_APPLICATION_ID, |
14 | | - HETA_APPLICATION_VERSION, |
15 | | - TEST_APPLICATION_VERSION_USE_LATEST_FALLBACK_SKIP, |
16 | | -) |
| 15 | + HETA_APPLICATION_ID, HETA_APPLICATION_VERSION, |
| 16 | + TEST_APPLICATION_VERSION_USE_LATEST_FALLBACK_SKIP) |
17 | 17 |
|
18 | 18 |
|
19 | 19 | @pytest.mark.unit |
@@ -439,3 +439,105 @@ def test_application_run_update_item_custom_metadata_not_found(mock_get_client: |
439 | 439 |
|
440 | 440 | with pytest.raises(NotFoundException, match="not found"): |
441 | 441 | service.application_run_update_item_custom_metadata("run-123", "invalid-item-id", {"key": "value"}) |
| 442 | + |
| 443 | +@pytest.fixture |
| 444 | +def create_dicom(): |
| 445 | + """Fixture that returns a function to create minimal but valid DICOM datasets.""" |
| 446 | + def _create_dicom(series_uid: str, rows: int, cols: int) -> pydicom.Dataset: |
| 447 | + """Create a minimal but valid DICOM dataset. |
| 448 | + |
| 449 | + Args: |
| 450 | + series_uid: The series instance UID |
| 451 | + rows: Number of rows (height) |
| 452 | + cols: Number of columns (width) |
| 453 | + |
| 454 | + Returns: |
| 455 | + A valid pydicom Dataset |
| 456 | + """ |
| 457 | + ds = pydicom.Dataset() |
| 458 | + |
| 459 | + # File Meta Information |
| 460 | + ds.file_meta = pydicom.Dataset() |
| 461 | + ds.file_meta.TransferSyntaxUID = pydicom.uid.ImplicitVRLittleEndian |
| 462 | + ds.file_meta.MediaStorageSOPClassUID = '1.2.840.10008.5.1.4.1.1.77.1.6' # VL Whole Slide Microscopy |
| 463 | + ds.file_meta.MediaStorageSOPInstanceUID = pydicom.uid.generate_uid() |
| 464 | + |
| 465 | + # Required DICOM attributes |
| 466 | + ds.SeriesInstanceUID = series_uid |
| 467 | + ds.SOPInstanceUID = pydicom.uid.generate_uid() |
| 468 | + ds.SOPClassUID = ds.file_meta.MediaStorageSOPClassUID |
| 469 | + ds.StudyInstanceUID = pydicom.uid.generate_uid() |
| 470 | + ds.Modality = 'SM' |
| 471 | + ds.Rows = rows |
| 472 | + ds.Columns = cols |
| 473 | + |
| 474 | + return ds |
| 475 | + |
| 476 | + return _create_dicom |
| 477 | + |
| 478 | + |
| 479 | +@pytest.mark.unit |
| 480 | +def test_filter_dicom_series_files_single_file(tmp_path: Path, create_dicom: Callable[..., Dataset]) -> None: |
| 481 | + """Test that single DICOM files are not filtered.""" |
| 482 | + ds = create_dicom("1.2.3.4.5", 1024, 1024) |
| 483 | + dcm_file = tmp_path / "test.dcm" |
| 484 | + ds.save_as(dcm_file, write_like_original=False) |
| 485 | + |
| 486 | + excluded = ApplicationService._filter_dicom_series_files(tmp_path) |
| 487 | + assert len(excluded) == 0 |
| 488 | + |
| 489 | + |
| 490 | +@pytest.mark.unit |
| 491 | +def test_filter_dicom_series_files_pyramid(tmp_path: Path, create_dicom: Callable[..., Dataset]) -> None: |
| 492 | + """Test that for multi-file DICOM series, only the highest resolution file is kept.""" |
| 493 | + series_uid = "1.2.3.4.5" |
| 494 | + |
| 495 | + # Create low resolution DICOM file |
| 496 | + ds_low = create_dicom(series_uid, 512, 512) |
| 497 | + dcm_file_low = tmp_path / "test_low.dcm" |
| 498 | + ds_low.save_as(dcm_file_low, write_like_original=False) |
| 499 | + |
| 500 | + # Create medium resolution DICOM file |
| 501 | + ds_med = create_dicom(series_uid, 1024, 1024) |
| 502 | + dcm_file_med = tmp_path / "test_med.dcm" |
| 503 | + ds_med.save_as(dcm_file_med, write_like_original=False) |
| 504 | + |
| 505 | + # Create high resolution DICOM file |
| 506 | + ds_high = create_dicom(series_uid, 2048, 2048) |
| 507 | + dcm_file_high = tmp_path / "test_high.dcm" |
| 508 | + ds_high.save_as(dcm_file_high, write_like_original=False) |
| 509 | + |
| 510 | + # Filter the series |
| 511 | + excluded = ApplicationService._filter_dicom_series_files(tmp_path) |
| 512 | + |
| 513 | + # Should exclude 2 files (low and medium), keeping only the highest resolution |
| 514 | + assert len(excluded) == 2 |
| 515 | + assert dcm_file_low in excluded |
| 516 | + assert dcm_file_med in excluded |
| 517 | + assert dcm_file_high not in excluded |
| 518 | + |
| 519 | + |
| 520 | +@pytest.mark.unit |
| 521 | +def test_filter_dicom_series_files_multiple_series(tmp_path: Path, create_dicom: Callable[..., Dataset]) -> None: |
| 522 | + """Test that files from different series are not filtered against each other.""" |
| 523 | + # Series 1 - two files |
| 524 | + ds1_low = create_dicom("1.2.3.4.5", 512, 512) |
| 525 | + dcm_file1_low = tmp_path / "series1_low.dcm" |
| 526 | + ds1_low.save_as(dcm_file1_low, write_like_original=False) |
| 527 | + |
| 528 | + ds1_high = create_dicom("1.2.3.4.5", 1024, 1024) |
| 529 | + dcm_file1_high = tmp_path / "series1_high.dcm" |
| 530 | + ds1_high.save_as(dcm_file1_high, write_like_original=False) |
| 531 | + |
| 532 | + # Series 2 - single file (should not be filtered) |
| 533 | + ds2 = create_dicom("6.7.8.9.0", 512, 512) |
| 534 | + dcm_file2 = tmp_path / "series2.dcm" |
| 535 | + ds2.save_as(dcm_file2, write_like_original=False) |
| 536 | + |
| 537 | + excluded = ApplicationService._filter_dicom_series_files(tmp_path) |
| 538 | + |
| 539 | + # Should exclude only the low-res file from series 1 |
| 540 | + assert len(excluded) == 1 |
| 541 | + assert dcm_file1_low in excluded |
| 542 | + assert dcm_file1_high not in excluded |
| 543 | + assert dcm_file2 not in excluded |
0 commit comments