tests.py

import concurrent.futures
import datetime
import pathlib
import unittest
import uuid

from django.test import TestCase
from rest_framework.test import APIRequestFactory, APITestCase

from ami.base.serializers import reverse_with_params
from ami.main.models import (
    Classification,
    Deployment,
    Detection,
    Event,
    Project,
    SourceImage,
    SourceImageCollection,
    Taxon,
    group_images_into_events,
)
from ami.ml.models import Algorithm, Pipeline, ProcessingService
from ami.ml.models.pipeline import collect_images, get_or_create_algorithm_and_category_map, save_results
from ami.ml.post_processing.small_size_filter import SmallSizeFilterTask
from ami.ml.schemas import (
    AlgorithmConfigResponse,
    AlgorithmReference,
    BoundingBox,
    ClassificationResponse,
    DetectionResponse,
    PipelineResultsResponse,
    SourceImageResponse,
)
from ami.tests.fixtures.main import (
    create_captures_from_files,
    create_processing_service,
    create_taxa,
    setup_test_project,
)
from ami.tests.fixtures.ml import ALGORITHM_CHOICES
from ami.users.models import User


class TestProcessingServiceAPI(APITestCase):
    """
    Test the Processing Services API endpoints.
    """

    def setUp(self):
        self.project = Project.objects.create(name="Processing Service Test Project")

        self.user = User.objects.create_user(  # type: ignore
            email="testuser@insectai.org",
            is_staff=True,
        )
        self.factory = APIRequestFactory()

    def _create_processing_service(self, name: str, endpoint_url: str):
        processing_services_create_url = reverse_with_params(
            "api:processingservice-list", params={"project_id": self.project.pk}
        )
        self.client.force_authenticate(user=self.user)
        processing_service_data = {
            "name": name,
            "endpoint_url": endpoint_url,
        }
        resp = self.client.post(processing_services_create_url, processing_service_data)
        self.client.force_authenticate(user=None)
        self.assertEqual(resp.status_code, 201)
        return resp.json()["instance"]

    def _delete_processing_service(self, processing_service_id: int):
        processing_services_delete_url = reverse_with_params(
            "api:processing-service-detail", kwargs={"pk": processing_service_id}
        )
        self.client.force_authenticate(user=self.user)
        resp = self.client.delete(processing_services_delete_url)
        self.client.force_authenticate(user=None)
        self.assertEqual(resp.status_code, 204)
        return resp

    def _register_pipelines(self, processing_service_id):
        processing_services_register_pipelines_url = reverse_with_params(
            "api:processingservice-register-pipelines", args=[processing_service_id]
        )
        self.client.force_authenticate(user=self.user)
        resp = self.client.post(processing_services_register_pipelines_url)
        data = resp.json()
        self.assertEqual(data["success"], True)
        return data

    def test_create_processing_service(self):
        self._create_processing_service(
            name="Processing Service Test",
            endpoint_url="http://processing_service:2000",
        )

    def test_project_was_added(self):
        response = self._create_processing_service(
            name="Processing Service Test",
            endpoint_url="http://processing_service:2000",
        )
        processing_service_id = response["id"]
        processing_service = ProcessingService.objects.get(pk=processing_service_id)
        self.assertIn(self.project, processing_service.projects.all())

    def test_processing_service_pipeline_registration(self):
        # register a processing service
        response = self._create_processing_service(
            name="Processing Service Test",
            endpoint_url="http://processing_service:2000",
        )
        processing_service_id = response["id"]

        # sync the processing service to create/add the associate pipelines
        response = self._register_pipelines(processing_service_id)
        processing_service = ProcessingService.objects.get(pk=processing_service_id)
        pipelines_queryset = processing_service.pipelines.all()

        self.assertEqual(pipelines_queryset.count(), len(response["pipelines"]))

    def test_create_processing_service_without_endpoint_url(self):
        """Test creating a ProcessingService without endpoint_url (pull mode)"""
        processing_services_create_url = reverse_with_params(
            "api:processingservice-list", params={"project_id": self.project.pk}
        )
        self.client.force_authenticate(user=self.user)
        processing_service_data = {
            "name": "Pull Mode Service",
            "description": "Service without endpoint",
        }
        resp = self.client.post(processing_services_create_url, processing_service_data)
        self.client.force_authenticate(user=None)

        self.assertEqual(resp.status_code, 201)
        data = resp.json()

        # Check that endpoint_url is null
        self.assertIsNone(data["instance"]["endpoint_url"])

        # Check that status indicates service is not yet live (no heartbeat received)
        self.assertFalse(data["status"]["request_successful"])
        self.assertFalse(data["status"]["server_live"])
        self.assertIsNone(data["status"]["endpoint_url"])

    def test_get_status_with_null_endpoint_url(self):
        """Test get_status method when endpoint_url is None"""
        service = ProcessingService.objects.create(name="Pull Mode Service", endpoint_url=None)
        service.projects.add(self.project)

        status = service.get_status()

        self.assertFalse(status.request_successful)
        self.assertFalse(status.server_live)  # No heartbeat received yet = not live
        self.assertIsNone(status.endpoint_url)
        self.assertEqual(status.pipelines_online, [])

    def test_get_pipeline_configs_with_null_endpoint_url(self):
        """Test get_pipeline_configs method when endpoint_url is None"""
        service = ProcessingService.objects.create(name="Pull Mode Service", endpoint_url=None)

        configs = service.get_pipeline_configs()

        self.assertEqual(configs, [])


class TestProcessingServiceLastSeen(TestCase):
    """Test the last_seen, last_seen_live, and last_seen_latency fields."""

    def setUp(self):
        self.project = Project.objects.create(name="Last Seen Test Project")

    def test_mark_seen_sets_fields(self):
        """Test that mark_seen() sets last_seen and last_seen_live."""
        service = ProcessingService.objects.create(name="Async Worker", endpoint_url=None)
        service.projects.add(self.project)

        self.assertIsNone(service.last_seen)
        self.assertIsNone(service.last_seen_live)

        service.mark_seen(live=True)
        service.refresh_from_db()

        self.assertIsNotNone(service.last_seen)
        self.assertTrue(service.last_seen_live)

    def test_mark_seen_offline(self):
        """Test that mark_seen(live=False) sets last_seen_live to False."""
        service = ProcessingService.objects.create(name="Async Worker Offline", endpoint_url=None)

        service.mark_seen(live=False)
        service.refresh_from_db()

        self.assertIsNotNone(service.last_seen)
        self.assertFalse(service.last_seen_live)

    def test_get_status_updates_last_seen_for_sync_service(self):
        """Test that get_status() updates last_seen fields for sync services (even if endpoint is unreachable)."""
        service = ProcessingService.objects.create(name="Sync Service", endpoint_url="http://nonexistent-host:9999")
        service.projects.add(self.project)

        # get_status should update the fields even for unreachable endpoints
        service.get_status(timeout=1)
        service.refresh_from_db()

        self.assertIsNotNone(service.last_seen)
        self.assertFalse(service.last_seen_live)  # unreachable = not live
        self.assertIsNotNone(service.last_seen_latency)

    def test_model_has_last_seen_fields(self):
        """Test that ProcessingService model has last_seen fields and not last_checked."""
        service = ProcessingService.objects.create(name="Field Test Service", endpoint_url=None)
        service.mark_seen(live=True)
        service.refresh_from_db()

        # Verify new fields exist
        self.assertTrue(hasattr(service, "last_seen"))
        self.assertTrue(hasattr(service, "last_seen_live"))
        self.assertTrue(hasattr(service, "last_seen_latency"))

        # Verify old fields don't exist
        self.assertFalse(hasattr(service, "last_checked"))
        self.assertFalse(hasattr(service, "last_checked_live"))
        self.assertFalse(hasattr(service, "last_checked_latency"))


