Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 7 additions & 7 deletions tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,9 +22,9 @@ def pytest_html_report_title(report):

def _readable_test_name(item):
"""
Build a human-readable name for the test report from the test's
docstring. Returns None when there is no docstring, so the default
nodeid is kept.
Use the test's docstring (first line) as its name in the report. Every
test is expected to have a docstring; falls back to the default node id
if one is missing.
"""
test_fn = getattr(item, "obj", None)
docstring = inspect.getdoc(test_fn) if test_fn else None
Expand All @@ -33,18 +33,18 @@ def _readable_test_name(item):

# First non-empty line, whitespace collapsed — docstrings are often
# multi-line and indented, which would render badly as a node id.
first_line = next((line.strip() for line in docstring.splitlines() if line.strip()), "")
if not first_line:
base = next((line.strip() for line in docstring.splitlines() if line.strip()), "")
if not base:
return None

# Keep parametrised cases distinct (they share one docstring).
param_id = getattr(getattr(item, "callspec", None), "id", None)
return f"{first_line} [{param_id}]" if param_id else first_line
return f"{base} [{param_id}]" if param_id else base


@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_makereport(item, call):
"""Use the test's docstring as its name in the HTML report, when present."""
"""Use the test's docstring as its name in the report."""
outcome = yield
report = outcome.get_result()

Expand Down
11 changes: 11 additions & 0 deletions tests/integration/test_c_find_returns_worklist_items.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,7 @@ def event(self):
return event

def test_cfind_returns_scheduled_items(self, event, storage):
"""C-FIND returns scheduled items."""
results = list(CFind(storage).call(event))
assert len(results) == 3

Expand Down Expand Up @@ -97,6 +98,7 @@ def test_cfind_returns_scheduled_items(self, event, storage):
assert ds is None

def test_cfind_filters_by_scheduled_date_range(self, event, storage):
"""C-FIND filters by scheduled date range."""
event.identifier.ScheduledProcedureStepSequence[0].ScheduledProcedureStepStartDate = "20240101-20240201"

results = list(CFind(storage).call(event))
Expand All @@ -117,6 +119,7 @@ def test_cfind_filters_by_scheduled_date_range(self, event, storage):
assert ds is None

def test_cfind_filters_by_accession_number(self, event, storage):
"""C-FIND filters by accession number."""
event.identifier.AccessionNumber = "ACC234567"
results = list(CFind(storage).call(event))
assert len(results) == 2
Expand All @@ -128,6 +131,7 @@ def test_cfind_filters_by_accession_number(self, event, storage):
assert ds.AccessionNumber == "ACC234567"

def test_cfind_filters_by_before_scheduled_date(self, event, storage):
"""C-FIND filters by before scheduled date."""
event.identifier.ScheduledProcedureStepSequence[0].ScheduledProcedureStepStartDate = "-20240101"

results = list(CFind(storage).call(event))
Expand All @@ -148,6 +152,7 @@ def test_cfind_filters_by_before_scheduled_date(self, event, storage):
assert ds is None

def test_cfind_filters_by_after_scheduled_date(self, event, storage):
"""C-FIND filters by after scheduled date."""
event.identifier.ScheduledProcedureStepSequence[0].ScheduledProcedureStepStartDate = "20240201-"
results = list(CFind(storage).call(event))

Expand All @@ -167,6 +172,7 @@ def test_cfind_filters_by_after_scheduled_date(self, event, storage):
assert ds is None

def test_cfind_filters_by_scheduled_time_range(self, event, storage):
"""C-FIND filters by scheduled time range."""
event.identifier.ScheduledProcedureStepSequence[0].ScheduledProcedureStepStartTime = "090000-093000"

results = list(CFind(storage).call(event))
Expand All @@ -187,6 +193,7 @@ def test_cfind_filters_by_scheduled_time_range(self, event, storage):
assert ds is None

def test_cfind_filters_by_before_scheduled_time(self, event, storage):
"""C-FIND filters by before scheduled time."""
event.identifier.ScheduledProcedureStepSequence[0].ScheduledProcedureStepStartTime = "-093000"

results = list(CFind(storage).call(event))
Expand All @@ -207,6 +214,7 @@ def test_cfind_filters_by_before_scheduled_time(self, event, storage):
assert ds is None

def test_cfind_filters_by_after_scheduled_time(self, event, storage):
"""C-FIND filters by after scheduled time."""
event.identifier.ScheduledProcedureStepSequence[0].ScheduledProcedureStepStartTime = "093000-"

