Sync station storage sources from the stations list, per station or all at once

ami/main/api/serializers.py

@@ -184,6 +184,7 @@ class DeploymentListSerializer(DefaultSerializer):
     device = DeviceNestedSerializer(read_only=True)
     research_site = SiteNestedSerializer(read_only=True)
     jobs = JobStatusSerializer(many=True, read_only=True)
+    data_source_connected = serializers.SerializerMethodField()
 
     class Meta:
         model = Deployment
@@ -208,8 +209,19 @@ class Meta:
             "device",
             "research_site",
             "jobs",
+            "data_source_connected",
         ]
 
+    def get_data_source_connected(self, obj: Deployment) -> bool:
+        """
+        Whether the station has a storage source configured.
+
+        The stations list uses this to show the per-row Sync button only where a
+        sync can succeed, and to count how many stations "Sync all" would cover.
+        Reads the foreign key id already on the row, so it adds no query.
+        """
+        return obj.data_source_id is not None
+
     def get_events(self, obj):
         """
         Return URL to the events endpoint filtered by this deployment.

ami/main/api/views.py

@@ -336,30 +336,69 @@ def get_queryset(self) -> QuerySet:
 
         return qs
 
+    @staticmethod
+    def _enqueue_sync_job(deployment: Deployment):
+        """
+        Create and enqueue a ``DataStorageSyncJob`` for a single deployment.
+
+        Shared by the per-row ``sync`` action and the bulk ``sync_all`` action so
+        both build the job the same way (one job per deployment, matching the
+        admin bulk action).
+        """
+        from ami.jobs.models import DataStorageSyncJob, Job
+
+        job = Job.objects.create(
+            name=f"Sync captures for deployment {deployment.pk}",
+            deployment=deployment,
+            project=deployment.project,
+            job_type_key=DataStorageSyncJob.key,
+        )
+        job.enqueue()
+        return job
+
     @action(detail=True, methods=["post"], name="sync")
     def sync(self, _request, pk=None) -> Response:
         """
         Queue a task to sync data from the deployment's data source.
         """
         deployment: Deployment = self.get_object()
         if deployment and deployment.data_source:
-            # queued_task = tasks.sync_source_images.delay(deployment.pk)
-            from ami.jobs.models import DataStorageSyncJob, Job
-
-            job = Job.objects.create(
-                name=f"Sync captures for deployment {deployment.pk}",
-                deployment=deployment,
-                project=deployment.project,
-                job_type_key=DataStorageSyncJob.key,
+            job = self._enqueue_sync_job(deployment)
+            logger.info(
+                f"Syncing captures for deployment {deployment.pk} from {deployment.data_source_uri} in background."
             )
-            job.enqueue()
-            msg = f"Syncing captures for deployment {deployment.pk} from {deployment.data_source_uri} in background."
-            logger.info(msg)
-            assert deployment.project
-            return Response({"job_id": job.pk, "project_id": deployment.project.pk})
+            return Response({"job_id": job.pk, "project_id": deployment.project_id})
         else:
             raise api_exceptions.ValidationError(detail="Deployment must have a data source to sync captures from")
 
+    @action(detail=False, methods=["post"], name="sync-all", url_path="sync-all")
+    def sync_all(self, request) -> Response:
+        """
+        Queue a sync job for every station in the project that has a storage source.
+
+        Enqueues one ``DataStorageSyncJob`` per connected station (separate jobs,
+        not one consolidated job), matching the per-row ``sync`` action and the
+        admin bulk action. Requires the ``project_id`` query parameter and the
+        sync permission on the project.
+        """
+        project = self.get_active_project()
+        if not project:
+            raise api_exceptions.ValidationError(detail="A project_id is required to sync all stations.")
+
+        # ObjectPermission.has_permission() is a no-op and has_object_permission()
+        # only runs for detail actions (via get_object()), so a detail=False action
+        # must check permissions itself. Probe the same sync permission the per-row
+        # action enforces, resolved against the project.
+        if not Deployment(project=project).check_permission(request.user, "sync"):
+            raise api_exceptions.PermissionDenied(
+                detail="You do not have permission to sync stations in this project."
+            )
+
+        deployments = self.get_queryset().filter(data_source__isnull=False)
+        job_ids = [self._enqueue_sync_job(deployment).pk for deployment in deployments]
+        logger.info(f"Queued {len(job_ids)} DataStorageSyncJob(s) for project {project.pk}: {job_ids}")
+        return Response({"job_ids": job_ids, "queued": len(job_ids), "project_id": project.pk})
+
     @action(detail=True, methods=["post"], name="regroup-sessions", url_path="regroup-sessions")
     def regroup_sessions(self, _request, pk=None) -> Response:
         """