class TestProjectPipelineRegistrationUpdatesLastSeen(APITestCase):
    """Test that async pipeline registration updates last_seen on the processing service."""

    def setUp(self):
        from ami.users.roles import ProjectManager, create_roles_for_project

        self.user = User.objects.create_user(email="lastseen@example.com")  # type: ignore
        self.project = Project.objects.create(name="Last Seen Project", owner=self.user, create_defaults=False)
        create_roles_for_project(self.project)
        ProjectManager.assign_user(self.user, self.project)

    def test_pipeline_registration_marks_service_as_seen(self):
        """Test that POSTing to the pipeline registration endpoint marks the service as last_seen_live."""
        url = f"/api/v2/projects/{self.project.pk}/pipelines/"
        payload = {
            "processing_service_name": "AsyncTestWorker",
            "pipelines": [],
        }

        self.client.force_authenticate(user=self.user)
        response = self.client.post(url, payload, format="json")
        self.assertEqual(response.status_code, 201)

        service = ProcessingService.objects.get(name="AsyncTestWorker")
        self.assertIsNotNone(service.last_seen)
        self.assertTrue(service.last_seen_live)

    def test_repeated_registration_updates_last_seen(self):
        """Test that re-registering updates the last_seen timestamp."""
        url = f"/api/v2/projects/{self.project.pk}/pipelines/"
        payload = {
            "processing_service_name": "AsyncTestWorkerRepeat",
            "pipelines": [],
        }

        self.client.force_authenticate(user=self.user)

        # First registration
        self.client.post(url, payload, format="json")
        service = ProcessingService.objects.get(name="AsyncTestWorkerRepeat")
        first_seen = service.last_seen

        # Second registration
        self.client.post(url, payload, format="json")
        service.refresh_from_db()
        second_seen = service.last_seen

        self.assertIsNotNone(first_seen)
        self.assertIsNotNone(second_seen)
        self.assertGreaterEqual(second_seen, first_seen)


