Enforce datasource publication write scope

src/govoplan_datasources/backend/service.py

@@ -1220,13 +1220,7 @@ def _publication_context(
     session: object,
     principal: object,
 ) -> tuple[Session, ApiPrincipal]:
-    if not isinstance(session, Session):
-        raise TypeError("Datasource providers require a SQLAlchemy session.")
-    if not isinstance(principal, ApiPrincipal):
-        raise DatasourceAccessError("A tenant API principal is required.")
-    if not principal.tenant_id:
-        raise DatasourceAccessError("A tenant API principal is required.")
-    return session, principal
+    return _context(session, principal, SOURCE_WRITE_SCOPE)
 
 
 def _validate_columns(

tests/test_lifecycle.py

@@ -352,7 +352,7 @@ class DatasourceLifecycleTests(unittest.TestCase):
         )
 
     def test_producer_publication_is_atomic_idempotent_and_addressable(self) -> None:
-        producer = principal(scopes=())
+        producer = principal(scopes=(SOURCE_WRITE_SCOPE,))
         request = DatasourcePublicationRequest(
             producer_module="dataflow",
             producer_run_ref="dataflow-run:run-1",
@@ -405,7 +405,7 @@ class DatasourceLifecycleTests(unittest.TestCase):
         )
 
     def test_publication_idempotency_key_rejects_different_output(self) -> None:
-        producer = principal(scopes=())
+        producer = principal(scopes=(SOURCE_WRITE_SCOPE,))
         base = DatasourcePublicationRequest(
             producer_module="dataflow",
             producer_run_ref="dataflow-run:run-1",
@@ -430,6 +430,21 @@ class DatasourceLifecycleTests(unittest.TestCase):
                 ),
             )
 
+    def test_publication_requires_source_write_scope(self) -> None:
+        with self.assertRaises(DatasourceAccessError):
+            self.provider.publish_rows(
+                self.session,
+                principal(scopes=()),
+                request=DatasourcePublicationRequest(
+                    producer_module="dataflow",
+                    producer_run_ref="dataflow-run:run-1",
+                    idempotency_key="denied",
+                    name="Result",
+                    source_name="result",
+                    rows=({"id": 1},),
+                ),
+            )
+
 
 if __name__ == "__main__":
     unittest.main()