ami/main/migrations/0095_grant_sync_deployment_to_mldatamanager.py

@@ -0,0 +1,114 @@
+"""
+Grant the existing ``sync_deployment`` permission to ``MLDataManager`` role
+groups on projects that already exist.
+
+``MLDataManager`` already holds ``run_data_storage_sync_job`` (it can run/retry a
+sync job) but not ``sync_deployment`` (which gates *starting* a sync from a
+station via ``POST /api/v2/deployments/<pk>/sync/`` and the bulk
+``.../sync-all/`` action). This closes that gap so ML data managers can trigger
+syncs, not only manage the resulting jobs. ``ProjectManager`` already has the
+permission and is unaffected.
+
+Permissions here are guardian **object-level** grants on each project, which is
+what ``get_perms(user, project)`` and ``user.has_perm("sync_deployment",
+project)`` read. Adding the permission only to ``group.permissions`` (a global
+Django permission) is not enough — it would not appear in ``get_perms`` and the
+Sync buttons / endpoint would stay inaccessible. So this migration mirrors
+``create_roles_for_project``: it adds the global permission for parity and, more
+importantly, creates the object-level ``GroupObjectPermission`` row per project.
+
+New projects pick this up automatically through ``create_roles_for_project``
+(``MLDataManager`` now includes the permission). A ``post_migrate`` signal
+(``ami.main.apps`` → ``create_roles``) also re-syncs every project's role
+permissions on migrate, so this backfill is belt-and-suspenders; it is kept so
+the grant is explicit and self-contained rather than relying on that signal.
+
+The ``sync_deployment`` permission itself is already defined on
+``Project.Meta.permissions``, so there is no model/schema change here — only a
+data backfill.
+"""
+
+from django.db import migrations
+from django.db.models import Q
+
+
+def _sync_permission(apps):
+    Permission = apps.get_model("auth", "Permission")
+    ContentType = apps.get_model("contenttypes", "ContentType")
+    try:
+        project_ct = ContentType.objects.get(app_label="main", model="project")
+    except ContentType.DoesNotExist:
+        return None, None
+    try:
+        return Permission.objects.get(codename="sync_deployment", content_type=project_ct), project_ct
+    except Permission.DoesNotExist:
+        return None, None
+
+
+def _project_pk_from_group(group):
+    # Group names are "{project_pk}_{project_name}_{RoleName}"; the pk is the
+    # immutable leading segment (the name can contain underscores).
+    try:
+        return int(group.name.split("_", 1)[0])
+    except (ValueError, IndexError):
+        return None
+
+
+def grant_sync_to_mldatamanager(apps, schema_editor):
+    Group = apps.get_model("auth", "Group")
+    GroupObjectPermission = apps.get_model("guardian", "GroupObjectPermission")
+
+    perm, project_ct = _sync_permission(apps)
+    if perm is None:
+        return
+
+    for group in Group.objects.filter(Q(name__endswith="_MLDataManager")):
+        project_pk = _project_pk_from_group(group)
+        if project_pk is None:
+            continue
+        # Global add for parity with create_roles_for_project; the object-level
+        # row below is what get_perms()/has_perm(perm, project) actually read.
+        group.permissions.add(perm)
+        GroupObjectPermission.objects.get_or_create(
+            permission=perm,
+            content_type=project_ct,
+            object_pk=str(project_pk),
+            group=group,
+        )
+
+
+def revoke_sync_from_mldatamanager(apps, schema_editor):
+    Group = apps.get_model("auth", "Group")
+    GroupObjectPermission = apps.get_model("guardian", "GroupObjectPermission")
+
+    perm, project_ct = _sync_permission(apps)
+    if perm is None:
+        return
+
+    # Only touch MLDataManager groups; ProjectManager holds sync_deployment
+    # independently and must keep it.
+    for group in Group.objects.filter(Q(name__endswith="_MLDataManager")):
+        project_pk = _project_pk_from_group(group)
+        if project_pk is None:
+            continue
+        group.permissions.remove(perm)
+        GroupObjectPermission.objects.filter(
+            permission=perm,
+            content_type=project_ct,
+            object_pk=str(project_pk),
+            group=group,
+        ).delete()
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("main", "0094_enable_async_pipeline_workers"),
+        ("guardian", "0002_generic_permissions_index"),
+    ]
+
+    operations = [
+        migrations.RunPython(
+            grant_sync_to_mldatamanager,
+            revoke_sync_from_mldatamanager,
+        ),
+    ]