class TestPipelineWithProcessingService(TestCase):
    def test_run_pipeline_with_errors_from_processing_service(self):
        """
        Run a real pipeline and verify that if an error occurs for one image, the error is logged in job.logs.stderr.
        """
        from ami.jobs.models import Job

        # Setup test project, images, and job
        project, deployment = setup_test_project()
        captures = create_captures_from_files(deployment, skip_existing=False)
        test_images = [image for image, frame in captures]
        processing_service_instance = create_processing_service(project)
        pipeline = processing_service_instance.pipelines.all().get(slug="constant")
        job = Job.objects.create(project=project, name="Test Job Real Pipeline Error Handling", pipeline=pipeline)

        # Simulate an error by passing an invalid image (e.g., missing file or corrupt)
        # Here, we manually set the path of one image to a non-existent file
        error_image = test_images[0]
        error_image.path = "/tmp/nonexistent_image.jpg"
        error_image.save()
        images = [error_image] + test_images[1:2]  # Only two images for brevity

        # Run the pipeline and catch any error
        try:
            pipeline.process_images(images, job_id=job.pk, project_id=project.pk)
        except Exception:
            pass  # Expected if the backend raises

        job.refresh_from_db()
        stderr_logs = job.logs.stderr
        # Check that an error message mentioning the failed image is present
        assert any(
            "Failed to process" in log for log in stderr_logs
        ), f"Expected error message in job.logs.stderr, got: {stderr_logs}"

    def setUp(self):
        self.project, self.deployment = setup_test_project()
        self.captures = create_captures_from_files(self.deployment, skip_existing=False)
        self.test_images = [image for image, frame in self.captures]
        self.processing_service_instance = create_processing_service(self.project)
        self.processing_service = self.processing_service_instance
        assert self.processing_service_instance.pipelines.exists()
        self.pipeline = self.processing_service_instance.pipelines.all().get(slug="constant")

    def test_run_pipeline(self):
        # Send images to Processing Service to process and return detections
        assert self.pipeline
        pipeline_response = self.pipeline.process_images(self.test_images, job_id=None, project_id=self.project.pk)
        assert pipeline_response.detections

    def test_created_category_maps(self):
        # Send images to ML backend to process and return detections
        assert self.pipeline
        pipeline_response = self.pipeline.process_images(self.test_images, project_id=self.project.pk)
        save_results(pipeline_response, return_created=True)

        source_images = SourceImage.objects.filter(pk__in=[image.id for image in pipeline_response.source_images])
        detections = Detection.objects.filter(source_image__in=source_images).select_related(
            "detection_algorithm",
            "detection_algorithm__category_map",
        )
        assert detections.count() > 0
        for detection in detections:
            # No detection algorithm should have category map at this time (but this may change!)
            assert detection.detection_algorithm
            assert detection.detection_algorithm.category_map is None

            # Ensure that all classification algorithms have a category map
            classification_taxa = set()
            for classification in detection.classifications.all().select_related(
                "algorithm",
                "algorithm__category_map",
            ):
                assert classification.algorithm is not None
                assert classification.category_map is not None
                assert classification.algorithm.category_map == classification.category_map

                _, top_score = list(classification.predictions(sort=True))[0]
                assert top_score == classification.score

                top_taxon, top_taxon_score = list(classification.predictions_with_taxa(sort=True))[0]
                assert top_taxon == classification.taxon
                assert top_taxon_score == classification.score

                classification_taxa.add(top_taxon)

            # Check the occurrence determination taxon
            assert detection.occurrence
            assert detection.occurrence.determination in classification_taxa

    def test_missing_category_map(self):
        # Test that an exception is raised if a classification algorithm is missing a category map
        from ami.ml.exceptions import PipelineNotConfigured

        # Get the response from the /info endpoint
        pipeline_configs = self.processing_service.get_pipeline_configs()

        # Assert that there is a least one classification algorithm with a category map
        self.assertTrue(
            any(
                algo.task_type in Algorithm.classification_task_types and algo.category_map is not None
                for pipeline in pipeline_configs
                for algo in pipeline.algorithms
            ),
            "Expected pipeline to have at least one classification algorithm with a category map",
        )

        # Remove the category map from one of the classification algorithms
        for pipeline_config in pipeline_configs:
            for algorithm in pipeline_config.algorithms:
                if algorithm.task_type in Algorithm.classification_task_types and algorithm.category_map is not None:
                    algorithm.category_map = None
                    # Change the key to ensure it's treated as a new algorithm
                    algorithm.key = "missing-category-map-classifier"
                    algorithm.name = "Classifier with Missing Category Map"
                    break

        with self.assertRaises(
            PipelineNotConfigured,
            msg="Expected an exception to be raised if a classification algorithm is missing a category map",
        ):
            self.processing_service.create_pipelines(pipeline_configs=pipeline_configs)

    def test_alignment_of_predictions_and_category_map(self):
        # Ensure that the scores and labels are aligned
        pipeline = self.processing_service_instance.pipelines.all().get(slug="random-detection-random-species")
        pipeline_response = pipeline.process_images(self.test_images, project_id=self.project.pk)
        results = save_results(pipeline_response, return_created=True)
        assert results is not None, "Expected results to be returned in a PipelineSaveResults object"
        assert results.classifications, "Expected classifications to be returned in the results"
        for classification in results.classifications:
            assert classification.scores
            taxa_with_scores = list(classification.predictions_with_taxa(sort=True))
            assert taxa_with_scores
            assert classification.score == taxa_with_scores[0][1]
            assert classification.taxon == taxa_with_scores[0][0]

    def test_top_n_alignment(self):
        # Ensure that the top_n parameter works
        pipeline = self.processing_service_instance.pipelines.all().get(slug="random-detection-random-species")
        pipeline_response = pipeline.process_images(self.test_images, project_id=self.project.pk)
        results = save_results(pipeline_response, return_created=True)
        assert results is not None, "Expecected results to be returned in a PipelineSaveResults object"
        assert results.classifications, "Expected classifications to be returned in the results"
        for classification in results.classifications:
            top_n = classification.top_n(n=3)
            assert classification.score == top_n[0]["score"]
            assert classification.taxon == top_n[0]["taxon"]

    def test_pipeline_reprocessing(self):
        """
        Test that reprocessing the same images with differet pipelines does not create duplicate
        detections. The 2 pipelines used are a random detection + random species classifier, and a
        constant species classifier.
        """
        if not self.project.feature_flags.reprocess_existing_detections:
            self.project.feature_flags.reprocess_existing_detections = True
            self.project.save()

        # Process the images once
        pipeline_one = self.processing_service_instance.pipelines.all().get(slug="random-detection-random-species")
        num_classifiers_pipeline_one = pipeline_one.algorithms.filter(task_type="classification").count()
        pipeline_response = pipeline_one.process_images(self.test_images, project_id=self.project.pk)
        results = save_results(pipeline_response, return_created=True)
        assert results is not None, "Expected results to be returned in a PipelineSaveResults object"
        assert results.detections, "Expected detections to be returned in the results"
        num_initial_detections = len(results.detections)

        # This particular pipeline produces 2 classifications per detection
        for det in results.detections:
            num_classifications = det.classifications.count()
            self.assertEqual(
                num_classifications,
                num_classifiers_pipeline_one,
                f"Expected {num_classifiers_pipeline_one} classifications per detection "
                "(random species and random binary classifier).",
            )

        source_images = SourceImage.objects.filter(pk__in=[image.id for image in pipeline_response.source_images])
        detections = Detection.objects.filter(source_image__in=source_images).select_related(
            "detection_algorithm",
            "detection_algorithm__category_map",
        )
        initial_detection_ids = sorted([det.pk for det in detections])
        assert detections.count() > 0

        # Reprocess the same images using a different pipeline
        pipeline_two = self.processing_service_instance.pipelines.all().get(slug="constant")
        num_classifiers_pipeline_two = pipeline_two.algorithms.filter(task_type="classification").count()
        pipeline_response = pipeline_two.process_images(self.test_images, project_id=self.project.pk)
        reprocessed_results = save_results(pipeline_response, return_created=True)
        assert reprocessed_results is not None, "Expected results to be returned in a PipelineSaveResults object"
        assert reprocessed_results.detections, "Expected detections to be returned in the results"
        num_reprocessed_detections = len(reprocessed_results.detections)
        self.assertEqual(
            num_reprocessed_detections,
            num_initial_detections,
            "Expected the same number of detections after reprocessing with a different pipeline.",
        )

        source_images = SourceImage.objects.filter(pk__in=[image.id for image in pipeline_response.source_images])
        detections = Detection.objects.filter(source_image__in=source_images).select_related(
            "detection_algorithm",
            "detection_algorithm__category_map",
        )

        # Check detections were re-processed, and not re-created
        reprocessed_detection_ids = sorted([det.pk for det in detections])
        assert initial_detection_ids == reprocessed_detection_ids, (
            "Expected the same detections to be returned after reprocessing with a different pipeline, "
            f"but found {initial_detection_ids} != {reprocessed_detection_ids}"
        )

        # The constant pipeline produces 1 classification per detection (added to the existing classifications)
        for detection in detections:
            self.assertEqual(
                detection.classifications.count(),
                num_classifiers_pipeline_one + num_classifiers_pipeline_two,
                f"Expected {num_classifiers_pipeline_one + num_classifiers_pipeline_two} "
                "classifications per detection (2 random classifiers + constant classifier).",
            )


