import tempfile from io import BytesIO from pathlib import Path from unittest.mock import patch from django.core.files.uploadedfile import SimpleUploadedFile from django.test import TestCase, override_settings from PIL import Image from rest_framework.test import APIClient from processing.models import ImageState from processing.storage import load_image_array def png_upload(color=(32, 64, 128), size=(4, 4), name="sample.png"): buffer = BytesIO() Image.new("RGB", size, color).save(buffer, format="PNG") return SimpleUploadedFile(name, buffer.getvalue(), content_type="image/png") def grayscale_png_upload(value=96, size=(4, 4), name="gray.png"): buffer = BytesIO() Image.new("L", size, value).save(buffer, format="PNG") return SimpleUploadedFile(name, buffer.getvalue(), content_type="image/png") class ApiTests(TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.override = override_settings(MEDIA_ROOT=Path(self.tmp.name), IMAGE_SESSION_TTL_HOURS=1) self.override.enable() self.client = APIClient() def tearDown(self): self.override.disable() self.tmp.cleanup() def test_upload_and_process(self): upload = self.client.post("/api/images/", {"image": png_upload()}, format="multipart") self.assertEqual(upload.status_code, 201) session_id = upload.data["session_id"] self.assertEqual(len(upload.data["original_histogram"]), 256) processed = self.client.post( "/api/process/", {"session_id": session_id, "operation": "gamma", "params": {"gamma": 1, "c": 1}}, format="json", ) self.assertEqual(processed.status_code, 200) self.assertEqual(len(processed.data["processed_histogram"]), 256) self.assertTrue(processed.data["image_data"].startswith("data:image/png;base64,")) def test_invalid_kernel_rejected(self): upload = self.client.post("/api/images/", {"image": png_upload()}, format="multipart") processed = self.client.post( "/api/process/", {"session_id": upload.data["session_id"], "operation": "median_filter", "params": {"size": 4}}, format="json", ) self.assertEqual(processed.status_code, 400) def test_operations_are_core_slide_set(self): response = self.client.get("/api/operations/") self.assertEqual(response.status_code, 200) operation_ids = {item["id"] for item in response.data["operations"]} operations = {item["id"]: item for item in response.data["operations"]} self.assertIn("histeq", operation_ids) self.assertIn("box_filter", operation_ids) self.assertIn("median_filter", operation_ids) self.assertIn("noise_filter", operation_ids) self.assertIn("average_noisy_copies", operation_ids) self.assertIn("rgb_to_gray", operation_ids) self.assertEqual(operations["histeq"]["params"], {}) self.assertEqual(operations["noise_filter"]["label"], "Noise Filter") self.assertEqual(operations["average_noisy_copies"]["label"], "Average N Noisy Copies") self.assertEqual(operations["average_noisy_copies"]["params"]["N"]["default"], 100) self.assertFalse(operations["average_noisy_copies"]["repeatable"]) self.assertEqual(operations["box_filter"]["label"], "Average / Box Filter") self.assertEqual(operations["gaussian_filter"]["label"], "Gaussian Filter") self.assertFalse(operations["negative"]["repeatable"]) self.assertFalse(operations["rgb_to_gray"]["repeatable"]) self.assertEqual(operations["rgb_to_gray"]["params"]["red_weight"]["default"], 0.299) self.assertEqual(operations["rgb_to_gray"]["params"]["green_weight"]["default"], 0.587) self.assertEqual(operations["rgb_to_gray"]["params"]["blue_weight"]["default"], 0.114) self.assertNotIn("crop", operation_ids) self.assertNotIn("identity", operation_ids) self.assertNotIn("threshold", operation_ids) self.assertNotIn("hist_match", operation_ids) self.assertNotIn("convolution", operation_ids) self.assertNotIn("bone_scan_workflow", operation_ids) self.assertNotIn("gaussian_noise", operation_ids) self.assertNotIn("salt_pepper_noise", operation_ids) self.assertNotIn("speckle_noise", operation_ids) self.assertNotIn("hsi_view", operation_ids) self.assertNotIn("hsi_intensity_filter", operation_ids) self.assertNotIn("frequency_filter", operation_ids) self.assertNotIn("frequency_laplacian", operation_ids) self.assertNotIn("pseudo_color_slices", operation_ids) self.assertNotIn("gray_to_color_transform", operation_ids) def test_upload_compacts_large_image(self): with override_settings(IMAGE_WORKSPACE_MAX_DIMENSION=4): upload = self.client.post("/api/images/", {"image": png_upload(size=(8, 4))}, format="multipart") self.assertEqual(upload.status_code, 201) self.assertEqual(upload.data["width"], 4) self.assertEqual(upload.data["height"], 2) def test_state_delete_removes_non_s0_and_keeps_children(self): upload = self.client.post("/api/images/", {"image": png_upload()}, format="multipart") s0_id = upload.data["states"][0]["state_id"] first = self.client.post( f"/api/states/{s0_id}/operations/", {"operation": "negative", "params": {}}, format="json", ) second = self.client.post( f"/api/states/{first.data['state_id']}/operations/", {"operation": "gamma", "params": {"gamma": 1, "c": 1}}, format="json", ) image_path = Path(self.tmp.name) / first.data["image_path"] self.assertTrue(image_path.exists()) deleted = self.client.delete(f"/api/states/{first.data['state_id']}/") self.assertEqual(deleted.status_code, 204) self.assertFalse(image_path.exists()) states = self.client.get(f"/api/sessions/{upload.data['session_id']}/states/") child = next(item for item in states.data["states"] if item["state_id"] == second.data["state_id"]) self.assertIsNone(child["parent_state_id"]) def test_state_numbering_does_not_reuse_deleted_sequence(self): upload = self.client.post("/api/images/", {"image": png_upload()}, format="multipart") s0_id = upload.data["states"][0]["state_id"] first = self.client.post( f"/api/states/{s0_id}/operations/", {"operation": "negative", "params": {}}, format="json", ) second = self.client.post( f"/api/states/{first.data['state_id']}/operations/", {"operation": "gamma", "params": {"gamma": 1, "c": 1}}, format="json", ) self.assertEqual(first.data["sequence"], 1) self.assertEqual(second.data["sequence"], 2) deleted = self.client.delete(f"/api/states/{first.data['state_id']}/") self.assertEqual(deleted.status_code, 204) third = self.client.post( f"/api/states/{second.data['state_id']}/operations/", {"operation": "negative", "params": {}}, format="json", ) self.assertEqual(third.data["sequence"], 3) self.assertTrue(third.data["label"].startswith("S3 ")) def test_state_delete_blocks_s0(self): upload = self.client.post("/api/images/", {"image": png_upload()}, format="multipart") s0_id = upload.data["states"][0]["state_id"] response = self.client.delete(f"/api/states/{s0_id}/") self.assertEqual(response.status_code, 400) def test_single_image_operation_uses_repeat_count(self): upload = self.client.post("/api/images/", {"image": png_upload()}, format="multipart") s0_id = upload.data["states"][0]["state_id"] response = self.client.post( f"/api/states/{s0_id}/operations/", {"operation": "box_filter", "params": {"_repeat": 2, "K": 3}}, format="json", ) self.assertEqual(response.status_code, 201) self.assertEqual(response.data["params"]["_repeat"], 2) self.assertEqual(response.data["operation"], "box_filter") def test_merged_noise_filter_creates_state(self): upload = self.client.post("/api/images/", {"image": png_upload()}, format="multipart") s0_id = upload.data["states"][0]["state_id"] response = self.client.post( f"/api/states/{s0_id}/operations/", {"operation": "noise_filter", "params": {"kind": "salt_pepper", "amount": 0.1, "salt_ratio": 0.5}}, format="json", ) self.assertEqual(response.status_code, 201) self.assertEqual(response.data["operation"], "noise_filter") self.assertEqual(response.data["params"]["kind"], "salt_pepper") def test_noise_filter_preserves_uploaded_grayscale_channel_count(self): upload = self.client.post("/api/images/", {"image": grayscale_png_upload()}, format="multipart") self.assertEqual(upload.status_code, 201) self.assertEqual(upload.data["channels"], 1) self.assertEqual(upload.data["color_mode"], "L") s0_id = upload.data["states"][0]["state_id"] response = self.client.post( f"/api/states/{s0_id}/operations/", {"operation": "noise_filter", "params": {"kind": "gaussian", "mean": 0, "variance": 0.01}}, format="json", ) self.assertEqual(response.status_code, 201) self.assertEqual(response.data["channels"], 1) self.assertEqual(response.data["color_mode"], "L") state = ImageState.objects.get(id=response.data["state_id"]) self.assertEqual(load_image_array(state.image).ndim, 2) def test_average_noisy_copies_creates_single_grayscale_result_state(self): upload = self.client.post("/api/images/", {"image": grayscale_png_upload(value=96)}, format="multipart") s0_id = upload.data["states"][0]["state_id"] response = self.client.post( f"/api/states/{s0_id}/operations/", {"operation": "average_noisy_copies", "params": {"N": 10, "kind": "gaussian", "mean": 0, "variance": 0}}, format="json", ) self.assertEqual(response.status_code, 201) self.assertEqual(response.data["operation"], "average_noisy_copies") self.assertEqual(response.data["params"]["N"], 10) self.assertEqual(response.data["channels"], 1) self.assertEqual(response.data["color_mode"], "L") self.assertEqual(ImageState.objects.count(), 2) state = ImageState.objects.get(id=response.data["state_id"]) result = load_image_array(state.image) self.assertEqual(result.ndim, 2) self.assertEqual(int(result[0, 0]), 96) def test_grayscale_operation_creates_state(self): upload = self.client.post("/api/images/", {"image": png_upload()}, format="multipart") s0_id = upload.data["states"][0]["state_id"] gray = self.client.post( f"/api/states/{s0_id}/operations/", {"operation": "rgb_to_gray", "params": {"red_weight": 0.299, "green_weight": 0.587, "blue_weight": 0.114}}, format="json", ) self.assertEqual(gray.status_code, 201) self.assertEqual(gray.data["operation"], "rgb_to_gray") self.assertEqual(gray.data["channels"], 1) self.assertEqual(gray.data["color_mode"], "L") noisy = self.client.post( f"/api/states/{gray.data['state_id']}/operations/", {"operation": "noise_filter", "params": {"kind": "salt_pepper", "amount": 0.1, "salt_ratio": 0.5}}, format="json", ) self.assertEqual(noisy.status_code, 201) self.assertEqual(noisy.data["channels"], 1) self.assertEqual(noisy.data["color_mode"], "L") @patch("processing.services.run_batch_job.delay") def test_batch_returns_job_id(self, delay): first = self.client.post("/api/images/", {"image": png_upload(name="a.png")}, format="multipart") second = self.client.post("/api/images/", {"image": png_upload(color=(96, 96, 96), name="b.png")}, format="multipart") response = self.client.post( "/api/batch/", {"operation": "average", "session_ids": [first.data["session_id"], second.data["session_id"]]}, format="json", ) self.assertEqual(response.status_code, 202) self.assertIn("job_id", response.data) delay.assert_called_once()