ami/main/tests.py

@@ -51,7 +51,14 @@
 )
 from ami.tests.fixtures.storage import populate_bucket
 from ami.users.models import User
-from ami.users.roles import BasicMember, Identifier, MLDataManager, ProjectManager, create_roles_for_project
+from ami.users.roles import (
+    BasicMember,
+    Identifier,
+    MLDataManager,
+    ProjectManager,
+    Researcher,
+    create_roles_for_project,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -2900,6 +2907,157 @@ def test_sync_creates_events_and_updates_counts(self):
         logger.info(f"Initial events count: {initial_events_count}, Updated events count: {updated_events.count()}")
 
 
+class TestDeploymentSyncAll(APITestCase):
+    """
+    The bulk "Sync all" endpoint enqueues one sync job per connected station and
+    is gated by the same sync permission the per-row action enforces.
+
+    Pins three guarantees: the ``data_source_connected`` flag the frontend reads,
+    the permission matrix (a ``detail=False`` action must check permissions
+    itself), and that only stations with a storage source get a job.
+    """
+
+    def setUp(self):
+        super().setUp()
+        from unittest import mock
+
+        self.project = Project.objects.create(name="Sync All Project", description="Sync-all tests")
+        create_roles_for_project(self.project)
+
+        self.superuser = User.objects.create_superuser(email="super-syncall@insectai.org", password="password123")
+        self.pm_user = User.objects.create_user(email="pm-syncall@insectai.org", password="password123")
+        self.ml_user = User.objects.create_user(email="ml-syncall@insectai.org", password="password123")
+        self.researcher = User.objects.create_user(email="researcher-syncall@insectai.org", password="password123")
+        self.identifier = User.objects.create_user(email="identifier-syncall@insectai.org", password="password123")
+        self.basic_user = User.objects.create_user(email="basic-syncall@insectai.org", password="password123")
+        self.outsider = User.objects.create_user(email="outsider-syncall@insectai.org", password="password123")
+        ProjectManager.assign_user(self.pm_user, self.project)
+        MLDataManager.assign_user(self.ml_user, self.project)
+        Researcher.assign_user(self.researcher, self.project)
+        Identifier.assign_user(self.identifier, self.project)
+        BasicMember.assign_user(self.basic_user, self.project)
+
+        source = S3StorageSource.objects.create(
+            name="Sync All Source",
+            bucket="test-bucket",
+            access_key="fake-access-key",
+            secret_key="fake-secret-key",
+            project=self.project,
+        )
+        self.connected = Deployment.objects.create(name="Connected", project=self.project, data_source=source)
+        self.unconnected = Deployment.objects.create(name="Unconnected", project=self.project)
+
+        # Keep the endpoint tests off the broker: assert the job rows and enqueue
+        # calls, not real Celery dispatch.
+        patcher = mock.patch("ami.jobs.models.Job.enqueue")
+        self.mock_enqueue = patcher.start()
+        self.addCleanup(patcher.stop)
+
+        self.url = "/api/v2/deployments/sync-all/"
+
+    def test_data_source_connected_field_in_list(self):
+        self.client.force_authenticate(self.superuser)
+        response = self.client.get(f"/api/v2/deployments/?project_id={self.project.pk}")
+        self.assertEqual(response.status_code, 200)
+        flags = {row["id"]: row["data_source_connected"] for row in response.data["results"]}
+        self.assertTrue(flags[self.connected.pk], "Connected station should report data_source_connected=True")
+        self.assertFalse(flags[self.unconnected.pk], "Unconnected station should report data_source_connected=False")
+
+    def test_requires_project_id(self):
+        self.client.force_authenticate(self.superuser)
+        response = self.client.post(self.url)
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_permission_matrix(self):
+        # Sync is allowed for MLDataManager and ProjectManager (and superusers),
+        # not for Researcher, Identifier, BasicMember, or non-members.
+        matrix = [
+            ("superuser", self.superuser, status.HTTP_200_OK),
+            ("ProjectManager", self.pm_user, status.HTTP_200_OK),
+            ("MLDataManager", self.ml_user, status.HTTP_200_OK),
+            ("Researcher", self.researcher, status.HTTP_403_FORBIDDEN),
+            ("Identifier", self.identifier, status.HTTP_403_FORBIDDEN),
+            ("BasicMember", self.basic_user, status.HTTP_403_FORBIDDEN),
+            ("outsider", self.outsider, status.HTTP_403_FORBIDDEN),
+        ]
+        for role_name, user, expected in matrix:
+            with self.subTest(role=role_name):
+                self.client.force_authenticate(user)
+                response = self.client.post(f"{self.url}?project_id={self.project.pk}")
+                self.assertEqual(response.status_code, expected, f"{role_name} got {response.status_code}")
+
+    def test_anonymous_denied(self):
+        self.client.force_authenticate(None)
+        response = self.client.post(f"{self.url}?project_id={self.project.pk}")
+        self.assertIn(response.status_code, (status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN))
+
+    def test_enqueues_one_job_per_connected_station(self):
+        from ami.jobs.models import DataStorageSyncJob
+
+        self.client.force_authenticate(self.pm_user)
+        response = self.client.post(f"{self.url}?project_id={self.project.pk}")
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(response.data["queued"], 1)
+        self.assertEqual(len(response.data["job_ids"]), 1)
+        self.assertEqual(self.mock_enqueue.call_count, 1, "One job should be enqueued")
+
+        jobs = Job.objects.filter(project=self.project, job_type_key=DataStorageSyncJob.key)
+        self.assertEqual(jobs.count(), 1, "Exactly one sync job, for the connected station")
+        self.assertEqual(jobs.first().deployment_id, self.connected.pk, "Unconnected station must be skipped")
+
+
+class TestSyncDeploymentBackfillMigration(APITestCase):
+    """The 0095 backfill grants OBJECT-LEVEL sync_deployment to existing projects'
+    MLDataManager groups, not just a global group permission.
+
+    A global-only grant (``group.permissions.add``) would leave ``get_perms`` and
+    ``has_perm(perm, project)`` — the checks the endpoint and UI actually use —
+    returning False, so the backfill would silently no-op for existing projects.
+    This pins the object-level path.
+    """
+
+    def test_backfill_grants_object_level_sync_to_mldatamanager(self):
+        import importlib
+        from unittest import mock
+
+        from django.apps import apps as global_apps
+        from django.contrib.auth.models import Group
+
+        project = Project.objects.create(name="Sync Backfill Project")
+        create_roles_for_project(project)
+        ml_user = User.objects.create_user(email="ml-backfill@insectai.org", password="password123")
+        MLDataManager.assign_user(ml_user, project)
+        source = S3StorageSource.objects.create(
+            name="Backfill Source",
+            bucket="test-bucket",
+            access_key="fake-access-key",
+            secret_key="fake-secret-key",
+            project=project,
+        )
+        Deployment.objects.create(name="Backfill Station", project=project, data_source=source)
+
+        mldm_group = Group.objects.get(name=f"{project.pk}_{project.name}_MLDataManager")
+
+        # Simulate a project created before MLDataManager gained the permission:
+        # strip the object-level grant so the sync endpoint is denied.
+        remove_perm("sync_deployment", mldm_group, project)
+        self.assertNotIn("sync_deployment", get_perms(mldm_group, project))
+
+        self.client.force_authenticate(ml_user)
+        denied = self.client.post(f"/api/v2/deployments/sync-all/?project_id={project.pk}")
+        self.assertEqual(denied.status_code, status.HTTP_403_FORBIDDEN)
+
+        # Run the backfill and confirm it restores the object-level permission
+        # (a global-only grant would leave get_perms unchanged and the POST 403).
+        migration = importlib.import_module("ami.main.migrations.0095_grant_sync_deployment_to_mldatamanager")
+        migration.grant_sync_to_mldatamanager(global_apps, None)
+
+        self.assertIn("sync_deployment", get_perms(mldm_group, project))
+        with mock.patch("ami.jobs.models.Job.enqueue"):
+            granted = self.client.post(f"/api/v2/deployments/sync-all/?project_id={project.pk}")
+        self.assertEqual(granted.status_code, status.HTTP_200_OK)
+
+
 class TestFineGrainedJobRunPermission(APITestCase):
     def setUp(self):
         super().setUp()