class TestPipeline(TestCase):
    def setUp(self):
        self.project = Project.objects.create(name="Test Project")
        # Create test images and collection
        self.test_images = [
            SourceImage.objects.create(path="test1-20240101000000.jpg"),
            SourceImage.objects.create(path="test2-20240101001000.jpg"),
        ]
        self.image_collection = SourceImageCollection.objects.create(
            name="Test Collection",
            project=self.project,
        )
        self.image_collection.images.set(self.test_images)

        # Create test pipeline and algorithms
        self.pipeline = Pipeline.objects.create(
            name="Test Pipeline (Random)",
        )
        self.pipeline_two = Pipeline.objects.create(
            name="Test Pipeline (Constant)",
        )

        self.algorithms = {
            key: get_or_create_algorithm_and_category_map(val) for key, val in ALGORITHM_CHOICES.items()
        }
        self.pipeline.algorithms.set(
            [
                self.algorithms["random-detector"],
                self.algorithms["random-binary-classifier"],
                self.algorithms["random-species-classifier"],
            ]
        )
        self.pipeline_two.algorithms.set(
            [
                self.algorithms["random-detector"],
                self.algorithms["random-binary-classifier"],
                self.algorithms["constant-species-classifier"],
            ]
        )

    def test_create_pipeline(self):
        assert self.pipeline.slug.startswith("test-pipeline")
        self.assertEqual(self.pipeline.algorithms.count(), 3)
        self.assertEqual(self.pipeline_two.algorithms.count(), 3)

        for algorithm in self.pipeline.algorithms.all():
            assert isinstance(algorithm, Algorithm)
            self.assertIn(algorithm.key, [algo.key for algo in ALGORITHM_CHOICES.values()])

    def test_collect_images(self):
        images = list(collect_images(collection=self.image_collection, pipeline=self.pipeline))
        assert len(images) == 2

    def fake_pipeline_results(
        self,
        source_images: list[SourceImage],
        pipeline: Pipeline,
        alt_species_classifier: AlgorithmConfigResponse | None = None,
    ):
        # @TODO use the pipeline passed in to get the algorithms
        source_image_results = [SourceImageResponse(id=image.pk, url=image.path) for image in source_images]
        detector = ALGORITHM_CHOICES["random-detector"]
        binary_classifier = ALGORITHM_CHOICES["random-binary-classifier"]
        assert binary_classifier.category_map

        if alt_species_classifier is None:
            species_classifier = ALGORITHM_CHOICES["random-species-classifier"]
        else:
            species_classifier = alt_species_classifier
        assert species_classifier.category_map

        detection_results = [
            DetectionResponse(
                source_image_id=image.pk,
                bbox=BoundingBox(x1=0.0, y1=0.0, x2=1.0, y2=1.0),
                inference_time=0.4,
                algorithm=AlgorithmReference(
                    name=detector.name,
                    key=detector.key,
                ),
                timestamp=datetime.datetime.now(),
                classifications=[
                    ClassificationResponse(
                        classification=binary_classifier.category_map.labels[0],
                        labels=binary_classifier.category_map.labels,
                        scores=[0.9213],
                        algorithm=AlgorithmReference(
                            name=binary_classifier.name,
                            key=binary_classifier.key,
                        ),
                        timestamp=datetime.datetime.now(),
                        terminal=False,
                    ),
                    ClassificationResponse(
                        classification=species_classifier.category_map.labels[0],
                        labels=species_classifier.category_map.labels,
                        scores=[0.64333],
                        algorithm=AlgorithmReference(
                            name=species_classifier.name,
                            key=species_classifier.key,
                        ),
                        timestamp=datetime.datetime.now(),
                        terminal=True,
                    ),
                ],
            )
            for image in self.test_images
        ]
        fake_results = PipelineResultsResponse(
            pipeline=pipeline.slug,
            algorithms={
                detector.key: detector,
                binary_classifier.key: binary_classifier,
                species_classifier.key: species_classifier,
            },
            total_time=0.01,
            source_images=source_image_results,
            detections=detection_results,
        )
        return fake_results

    def test_save_results(self):
        results = self.fake_pipeline_results(self.test_images, self.pipeline)
        save_results(results)

        for image in self.test_images:
            image.save()
            self.assertEqual(image.detections_count, 1)

        # @TODO test the cached counts for detections, etc are updated on Events, Deployments, etc.

    def test_skip_existing_when_all_matching(self):
        """
        When processing images, skip images that have already been processed by the same set of algorithms.
        (must be the same detection algorithm and all classification algorithms)
        """

        images = list(collect_images(collection=self.image_collection, pipeline=self.pipeline))
        total_images = len(images)
        self.assertEqual(total_images, self.image_collection.images.count())

        created = save_results(self.fake_pipeline_results(images, self.pipeline), return_created=True)
        assert created, "Expected created objects to be returned in a PipelineSaveResults object"

        # Collect all detection algorithms used on the detections
        detections = created.detections
        detection_algos_used = {
            detection.detection_algorithm.name for detection in detections if detection.detection_algorithm
        }
        # detection_algos_used = set(Detection.objects.all().values_list("detection_algorithm__name", flat=True))

        # Assert it was only one algorithm, and it was the one we used
        self.assertEqual(
            detection_algos_used, {self.algorithms["random-detector"].name}, "Wrong detection algorithm used."
        )

        # Collect all classification algorithms used on the classifications
        classifications = created.classifications
        classification_algos_used = {
            classification.algorithm.name for classification in classifications if classification.algorithm
        }
        # classification_algos_used = set(Classification.objects.all().values_list("algorithm__name", flat=True))
        # Assert it was only one algorithm, and it was the one we used
        self.assertEqual(
            classification_algos_used,
            {self.algorithms["random-species-classifier"].name, self.algorithms["random-binary-classifier"].name},
            "Wrong classification algorithms used.",
        )

        images_again = list(collect_images(collection=self.image_collection, pipeline=self.pipeline))
        remaining_images_to_process = len(images_again)
        self.assertEqual(remaining_images_to_process, 0)

    def test_skip_existing_with_new_detector(self):
        images = list(collect_images(collection=self.image_collection, pipeline=self.pipeline))
        total_images = len(images)
        self.assertEqual(total_images, self.image_collection.images.count())
        pipeline_response = self.fake_pipeline_results(images, self.pipeline)
        save_results(pipeline_response)
        # Find the fist algo used where task_type is classification
        classifiers = [algo for algo in pipeline_response.algorithms.values() if algo.task_type == "classification"]
        last_classifier = Algorithm.objects.get(key=classifiers[-1].key)
        self.pipeline.algorithms.set(
            [
                Algorithm.objects.create(name="NEW Object Detector 2.0"),
                last_classifier,
            ]
        )
        images_again = list(collect_images(collection=self.image_collection, pipeline=self.pipeline))
        remaining_images_to_process = len(images_again)
        self.assertEqual(remaining_images_to_process, total_images)

    @unittest.skip("Not implemented yet")
    def test_skip_existing_with_new_classifier(self):
        """
        @TODO add support for skipping the detection model if only the classifier has changed.
        """
        images = list(collect_images(collection=self.image_collection, pipeline=self.pipeline))
        total_images = len(images)
        self.assertEqual(total_images, self.image_collection.images.count())
        pipeline_response = self.fake_pipeline_results(images, self.pipeline)
        # Find the fist algo used where task_type is detection
        first_detector_in_response = next(
            algo for algo in pipeline_response.algorithms.values() if algo.task_type == "detection"
        )
        first_detector = Algorithm.objects.get(key=first_detector_in_response.key)
        save_results(pipeline_response)
        self.pipeline.algorithms.set(
            [
                first_detector,
                Algorithm.objects.create(name="NEW Classifier 2.0"),
            ]
        )
        images_again = list(collect_images(collection=self.image_collection, pipeline=self.pipeline))
        remaining_images_to_process = len(images_again)
        self.assertEqual(remaining_images_to_process, total_images)

    @unittest.skip("Not implemented yet")
    def test_skip_existing_per_batch_during_processing(self):
        # Send the same batch to two simultaneous processing pipelines
        # @TODO this needs to test the `process_images()` function with a real pipeline
        # @TODO enable test when a pipeline is added to the CI environment in PR #576
        pass

    def test_unknown_algorithm_returned_by_processing_service(self):
        """
        Test that unknown algorithms returned by the processing service are handled correctly.

        Previously we allowed unknown algorithms to be returned by the pipeline,
        now all algorithms must be registered first from the processing service's /info
        endpoint.
        """
        fake_results = self.fake_pipeline_results(self.test_images, self.pipeline)

        new_detector = AlgorithmConfigResponse(
            name="Unknown Detector 5.1b-mobile", key="unknown-detector", task_type="detection"
        )
        new_classifier = AlgorithmConfigResponse(
            name="Unknown Classifier 3.0b-mega", key="unknown-classifier", task_type="classification"
        )

        fake_results.algorithms[new_detector.key] = new_detector
        fake_results.algorithms[new_classifier.key] = new_classifier

        for detection in fake_results.detections:
            detection.algorithm = AlgorithmReference(name=new_detector.name, key=new_detector.key)

            for classification in detection.classifications:
                classification.algorithm = AlgorithmReference(name=new_classifier.name, key=new_classifier.key)

        current_total_algorithm_count = Algorithm.objects.count()

        # Ensure an exception is raised that a new algorithm was not
        # pre-registered from the /info endpoint
        from ami.ml.exceptions import PipelineNotConfigured

        with self.assertRaises(PipelineNotConfigured):
            save_results(fake_results)

        # Ensure no new algorithms were added to the database
        new_algorithm_count = Algorithm.objects.count()
        self.assertEqual(new_algorithm_count, current_total_algorithm_count)

        # Ensure new algorithms were also added to the pipeline

    def test_yes_reprocess_if_new_terminal_algorithm_same_intermediate(self):
        """
        Test two pipelines with the same detector and same moth/non-moth classifier, but a new species classifier.

        The first pipeline should process the images and save the results.
        The second pipeline should reprocess the images.
        """

        images = list(collect_images(collection=self.image_collection, pipeline=self.pipeline))
        assert len(images), "No images to process"

        detector = Algorithm.objects.get(key="random-detector")
        binary_classifier = Algorithm.objects.get(key="random-binary-classifier")
        old_species_classifier = Algorithm.objects.get(key="random-species-classifier")

        Detection.objects.all().delete()
        results = save_results(self.fake_pipeline_results(images, self.pipeline), return_created=True)
        assert results is not None, "Expecected results to be returned in a PipelineSaveResults object"

        for raw_detection in results.detections:
            self.assertEqual(raw_detection.detection_algorithm, detector)

        # Ensure all results have the binary classifier and the old species classifier
        for saved_detection in Detection.objects.all():
            self.assertEqual(saved_detection.detection_algorithm, detector)
            # Assert that the binary classifier was used
            self.assertTrue(
                saved_detection.classifications.filter(algorithm=binary_classifier).exists(),
                "Binary classifier not used in first run",
            )
            # Assert that the old species classifier was used
            self.assertTrue(
                saved_detection.classifications.filter(algorithm=old_species_classifier).exists(),
                "Old species classifier not used in first run",
            )

        # Get another species classifier
        new_species_classifier_key = "constant-species-classifier"
        new_species_classifier = Algorithm.objects.get(key=new_species_classifier_key)
        # new_species_classifier_response = ALGORITHM_CHOICES[new_species_classifier_key]

        # Create a new pipeline with the same detector and the new species classifier
        new_pipeline = Pipeline.objects.create(
            name="New Pipeline",
        )

        new_pipeline.algorithms.set(
            [
                detector,
                binary_classifier,
                new_species_classifier,
            ]
        )

        # Process the images with the new pipeline
        images_again = list(collect_images(collection=self.image_collection, pipeline=new_pipeline))
        remaining_images_to_process = len(images_again)
        self.assertEqual(remaining_images_to_process, len(images), "Images not re-processed with new pipeline")

    def test_project_pipeline_config(self):
        """
        Test the default_config for a pipeline, as well as the project pipeline config.
        Ensure the project pipeline parameters override the pipeline defaults.
        """
        from ami.ml.models import ProjectPipelineConfig
        from ami.ml.schemas import PipelineRequestConfigParameters

        # Add config to the pipeline & project
        self.pipeline.default_config = PipelineRequestConfigParameters({"test_param": "test_value"})
        self.pipeline.save()
        self.project_pipeline_config = ProjectPipelineConfig.objects.create(
            project=self.project,
            pipeline=self.pipeline,
            config={"test_param": "project_value"},
        )
        self.project_pipeline_config.save()

        # Check the final config
        default_config = self.pipeline.get_config()
        self.assertEqual(default_config["test_param"], "test_value")
        final_config = self.pipeline.get_config(self.project.pk)
        self.assertEqual(final_config["test_param"], "project_value")

    def test_image_with_null_detection(self):
        """
        Test saving results for a pipeline that returns null detections for some images.
        """
        image = self.test_images[0]
        results = self.fake_pipeline_results([image], self.pipeline)

        # Manually change the results for a single image to a list of empty detections
        results.detections = []

        save_results(results)

        image.save()
        self.assertEqual(image.get_detections_count(), 0)  # detections_count should exclude null detections
        total_num_detections = image.detections.distinct().count()
        self.assertEqual(total_num_detections, 1)

        was_processed = image.get_was_processed()
        self.assertEqual(was_processed, True)

        # Also test filtering by algorithm
        was_processed = image.get_was_processed(algorithm_key="random-detector")
        self.assertEqual(was_processed, True)

    def test_filter_processed_images_skips_null_only_image(self):
        """
        An image with only null detections (processed, nothing found) should be
        skipped by filter_processed_images — it doesn't need reprocessing.
        """
        from ami.ml.models.pipeline import filter_processed_images

        image = self.test_images[0]
        detector = self.algorithms["random-detector"]

        # Simulate a previous run that found nothing: create a null detection
        Detection.objects.create(
            source_image=image,
            detection_algorithm=detector,
            bbox=None,
        )

        result = list(filter_processed_images([image], self.pipeline))
        self.assertEqual(result, [], "Image with only null detections should be skipped")

    def test_filter_processed_images_yields_image_with_null_and_real_unclassified(self):
        """
        An image with BOTH a null detection AND a real detection lacking classifications
        should NOT be skipped — the real detection still needs to be classified.
        """
        from ami.ml.models.pipeline import filter_processed_images

        image = self.test_images[0]
        detector = self.algorithms["random-detector"]

        # Null detection from a prior empty run
        Detection.objects.create(
            source_image=image,
            detection_algorithm=detector,
            bbox=None,
        )
        # Real detection with no classification yet
        Detection.objects.create(
            source_image=image,
            detection_algorithm=detector,
            bbox=[0.1, 0.2, 0.3, 0.4],
        )

        result = list(filter_processed_images([image], self.pipeline))
        self.assertEqual(result, [image], "Image with real unclassified detections should be yielded")

    def test_filter_processed_images_skips_null_and_fully_classified(self):
        """
        An image with a null detection AND a real detection that is fully classified
        by all pipeline algorithms should be skipped — it's fully processed.
        """
        from ami.ml.models.pipeline import filter_processed_images

        image = self.test_images[0]
        detector = self.algorithms["random-detector"]
        binary_classifier = self.algorithms["random-binary-classifier"]
        species_classifier = self.algorithms["random-species-classifier"]

        # Null detection from a prior empty run
        Detection.objects.create(
            source_image=image,
            detection_algorithm=detector,
            bbox=None,
        )
        # Real detection with classifications from all pipeline algorithms
        real_det = Detection.objects.create(
            source_image=image,
            detection_algorithm=detector,
            bbox=[0.1, 0.2, 0.3, 0.4],
        )
        taxon = Taxon.objects.create(name="Test Species Filtered")
        Classification.objects.create(
            detection=real_det,
            taxon=taxon,
            algorithm=binary_classifier,
            score=0.9,
            timestamp=datetime.datetime.now(),
        )
        Classification.objects.create(
            detection=real_det,
            taxon=taxon,
            algorithm=species_classifier,
            score=0.8,
            timestamp=datetime.datetime.now(),
        )

        result = list(filter_processed_images([image], self.pipeline))
        self.assertEqual(result, [], "Fully classified image with null detection should be skipped")

    def test_null_detections_are_algorithm_specific(self):
        """
        Null detections from different pipelines/algorithms should not be shared.
        Each algorithm's null detection is tracked separately so that
        get_was_processed(algorithm_key=...) returns the correct per-algorithm status.
        """
        from ami.ml.models.pipeline import save_results

        image = self.test_images[0]

        # Pipeline 1 processes image, finds nothing
        results_1 = self.fake_pipeline_results([image], self.pipeline)
        results_1.detections = []
        save_results(results_1)

        # Create a second pipeline with a DIFFERENT detector algorithm
        detector_2, _ = Algorithm.objects.get_or_create(
            key="constant-detector",
            defaults={"name": "Constant Detector", "task_type": "detection"},
        )
        pipeline_2 = Pipeline.objects.create(name="Test Pipeline 2 Null Detect")
        pipeline_2.algorithms.set([detector_2])

        # Pipeline 2 processes the same image, also finds nothing
        results_2 = self.fake_pipeline_results([image], pipeline_2)
        results_2.detections = []
        save_results(results_2)

        # Both algorithms should independently mark the image as processed
        detector_1_key = self.algorithms["random-detector"].key
        self.assertTrue(image.get_was_processed(algorithm_key=detector_1_key))
        self.assertTrue(
            image.get_was_processed(algorithm_key="constant-detector"),
            "Pipeline 2's null detection should be created separately",
        )

        # Each pipeline must have its own null detection in the DB
        null_detections = image.detections.filter(bbox__isnull=True)
        self.assertEqual(null_detections.count(), 2, "Each pipeline should have its own null detection")

    def test_null_detection_deduplication_same_pipeline(self):
        """
        Running the same pipeline twice on the same image should not create
        duplicate null detections — the second run reuses the existing one.
        """
        from ami.ml.models.pipeline import save_results

        image = self.test_images[0]

        # Run pipeline twice, both with no detections
        results_1 = self.fake_pipeline_results([image], self.pipeline)
        results_1.detections = []
        save_results(results_1)

        results_2 = self.fake_pipeline_results([image], self.pipeline)
        results_2.detections = []
        save_results(results_2)

        # Should still be exactly one null detection
        null_detections = image.detections.filter(bbox__isnull=True)
        self.assertEqual(null_detections.count(), 1, "Same pipeline should not create duplicate null detections")


