376 lines
18 KiB
Python
376 lines
18 KiB
Python
import tempfile
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import numpy as np
|
|
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("periodic_noise", operation_ids)
|
|
self.assertIn("inverse_fft_reconstruction", 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["periodic_noise"]["label"], "Periodic Noise")
|
|
self.assertEqual(operations["periodic_noise"]["params"]["A"]["default"], 0.2)
|
|
self.assertEqual(operations["periodic_noise"]["params"]["T"]["default"], 100)
|
|
self.assertEqual(operations["inverse_fft_reconstruction"]["label"], "Inverse FFT Reconstruction")
|
|
self.assertFalse(operations["inverse_fft_reconstruction"]["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_periodic_noise_preserves_grayscale_and_zero_amplitude(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": "periodic_noise", "params": {"A": 0, "T": 100}},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 201)
|
|
self.assertEqual(response.data["channels"], 1)
|
|
self.assertEqual(response.data["color_mode"], "L")
|
|
result = load_image_array(ImageState.objects.get(id=response.data["state_id"]).image)
|
|
self.assertEqual(result.ndim, 2)
|
|
self.assertTrue(np.all(result == 96))
|
|
|
|
def test_periodic_noise_rgb_uses_shared_channel_mask(self):
|
|
upload = self.client.post("/api/images/", {"image": png_upload(color=(80, 80, 80), size=(8, 8))}, format="multipart")
|
|
s0_id = upload.data["states"][0]["state_id"]
|
|
response = self.client.post(
|
|
f"/api/states/{s0_id}/operations/",
|
|
{"operation": "periodic_noise", "params": {"A": 0.1, "T": 4}},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 201)
|
|
result = load_image_array(ImageState.objects.get(id=response.data["state_id"]).image)
|
|
np.testing.assert_array_equal(result[:, :, 0], result[:, :, 1])
|
|
np.testing.assert_array_equal(result[:, :, 1], result[:, :, 2])
|
|
|
|
def test_periodic_noise_rejects_invalid_period(self):
|
|
upload = self.client.post("/api/images/", {"image": grayscale_png_upload()}, format="multipart")
|
|
s0_id = upload.data["states"][0]["state_id"]
|
|
response = self.client.post(
|
|
f"/api/states/{s0_id}/operations/",
|
|
{"operation": "periodic_noise", "params": {"A": 0.2, "T": 0}},
|
|
format="json",
|
|
)
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_inverse_fft_reconstruction_rejects_non_fft_state(self):
|
|
upload = self.client.post("/api/images/", {"image": png_upload(color=(32, 64, 128), size=(4, 4))}, format="multipart")
|
|
s0_id = upload.data["states"][0]["state_id"]
|
|
response = self.client.post(
|
|
f"/api/states/{s0_id}/operations/",
|
|
{"operation": "inverse_fft_reconstruction", "params": {}},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_fft_spectrum_state_stores_complex_data(self):
|
|
upload = self.client.post("/api/images/", {"image": png_upload(color=(32, 64, 128), size=(4, 4))}, format="multipart")
|
|
s0_id = upload.data["states"][0]["state_id"]
|
|
spectrum = self.client.post(
|
|
f"/api/states/{s0_id}/operations/",
|
|
{"operation": "fft_spectrum", "params": {"mode": "log_magnitude"}},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(spectrum.status_code, 201)
|
|
state = ImageState.objects.get(id=spectrum.data["state_id"])
|
|
self.assertEqual(state.operation, "fft_spectrum")
|
|
self.assertIn("fft_data_path", state.params)
|
|
self.assertTrue((Path(self.tmp.name) / state.params["fft_data_path"]).exists())
|
|
|
|
def test_inverse_fft_reconstruction_undoes_saved_fft_state(self):
|
|
upload = self.client.post("/api/images/", {"image": png_upload(color=(32, 64, 128), size=(4, 4))}, format="multipart")
|
|
s0_id = upload.data["states"][0]["state_id"]
|
|
spectrum = self.client.post(
|
|
f"/api/states/{s0_id}/operations/",
|
|
{"operation": "fft_spectrum", "params": {"mode": "log_magnitude"}},
|
|
format="json",
|
|
)
|
|
response = self.client.post(
|
|
f"/api/states/{spectrum.data['state_id']}/operations/",
|
|
{"operation": "inverse_fft_reconstruction", "params": {}},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 201)
|
|
self.assertEqual(response.data["parent_state_id"], spectrum.data["state_id"])
|
|
result = load_image_array(ImageState.objects.get(id=response.data["state_id"]).image)
|
|
expected = load_image_array(ImageState.objects.get(id=s0_id).image)
|
|
np.testing.assert_allclose(result, expected, atol=1)
|
|
|
|
def test_inverse_fft_reconstruction_preserves_grayscale(self):
|
|
upload = self.client.post("/api/images/", {"image": grayscale_png_upload(value=96)}, format="multipart")
|
|
s0_id = upload.data["states"][0]["state_id"]
|
|
spectrum = self.client.post(
|
|
f"/api/states/{s0_id}/operations/",
|
|
{"operation": "fft_spectrum", "params": {"mode": "log_magnitude"}},
|
|
format="json",
|
|
)
|
|
response = self.client.post(
|
|
f"/api/states/{spectrum.data['state_id']}/operations/",
|
|
{"operation": "inverse_fft_reconstruction", "params": {}},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 201)
|
|
self.assertEqual(response.data["channels"], 1)
|
|
result = load_image_array(ImageState.objects.get(id=response.data["state_id"]).image)
|
|
self.assertEqual(result.ndim, 2)
|
|
self.assertTrue(np.allclose(result, 96, atol=1))
|
|
|
|
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()
|