results = list(CFind(storage).call(event))
Expand All @@ -227,6 +235,7 @@ def test_cfind_filters_by_after_scheduled_time(self, event, storage):
assert ds is None

def test_cfind_filters_by_date_and_time_range(self, event, storage):
"""C-FIND filters by date and time range."""
event.identifier.ScheduledProcedureStepSequence[0].ScheduledProcedureStepStartDate = "20240101-20240201"
event.identifier.ScheduledProcedureStepSequence[0].ScheduledProcedureStepStartTime = "090000-093000"

Expand All @@ -249,6 +258,7 @@ def test_cfind_filters_by_date_and_time_range(self, event, storage):
assert ds is None

def test_cfind_filters_by_modality(self, event, storage):
"""C-FIND filters by modality."""
storage.store_worklist_item(
WorklistItem(
accession_number="ACC999999",
Expand Down Expand Up @@ -284,6 +294,7 @@ def test_cfind_filters_by_modality(self, event, storage):
assert ds is None

def test_cfind_filters_by_patient_id(self, event, storage):
"""C-FIND filters by patient id."""
event.identifier.PatientID = "999234567"

results = list(CFind(storage).call(event))
Expand Down
3 changes: 3 additions & 0 deletions tests/integration/test_c_store_saves_metadata.py
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@ def mwl_storage(self, tmp_dir):
return MWLStorage(f"{tmp_dir}/worklist.db")

def test_existing_sop_instance_uid(self, storage, mock_event):
"""Existing SOP instance UID."""
sop_instance_uid = "1.2.3.4.5.6" # gitleaks:allow
subject = CStore(storage)
mock_event.dataset.file_meta = mock_event.file_meta
Expand All @@ -67,6 +68,7 @@ def test_existing_sop_instance_uid(self, storage, mock_event):
assert len(results) == 1

def test_valid_event_is_stored(self, storage, mock_event):
"""Valid event is stored."""
subject = CStore(storage)

assert subject.call(mock_event) == SUCCESS
Expand All @@ -91,6 +93,7 @@ def test_valid_event_is_stored(self, storage, mock_event):
assert Path(f"{storage.storage_root}/{storage_path}").is_file()

def test_c_store_marks_worklist_in_progress(self, storage, mwl_storage, mock_event):
"""C-STORE marks worklist in progress."""
item = WorklistItem(
accession_number="ABC123",
modality="MG",
Expand Down
7 changes: 7 additions & 0 deletions tests/integration/test_mwl_storage_patient_name_search.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,31 +41,38 @@ def items(storage):

class TestPatientNameSearch:
def test_trailing_wildcard(self, storage):
"""Patient name search: Trailing wildcard."""
results = storage.find_worklist_items(patient_name="SMITH*")
assert {r.patient_name for r in results} == {"SMITH^SARITA", "SMITH^JANE"}

def test_wildcard_on_given_name(self, storage):
"""Wildcard on given name."""
results = storage.find_worklist_items(patient_name="*SARITA")
assert {r.patient_name for r in results} == {"SMITH^SARITA", "JONES^SARITA"}

def test_single_character_wildcard(self, storage):
"""Single character wildcard."""
results = storage.find_worklist_items(patient_name="SMITH^J?NE")
assert {r.patient_name for r in results} == {"SMITH^JANE"}

def test_exact_match(self, storage):
"""Patient name search: Exact match."""
results = storage.find_worklist_items(patient_name="JONES^SARITA")
assert len(results) == 1
assert results[0].patient_name == "JONES^SARITA"

def test_no_match(self, storage):
"""Patient name search: No match."""
results = storage.find_worklist_items(patient_name="BROWN*")
assert results == []

def test_case_insensitive_match(self, storage):
"""Case insensitive match."""
results = storage.find_worklist_items(patient_name="smith*")
assert {r.patient_name for r in results} == {"SMITH^SARITA", "SMITH^JANE"}

@pytest.mark.xfail(reason="SQLite's UPPER() is ASCII-only; non-ASCII case folding requires ICU compilation")
def test_case_insensitive_non_ascii(self, storage):
"""Case insensitive non ascii."""
results = storage.find_worklist_items(patient_name="müller*")
assert {r.patient_name for r in results} == {"MÜLLER^DILMA"}
1 change: 1 addition & 0 deletions tests/integration/test_n_create_updates_worklist_status.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@ def worklist_item(self):
)

def test_n_create_updates_worklist_status(self, tmp_dir, worklist_item):
"""N-CREATE updates worklist status."""
storage = MWLStorage(f"{tmp_dir}/test.db")
study_instance_uid = generate_uid()
accession_number = storage.store_worklist_item(worklist_item)
Expand Down
1 change: 1 addition & 0 deletions tests/integration/test_n_set_updates_worklist_status.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@ def worklist_item(self):
)

def test_n_set_updates_worklist_status(self, tmp_dir, worklist_item, mpps_instance_uid):
"""N-SET updates worklist status."""
storage = MWLStorage(f"{tmp_dir}/test.db")
accession_number = storage.store_worklist_item(worklist_item)
storage.update_status(accession_number, "IN PROGRESS", mpps_instance_uid)
Expand Down
2 changes: 2 additions & 0 deletions tests/integration/test_relay_listener_processes_actions.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ def update_payload(self):

@pytest.mark.asyncio
async def test_relay_listener_creates_worklist_items(self, listener_payload, tmp_dir, fake_relay):
"""Relay listener creates worklist items."""
storage = MWLStorage(f"{tmp_dir}/test_worklist.db")
listener = RelayListener(storage)
relay_message = json.dumps({"accept": {"address": "wss://accept-url"}})
Expand All @@ -34,6 +35,7 @@ async def test_relay_listener_creates_worklist_items(self, listener_payload, tmp

@pytest.mark.asyncio
async def test_relay_listener_updates_worklist_item_status(self, update_payload, tmp_dir, fake_relay):
"""Relay listener updates worklist item status."""
storage = MWLStorage(f"{tmp_dir}/test_worklist.db")
listener = RelayListener(storage)
relay_message = json.dumps({"accept": {"address": "wss://accept-url"}})
Expand Down
2 changes: 2 additions & 0 deletions tests/integration/test_request_cfind_on_worklist.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@ def with_pacs_server(self, tmp_dir):
server.stop()

def test_cfind_request_to_worklist_server(self):
"""C-FIND request to worklist server."""
ae = AE(ae_title="LOCAL_AE_TITLE")
ae.add_requested_context(ModalityWorklistInformationFind)
assoc = ae.associate("0.0.0.0", 4243, ae_title="MWL_SCP_AE_TITLE")
Expand Down Expand Up @@ -82,6 +83,7 @@ def test_cfind_request_to_worklist_server(self):
assert ds is None

def test_cfind_with_filters_request_to_worklist_server(self):
"""C-FIND with filters request to worklist server."""
ae = AE(ae_title="LOCAL_AE_TITLE")
ae.add_requested_context(ModalityWorklistInformationFind)
assoc = ae.associate("0.0.0.0", 4243, ae_title="MWL_SCP_AE_TITLE")
Expand Down
1 change: 1 addition & 0 deletions tests/integration/test_send_c_store_to_gateway.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ def with_pacs_server(self, tmp_dir):
server.stop()

def test_send_dicom_series_to_gateway(self, tmp_dir):
"""Send DICOM series to gateway."""
number_of_instances = 5
storage = PACSStorage(f"{tmp_dir}/test.db", str(tmp_dir))
send_random_dicom_series(
Expand Down
7 changes: 7 additions & 0 deletions tests/scripts/test_database.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ def worklist_db(tmp_dir, monkeypatch):

# Tests for backup_database
def test_backup_creates_file(tmp_dir):
"""Backup creates file."""
db_path = f"{tmp_dir}/test.db"
sqlite3.connect(db_path).close()

Expand All @@ -34,6 +35,7 @@ def test_backup_creates_file(tmp_dir):


def test_backup_returns_timestamped_path(tmp_dir):
"""Backup returns timestamped path."""
db_path = f"{tmp_dir}/test.db"
sqlite3.connect(db_path).close()

Expand All @@ -43,6 +45,7 @@ def test_backup_returns_timestamped_path(tmp_dir):


def test_backup_creates_backup_dir_if_missing(tmp_dir):
"""Backup creates backup dir if missing."""
db_path = f"{tmp_dir}/test.db"
sqlite3.connect(db_path).close()
backup_dir = f"{tmp_dir}/backups/nested"
Expand All @@ -53,12 +56,14 @@ def test_backup_creates_backup_dir_if_missing(tmp_dir):


def test_backup_database_creates_backup(worklist_db):
"""Backup database creates backup."""
backup_path = backup_database(worklist_db, str(Path(worklist_db).parent / "backups"))
assert Path(backup_path).exists()


# Tests for reset_worklist_database
def test_reset_worklist_database_deletes_all_rows(worklist_db):
"""Reset worklist database deletes all rows."""
reset_worklist_database()

with sqlite3.connect(worklist_db) as conn:
Expand All @@ -67,10 +72,12 @@ def test_reset_worklist_database_deletes_all_rows(worklist_db):


def test_reset_worklist_database_returns_row_count(worklist_db):
"""Reset worklist database returns row count."""
assert reset_worklist_database() == 2


def test_reset_worklist_database_returns_zero_when_empty(tmp_dir, monkeypatch):
"""Reset worklist database returns zero when empty."""
db_path = f"{tmp_dir}/worklist.db"
with sqlite3.connect(db_path) as conn:
conn.execute("CREATE TABLE worklist_items (accession_number TEXT PRIMARY KEY)")
Expand Down
1 change: 1 addition & 0 deletions tests/services/dicom/test_c_echo.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,5 +8,6 @@

class TestCEcho:
def test_call(self):
"""C-ECHO: Call."""
event = MagicMock(spec=Event)
assert CEcho().call(event) == SUCCESS
11 changes: 11 additions & 0 deletions tests/services/dicom/test_c_store.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,25 +38,29 @@ def mock_storage(self, mock_pacs_storage):
return mock_pacs_storage.return_value

def test_no_sop_instance_uid_fails(self, mock_storage, mock_event):
"""No SOP instance UID fails."""
subject = CStore(mock_storage)
mock_event.dataset.SOPInstanceUID = None

assert subject.call(mock_event) == FAILURE

def test_no_patient_id_fails(self, mock_storage, mock_event):
"""No patient id fails."""
subject = CStore(mock_storage)
mock_event.dataset.PatientID = None

assert subject.call(mock_event) == FAILURE

def test_existing_sop_instance_uid(self, mock_storage, mock_event):
"""Existing SOP instance UID."""
mock_storage.store_instance.side_effect = pydicom.uid.generate_uid()
subject = CStore(mock_storage)

assert subject.call(mock_event) == SUCCESS
mock_storage.store_instance.assert_called_once()

def test_valid_event_is_stored(self, mock_storage, mock_event):
"""Valid event is stored."""
mock_storage.instance_exists.return_value = False
subject = CStore(mock_storage)

Expand All @@ -74,18 +78,22 @@ def test_valid_event_is_stored(self, mock_storage, mock_event):
assert call_args[0][3] == "ae-title" # AE Title

def test_storage_error_fails(self, mock_storage, mock_event):
"""Storage error fails."""
mock_storage.store_instance.side_effect = Exception("Nooooo!")
subject = CStore(mock_storage)

assert subject.call(mock_event) == FAILURE

def test_failure_hexcode(self):
"""C-STORE: Failure hexcode."""
assert FAILURE == 0xC000

def test_success_hexcode(self):
"""C-STORE: Success hexcode."""
assert SUCCESS == 0x0000

def test_compressor_is_called(self, mock_storage, mock_event):
"""Compressor is called."""
mock_storage.instance_exists.return_value = False
mock_compressor = Mock(spec=ImageCompressor)
mock_compressor.compress.return_value = mock_event.dataset
Expand Down Expand Up @@ -125,6 +133,7 @@ def test_validation_failure_notifies_manage(self, mock_storage, mock_event):
mock_mwl.get_source_message_id.assert_called_once_with("ABC123")

def test_worklist_marked_in_progress_on_success(self, mock_storage, mock_event):
"""Worklist marked in progress on success."""
mock_mwl = Mock(spec=MWLStorage)
subject = CStore(mock_storage, mwl_storage=mock_mwl)

Expand All @@ -133,6 +142,7 @@ def test_worklist_marked_in_progress_on_success(self, mock_storage, mock_event):
mock_mwl.update_status.assert_called_once_with("ABC123", "IN PROGRESS")

def test_worklist_not_updated_on_store_failure(self, mock_storage, mock_event):
"""Worklist not updated on store failure."""
mock_storage.store_instance.side_effect = Exception("store failed")
mock_mwl = Mock(spec=MWLStorage)
subject = CStore(mock_storage, mwl_storage=mock_mwl)
Expand All @@ -142,6 +152,7 @@ def test_worklist_not_updated_on_store_failure(self, mock_storage, mock_event):
mock_mwl.update_status.assert_not_called()

def test_worklist_update_error_does_not_fail_store(self, mock_storage, mock_event):
"""Worklist update error does not fail store."""
mock_mwl = Mock(spec=MWLStorage)
mock_mwl.update_status.side_effect = Exception("db error")
subject = CStore(mock_storage, mwl_storage=mock_mwl)
Expand Down
Loading
Loading