class TestAlgorithmCategoryMaps(TestCase):
    def setUp(self):
        self.algorithm_responses = {
            key: get_or_create_algorithm_and_category_map(val) for key, val in ALGORITHM_CHOICES.items()
        }
        self.algorithms = {key: Algorithm.objects.get(key=key) for key in ALGORITHM_CHOICES.keys()}

    def test_create_algorithms_and_category_map(self):
        assert len(self.algorithms) > 0
        assert (
            Algorithm.objects.filter(
                key__in=self.algorithms.keys(),
            )
            .exclude(category_map=None)
            .count()
        ) > 0

    def test_algorithm_category_maps(self):
        for algorithm in Algorithm.objects.filter(
            key__in=self.algorithms.keys(),
        ).exclude(category_map=None):
            assert algorithm.category_map  # For type checker, not the test
            assert algorithm.category_map.labels
            assert algorithm.category_map.labels_hash
            assert algorithm.category_map.data

            # Ensure the full labels in the data match the simple, ordered list of labels
            sorted_data = sorted(algorithm.category_map.data, key=lambda x: x["index"])
            assert [category["label"] for category in sorted_data] == algorithm.category_map.labels

    def test_labels_hash_auto_generation(self):
        """Test that labels_hash is automatically generated when creating AlgorithmCategoryMap instances."""
        from ami.ml.models import AlgorithmCategoryMap

        # Test data
        test_data = [
            {"index": 0, "label": "coleoptera"},
            {"index": 1, "label": "diptera"},
            {"index": 2, "label": "lepidoptera"},
        ]
        test_labels = AlgorithmCategoryMap.labels_from_data(test_data)

        # Create instance using objects.create()
        category_map = AlgorithmCategoryMap.objects.create(labels=test_labels, data=test_data, version="test-v1")

        # Verify labels_hash was automatically generated
        self.assertIsNotNone(category_map.labels_hash)

        # Verify the hash matches what make_labels_hash would produce
        expected_hash = AlgorithmCategoryMap.make_labels_hash(test_labels)
        self.assertEqual(category_map.labels_hash, expected_hash)

        # Test that creating another instance with same labels produces same hash
        category_map2 = AlgorithmCategoryMap.objects.create(labels=test_labels, data=test_data, version="test-v2")

        self.assertEqual(category_map.labels_hash, category_map2.labels_hash)

    def test_labels_data_conversion_methods(self):
        from ami.ml.models import AlgorithmCategoryMap

        # Test data
        test_data = [
            {"index": 0, "label": "coleoptera"},
            {"index": 1, "label": "diptera"},
            {"index": 2, "label": "lepidoptera"},
        ]
        test_labels = AlgorithmCategoryMap.labels_from_data(test_data)

        # Convert labels to data and back
        converted_data = AlgorithmCategoryMap.data_from_labels(test_labels)
        converted_labels = AlgorithmCategoryMap.labels_from_data(converted_data)

        # Verify conversions are correct
        self.assertEqual(test_data, converted_data)
        self.assertEqual(test_labels, converted_labels)


