Spaces:
Sleeping
Sleeping
File size: 11,677 Bytes
0234c58 4780d8d 0234c58 4780d8d 0234c58 b6dcc8c 0234c58 4780d8d 0234c58 4780d8d 0234c58 4780d8d 0234c58 4780d8d 0234c58 b6dcc8c 0234c58 b6dcc8c 4780d8d b6dcc8c 4780d8d b6dcc8c 4780d8d b6dcc8c 4780d8d b6dcc8c 4780d8d b6dcc8c 4780d8d b6dcc8c 0234c58 4780d8d 0234c58 b6dcc8c 0234c58 b6dcc8c 0234c58 4780d8d 0234c58 4780d8d 0234c58 b6dcc8c 0234c58 4780d8d 0234c58 4780d8d 0234c58 4780d8d 0234c58 4780d8d 0234c58 4780d8d b6dcc8c 0234c58 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 |
"""Regression tests for single-slide analysis.
Ensures that single-slide analysis produces identical results before and after
the batch processing optimization.
"""
import pytest
import pandas as pd
from pathlib import Path
from unittest.mock import Mock, patch, MagicMock
import numpy as np
from mosaic.analysis import analyze_slide
from mosaic.ui.app import analyze_slides
class TestSingleSlideRegression:
"""Regression tests to ensure single-slide mode is unchanged."""
@pytest.fixture
def mock_slide_path(self):
"""Mock slide path for testing."""
return "/path/to/test_slide.svs"
@pytest.fixture
def cancer_subtype_name_map(self):
"""Sample cancer subtype name mapping."""
return {
"Unknown": "Unknown",
"Lung Adenocarcinoma": "LUAD",
}
@patch("mosaic.analysis.segment_tissue")
@patch("mosaic.analysis.draw_slide_mask")
@patch("mosaic.model_manager.load_all_models")
@patch("mosaic.analysis._extract_ctranspath_features")
@patch("mosaic.analysis.filter_features")
@patch("mosaic.analysis._extract_optimus_features")
@patch("mosaic.analysis._run_aeon_inference_with_model")
@patch("mosaic.analysis._run_paladin_inference_with_models")
def test_single_slide_analyze_slide_unchanged(
self,
mock_paladin,
mock_aeon,
mock_optimus,
mock_filter,
mock_ctranspath,
mock_load_models,
mock_mask,
mock_segment,
mock_slide_path,
cancer_subtype_name_map,
):
"""Test that analyze_slide function behavior is unchanged."""
# Setup mocks
mock_coords = np.array([[0, 0], [1, 1]])
mock_attrs = {"level": 0}
mock_polygon = Mock()
# segment_tissue returns (polygon, _, coords, attrs)
mock_segment.return_value = (mock_polygon, None, mock_coords, mock_attrs)
mock_mask_image = Mock()
mock_mask.return_value = mock_mask_image
# Mock ModelCache with required attributes
mock_model_cache = Mock()
mock_model_cache.ctranspath_model = Mock()
mock_model_cache.optimus_model = Mock()
mock_model_cache.marker_classifier = Mock()
mock_model_cache.aeon_model = Mock()
mock_model_cache.device = Mock()
mock_model_cache.cleanup = Mock()
mock_load_models.return_value = mock_model_cache
mock_features = np.random.rand(100, 768)
mock_ctranspath.return_value = (mock_features, mock_coords)
mock_filtered_coords = mock_coords[:50]
mock_filter.return_value = (None, mock_filtered_coords)
mock_optimus_features = np.random.rand(50, 1536)
mock_optimus.return_value = mock_optimus_features
mock_aeon_results = pd.DataFrame(
{"Cancer Subtype": ["LUAD", "LUSC"], "Confidence": [0.85, 0.15]}
)
mock_aeon.return_value = mock_aeon_results
mock_paladin_results = pd.DataFrame(
{"Cancer Subtype": ["LUAD"], "Biomarker": ["EGFR"], "Score": [0.75]}
)
mock_paladin.return_value = mock_paladin_results
# Run analyze_slide
slide_mask, aeon_results, paladin_results = analyze_slide(
slide_path=mock_slide_path,
seg_config="Biopsy",
site_type="Primary",
sex="Male",
tissue_site="Lung",
cancer_subtype="Unknown",
cancer_subtype_name_map=cancer_subtype_name_map,
)
# Verify the pipeline was called in correct order
mock_segment.assert_called_once()
mock_mask.assert_called_once()
mock_ctranspath.assert_called_once()
mock_filter.assert_called_once()
mock_optimus.assert_called_once()
mock_aeon.assert_called_once()
mock_paladin.assert_called_once()
# Verify results structure
assert slide_mask == mock_mask_image
assert isinstance(aeon_results, pd.DataFrame)
assert isinstance(paladin_results, pd.DataFrame)
@patch("mosaic.ui.app.analyze_slide")
@patch("mosaic.ui.app.create_user_directory")
@patch("mosaic.ui.app.validate_settings")
@patch("pandas.DataFrame.to_csv") # Mock CSV writing to avoid directory issues
def test_gradio_single_slide_uses_analyze_slide(
self,
mock_to_csv,
mock_validate,
mock_create_dir,
mock_analyze_slide,
):
"""Test that Gradio UI uses analyze_slide for single slide (not batch mode)."""
# Setup
import tempfile
with tempfile.TemporaryDirectory() as tmpdir:
mock_dir = Path(tmpdir) / "test_user"
mock_dir.mkdir()
mock_create_dir.return_value = mock_dir
settings_df = pd.DataFrame(
{
"Slide": ["test.svs"],
"Site Type": ["Primary"],
"Sex": ["Male"],
"Tissue Site": ["Lung"],
"Cancer Subtype": ["Unknown"],
"IHC Subtype": [""],
"Segmentation Config": ["Biopsy"],
}
)
mock_validate.return_value = settings_df
mock_mask = Mock()
mock_aeon = pd.DataFrame({"Cancer Subtype": ["LUAD"], "Confidence": [0.9]})
mock_paladin = pd.DataFrame(
{"Cancer Subtype": ["LUAD"], "Biomarker": ["EGFR"], "Score": [0.8]}
)
mock_analyze_slide.return_value = (mock_mask, mock_aeon, mock_paladin)
from mosaic.ui.app import cancer_subtype_name_map
# Call analyze_slides with a single slide (generator function)
with patch(
"mosaic.ui.app.get_oncotree_code_name",
return_value="Lung Adenocarcinoma",
):
gen = analyze_slides(
slides=["test.svs"],
settings_input=settings_df,
site_type="Primary",
sex="Male",
tissue_site="Lung",
cancer_subtype="Unknown",
ihc_subtype="",
seg_config="Biopsy",
user_dir=mock_dir,
)
# Consume generator to get final result
results = list(gen)
masks, aeon, aeon_btn, paladin, paladin_btn, user_dir = results[-1]
# Verify analyze_slide was called (not analyze_slides_batch)
mock_analyze_slide.assert_called_once()
# Verify results
assert len(masks) == 1
@patch("mosaic.analysis.segment_tissue")
@patch("mosaic.analysis.gr.Warning")
def test_single_slide_no_tissue_found(
self, mock_warning, mock_segment, mock_slide_path, cancer_subtype_name_map
):
"""Test single-slide analysis when no tissue is found."""
# No tissue tiles found
mock_segment.return_value = None # segment_tissue returns None when no tissue
slide_mask, aeon_results, paladin_results = analyze_slide(
slide_path=mock_slide_path,
seg_config="Biopsy",
site_type="Primary",
sex="Unknown",
tissue_site="Unknown",
cancer_subtype="Unknown",
cancer_subtype_name_map=cancer_subtype_name_map,
)
# Should return None for all results
assert slide_mask is None
assert aeon_results is None
assert paladin_results is None
# Verify warning was raised
mock_warning.assert_called_once()
@patch("mosaic.analysis.segment_tissue")
@patch("mosaic.analysis.draw_slide_mask")
@patch("mosaic.model_manager.load_all_models")
@patch("mosaic.analysis._extract_ctranspath_features")
@patch("mosaic.analysis.filter_features")
@patch("mosaic.analysis._extract_optimus_features")
@patch("mosaic.analysis._run_paladin_inference_with_models")
def test_single_slide_known_cancer_subtype_skips_aeon(
self,
mock_paladin,
mock_optimus,
mock_filter,
mock_ctranspath,
mock_load_models,
mock_mask,
mock_segment,
mock_slide_path,
cancer_subtype_name_map,
):
"""Test that single-slide with known subtype skips Aeon inference."""
# Setup minimal mocks
mock_polygon = Mock()
mock_coords = np.array([[0, 0]])
mock_attrs = {}
mock_segment.return_value = (mock_polygon, None, mock_coords, mock_attrs)
mock_mask.return_value = Mock()
# Mock ModelCache
mock_model_cache = Mock()
mock_model_cache.ctranspath_model = Mock()
mock_model_cache.optimus_model = Mock()
mock_model_cache.marker_classifier = Mock()
mock_model_cache.aeon_model = Mock()
mock_model_cache.device = Mock()
mock_model_cache.cleanup = Mock()
mock_load_models.return_value = mock_model_cache
mock_ctranspath.return_value = (np.random.rand(10, 768), np.array([[0, 0]]))
mock_filter.return_value = (None, np.array([[0, 0]]))
mock_optimus.return_value = np.random.rand(10, 1536)
mock_paladin.return_value = pd.DataFrame(
{"Cancer Subtype": ["LUAD"], "Biomarker": ["EGFR"], "Score": [0.8]}
)
with patch("mosaic.analysis._run_aeon_inference_with_model") as mock_aeon:
slide_mask, aeon_results, paladin_results = analyze_slide(
slide_path=mock_slide_path,
seg_config="Biopsy",
site_type="Primary",
sex="Unknown",
tissue_site="Unknown",
cancer_subtype="Lung Adenocarcinoma", # Known subtype
cancer_subtype_name_map=cancer_subtype_name_map,
)
# Aeon inference should NOT be called
mock_aeon.assert_not_called()
# But Paladin should still be called
mock_paladin.assert_called_once()
class TestBackwardCompatibility:
"""Tests to ensure API backward compatibility."""
def test_analyze_slide_signature_unchanged(self):
"""Test that analyze_slide function signature is unchanged."""
from inspect import signature
sig = signature(analyze_slide)
# Verify required parameters exist
params = list(sig.parameters.keys())
assert "slide_path" in params
assert "seg_config" in params
assert "site_type" in params
assert "sex" in params
assert "tissue_site" in params
assert "cancer_subtype" in params
assert "cancer_subtype_name_map" in params
assert "ihc_subtype" in params
assert "num_workers" in params
assert "progress" in params
def test_analyze_slide_return_type_unchanged(self):
"""Test that analyze_slide returns the same tuple structure."""
with patch("mosaic.analysis.segment_tissue", return_value=None): # No tissue
with patch("mosaic.analysis.gr.Warning"): # Mock the warning
result = analyze_slide(
slide_path="test.svs",
seg_config="Biopsy",
site_type="Primary",
sex="Unknown",
tissue_site="Unknown",
cancer_subtype="Unknown",
cancer_subtype_name_map={"Unknown": "Unknown"},
)
# Should return tuple of 3 elements
assert isinstance(result, tuple)
assert len(result) == 3
if __name__ == "__main__":
pytest.main([__file__, "-v"])
|