class TestPostProcessingTasks(TestCase):
    def setUp(self):
        # Create test project, deployment, and default setup
        self.project, self.deployment = setup_test_project()
        create_taxa(project=self.project)
        self._create_images_with_dimensions(deployment=self.deployment)
        group_images_into_events(deployment=self.deployment)

        # Create a simple SourceImageCollection for testing
        self.collection = SourceImageCollection.objects.create(
            name="Test PostProcessing Collection",
            project=self.project,
            method="manual",
            kwargs={"image_ids": list(self.deployment.captures.values_list("pk", flat=True))},
        )
        self.collection.populate_sample()

    def _create_images_with_dimensions(
        self,
        deployment,
        num_images: int = 5,
        width: int = 640,
        height: int = 480,
        update_deployment: bool = True,
    ):
        """
        Create SourceImages for a deployment with specified width and height.
        """

        created = []
        base_time = datetime.datetime.now(datetime.timezone.utc)

        for i in range(num_images):
            random_prefix = uuid.uuid4().hex[:8]
            path = pathlib.Path("test") / f"{random_prefix}_{i}.jpg"

            image = SourceImage.objects.create(
                deployment=deployment,
                project=deployment.project,
                timestamp=base_time + datetime.timedelta(minutes=i * 5),
                path=path,
                width=width,
                height=height,
            )
            created.append(image)

        if update_deployment:
            deployment.save(update_calculated_fields=True, regroup_async=False)

    def test_small_size_filter_assigns_not_identifiable(self):
        """
        Test that SmallSizeFilterTask correctly assigns 'Not identifiable'
        to detections below the configured minimum size.
        """
        # Create small detections on the collection images
        for image in self.collection.images.all():
            Detection.objects.create(
                source_image=image,
                bbox=[0, 0, 10, 10],  # small detection
                created_at=datetime.datetime.now(datetime.timezone.utc),
            ).associate_new_occurrence()

        # Prepare the task configuration
        task = SmallSizeFilterTask(
            source_image_collection_id=self.collection.pk,
            size_threshold=0.01,
        )

        task.run()

        # Verify that all small detections are now classified as "Not identifiable"
        not_identifiable_taxon = Taxon.objects.get(name="Not identifiable")
        detections = Detection.objects.filter(source_image__in=self.collection.images.all())

        for det in detections:
            latest_classification = Classification.objects.filter(detection=det).order_by("-created_at").first()
            self.assertIsNotNone(latest_classification, "Each detection should have a classification.")
            self.assertEqual(
                latest_classification.taxon,
                not_identifiable_taxon,
                f"Detection {det.pk} should be classified as 'Not identifiable'",
            )
            occurrence = det.occurrence
            self.assertIsNotNone(occurrence, f"Detection {det.pk} should belong to an occurrence.")
            occurrence.refresh_from_db()
            self.assertEqual(
                occurrence.determination,
                not_identifiable_taxon,
                f"Occurrence {occurrence.pk} should have its determination set to 'Not identifiable'.",
            )


class TestTaskStateManager(TestCase):
    """Test TaskStateManager for job progress tracking."""

    def setUp(self):
        """Set up test fixtures."""
        from django.core.cache import cache

        from ami.ml.orchestration.async_job_state import AsyncJobStateManager

        cache.clear()
        self.job_id = 123
        self.manager = AsyncJobStateManager(self.job_id)
        self.image_ids = ["img1", "img2", "img3", "img4", "img5"]

    def _init_and_verify(self, image_ids):
        """Helper to initialize job and verify initial state."""
        self.manager.initialize_job(image_ids)
        progress = self.manager.get_progress("process")
        assert progress is not None
        self.assertEqual(progress.total, len(image_ids))
        self.assertEqual(progress.remaining, len(image_ids))
        self.assertEqual(progress.processed, 0)
        self.assertEqual(progress.percentage, 0.0)
        self.assertEqual(progress.failed, 0)
        return progress

    def test_initialize_job(self):
        """Test job initialization sets up tracking for all stages."""
        self._init_and_verify(self.image_ids)

        # Verify both stages are initialized
        for stage in self.manager.STAGES:
            progress = self.manager.get_progress(stage)
            assert progress is not None
            self.assertEqual(progress.total, len(self.image_ids))
            self.assertEqual(progress.failed, 0)

    def test_progress_tracking(self):
        """Test progress updates correctly as images are processed."""
        self._init_and_verify(self.image_ids)

        # Process 2 images
        progress = self.manager.update_state({"img1", "img2"}, "process")
        assert progress is not None
        self.assertEqual(progress.remaining, 3)
        self.assertEqual(progress.processed, 2)
        self.assertEqual(progress.percentage, 0.4)

        # Process 2 more images
        progress = self.manager.update_state({"img3", "img4"}, "process")
        assert progress is not None
        self.assertEqual(progress.remaining, 1)
        self.assertEqual(progress.processed, 4)
        self.assertEqual(progress.percentage, 0.8)

        # Process last image
        progress = self.manager.update_state({"img5"}, "process")
        assert progress is not None
        self.assertEqual(progress.remaining, 0)
        self.assertEqual(progress.processed, 5)
        self.assertEqual(progress.percentage, 1.0)

    def test_update_state_concurrent(self):
        """Test that concurrent workers update state correctly without data races."""
        self._init_and_verify(self.image_ids)

        # Three workers process disjoint image sets truly concurrently
        errors: list[BaseException] = []
        with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
            futures = [
                executor.submit(self.manager.update_state, {"img1", "img2"}, "process"),
                executor.submit(self.manager.update_state, {"img3"}, "process"),
                executor.submit(self.manager.update_state, {"img4", "img5"}, "process"),
            ]
            _errors = [f.exception() for f in concurrent.futures.as_completed(futures)]
            errors = [e for e in _errors if e is not None]

        self.assertEqual(errors, [], f"Concurrent workers raised exceptions: {errors}")

        # Final state reflects all concurrent updates
        final = self.manager.get_progress("process")
        assert final is not None
        self.assertEqual(final.processed, 5)
        self.assertEqual(final.remaining, 0)

        # SREM is idempotent: retrying already-processed images doesn't change counts
        progress_retry = self.manager.update_state({"img1", "img2"}, "process")
        assert progress_retry is not None
        self.assertEqual(progress_retry.processed, 5)

    def test_stages_independent(self):
        """Test that different stages track progress independently."""
        self._init_and_verify(self.image_ids)

        # Update process stage
        self.manager.update_state({"img1", "img2"}, "process")
        progress_process = self.manager.get_progress("process")
        assert progress_process is not None
        self.assertEqual(progress_process.remaining, 3)

        # Results stage should still have all images pending
        progress_results = self.manager.get_progress("results")
        assert progress_results is not None
        self.assertEqual(progress_results.remaining, 5)

    def test_empty_job(self):
        """Test handling of job with no images."""
        self.manager.initialize_job([])
        progress = self.manager.get_progress("process")
        assert progress is not None
        self.assertEqual(progress.total, 0)
        self.assertEqual(progress.percentage, 1.0)  # Empty job is 100% complete

    def test_cleanup(self):
        """Test cleanup removes all tracking keys."""
        self._init_and_verify(self.image_ids)

        # Verify keys exist
        progress = self.manager.get_progress("process")
        self.assertIsNotNone(progress)

        # Cleanup
        self.manager.cleanup()

        # Verify keys are gone
        progress = self.manager.get_progress("process")
        self.assertIsNone(progress)

    def test_failed_image_tracking(self):
        """Test basic failed image tracking with no double-counting on retries."""
        self._init_and_verify(self.image_ids)

        # Mark 2 images as failed in process stage
        progress = self.manager.update_state({"img1", "img2"}, "process", failed_image_ids={"img1", "img2"})
        assert progress is not None
        self.assertEqual(progress.failed, 2)

        # Retry same 2 images (fail again) - SADD is idempotent, no double-counting
        progress = self.manager.update_state(set(), "process", failed_image_ids={"img1", "img2"})
        assert progress is not None
        self.assertEqual(progress.failed, 2)

        # Fail a different image
        progress = self.manager.update_state(set(), "process", failed_image_ids={"img3"})
        assert progress is not None
        self.assertEqual(progress.failed, 3)

    def test_failed_and_processed_mixed(self):
        """Test mixed successful and failed processing in same batch."""
        self._init_and_verify(self.image_ids)

        # Process 2 successfully, 2 fail, 1 remains pending
        progress = self.manager.update_state(
            {"img1", "img2", "img3", "img4"}, "process", failed_image_ids={"img3", "img4"}
        )
        assert progress is not None
        self.assertEqual(progress.processed, 4)
        self.assertEqual(progress.failed, 2)
        self.assertEqual(progress.remaining, 1)
        self.assertEqual(progress.percentage, 0.8)

    def test_cleanup_removes_failed_set(self):
        """Test that cleanup removes failed image set."""
        self._init_and_verify(self.image_ids)

        # Add failed images and verify they're tracked
        progress = self.manager.update_state({"img1", "img2"}, "process", failed_image_ids={"img1", "img2"})
        assert progress is not None
        self.assertEqual(progress.failed, 2)

        # Cleanup
        self.manager.cleanup()

        # Verify all state is gone (get_progress returns None when total_key is deleted)
        progress = self.manager.get_progress("process")
        self.assertIsNone(progress)

    def test_update_state_raises_on_redis_error(self):
        """
        A transient Redis failure during update_state must propagate, not be
        swallowed as None. The None return is reserved for the genuine
        "state actually gone" case (see test below). Conflating the two is
        the #1219 bug that escalated transient connection resets into fatal
        job FAILUREs.
        """
        from unittest.mock import MagicMock, patch

        from redis.exceptions import RedisError

        self._init_and_verify(self.image_ids)

        # Replace the pipeline context manager with one whose execute() raises.
        # Everything upstream of execute() is safely called (srem/sadd/scard/get
        # on a pipeline only queue commands; they don't hit the network until
        # execute runs), so we only need to blow up at the execute boundary.
        pipe = MagicMock()
        pipe.execute.side_effect = RedisError("Connection reset by peer")
        fake_redis = MagicMock()
        fake_redis.pipeline.return_value.__enter__.return_value = pipe

        with patch.object(self.manager, "_get_redis", return_value=fake_redis):
            with self.assertRaises(RedisError):
                self.manager.update_state({"img1", "img2"}, "process")

    def test_update_state_returns_none_when_state_genuinely_missing(self):
        """
        When the job's total-images key is actually missing from Redis (job
        was never initialized, cleaned up, or TTL expired), update_state
        returns None. This is the only case that should trigger the
        terminal "state missing" failure path in the caller.
        """
        # Do NOT call initialize_job — the total key doesn't exist.
        progress = self.manager.update_state({"img1", "img2"}, "process")
        self.assertIsNone(progress)

    def test_diagnose_missing_state_when_never_initialized(self):
        """
        Diagnostic string for the "never initialized" case: no keys are
        present under ``job:{id}:*``. Output must still name the Redis host
        and DB so cross-host DB drift is distinguishable from eviction and
        truly-never-initialized state in one log line.
        """
        # initialize_job has NOT been called; nothing under job:123:*.
        diagnosis = self.manager.diagnose_missing_state()
        self.assertIn("redis=", diagnosis)
        self.assertIn("/db", diagnosis)
        self.assertIn("keys_for_job=<none>", diagnosis)

    def test_diagnose_missing_state_lists_present_keys(self):
        """
        Diagnostic string for the partial-cleanup / eviction case: some keys
        remain under ``job:{id}:*`` and their SCARDs should appear so the
        operator can tell "total key evicted but pending sets still present"
        from "nothing here, this DB never saw the job".
        """
        self.manager.initialize_job(self.image_ids)
        # Drop the total key to simulate eviction while pending sets survive.
        redis = self.manager._get_redis()
        redis.delete(self.manager._total_key)

        diagnosis = self.manager.diagnose_missing_state()
        self.assertIn(f"job:{self.job_id}:pending_images:process=SCARD:", diagnosis)
        self.assertNotIn(self.manager._total_key, diagnosis)


class TestSaveResultsRefreshesDeploymentCounts(TestCase):
    """save_results must refresh Deployment cached counts, not just Event counts.

    Reproduces the "Station counts for occurrences and taxa are not always
    getting updated" report: prior to the fix, save_results refreshed
    update_calculated_fields_for_events but never the parent Deployment, so
    deployment.occurrences_count / taxa_count stayed at the pre-job value
    until something else (a manual deployment.save) ran.
    """

    def setUp(self):
        self.project = Project.objects.create(name="Refresh Counts Project")
        self.deployment = Deployment.objects.create(name="d1", project=self.project)
        event_time = datetime.datetime(2026, 4, 16, 22, 0, 0)
        self.event = Event.objects.create(
            project=self.project,
            deployment=self.deployment,
            group_by="2026-04-16",
            start=event_time,
            end=event_time,
        )
        self.image = SourceImage.objects.create(
            deployment=self.deployment,
            project=self.project,
            event=self.event,
            timestamp=event_time,
            path="refresh_counts_test.jpg",
        )
        self.collection = SourceImageCollection.objects.create(project=self.project, name="c")
        self.collection.images.add(self.image)

        self.pipeline = Pipeline.objects.create(name="Refresh Counts Pipeline (Random)")
        self.algorithms = {
            key: get_or_create_algorithm_and_category_map(val) for key, val in ALGORITHM_CHOICES.items()
        }
        self.pipeline.algorithms.set(
            [
                self.algorithms["random-detector"],
                self.algorithms["random-binary-classifier"],
                self.algorithms["random-species-classifier"],
            ]
        )

        self.deployment.update_calculated_fields(save=True)
        self.deployment.refresh_from_db()
        self.assertEqual(self.deployment.occurrences_count, 0)
        self.assertEqual(self.deployment.taxa_count, 0)

    def _fake_results(self):
        detector = ALGORITHM_CHOICES["random-detector"]
        binary_classifier = ALGORITHM_CHOICES["random-binary-classifier"]
        species_classifier = ALGORITHM_CHOICES["random-species-classifier"]
        assert binary_classifier.category_map and species_classifier.category_map

        detection = DetectionResponse(
            source_image_id=self.image.pk,
            bbox=BoundingBox(x1=0.0, y1=0.0, x2=1.0, y2=1.0),
            inference_time=0.1,
            algorithm=AlgorithmReference(name=detector.name, key=detector.key),
            timestamp=self.image.timestamp,
            classifications=[
                ClassificationResponse(
                    classification=binary_classifier.category_map.labels[0],
                    labels=binary_classifier.category_map.labels,
                    scores=[0.95],
                    algorithm=AlgorithmReference(name=binary_classifier.name, key=binary_classifier.key),
                    timestamp=self.image.timestamp,
                    terminal=False,
                ),
                ClassificationResponse(
                    classification=species_classifier.category_map.labels[0],
                    labels=species_classifier.category_map.labels,
                    scores=[0.85],
                    algorithm=AlgorithmReference(name=species_classifier.name, key=species_classifier.key),
                    timestamp=self.image.timestamp,
                    terminal=True,
                ),
            ],
        )
        return PipelineResultsResponse(
            pipeline=self.pipeline.slug,
            algorithms={
                detector.key: detector,
                binary_classifier.key: binary_classifier,
                species_classifier.key: species_classifier,
            },
            total_time=0.01,
            source_images=[SourceImageResponse(id=self.image.pk, url=self.image.path)],
            detections=[detection],
        )

    def test_deployment_counts_refresh_after_save_results(self):
        save_results(self._fake_results())

        self.deployment.refresh_from_db()
        self.assertGreater(
            self.deployment.occurrences_count,
            0,
            "Deployment.occurrences_count should reflect occurrences created by save_results",
        )
        self.assertGreater(
            self.deployment.taxa_count,
            0,
            "Deployment.taxa_count should reflect taxa from occurrences created by save_results",
        )