Restriction validation

cds_migrator_kit/rdm/records/load/load.py

@@ -102,6 +102,51 @@ def _save_original_dumped_record(self, entry: MigrationEntry, recid_state, uow):
         )
         uow.register(ModelCommitOp(_original_dump_model))
 
+    @staticmethod
+    def _verify_restricted_files(entry: MigrationEntry, records):
+        """Check that files of a record with legacy file restrictions are restricted.
+
+        If any legacy file of the entry has a non-empty ``status``, every
+        published version of the record must have its files access set to
+        ``restricted``. Runs as the last step of the load, inside the unit of
+        work, so raising rolls the whole record back.
+
+        EP approval entries are excused: they are split into a public and a
+        restricted record anyway, so the public one legitimately has public files.
+
+        :raises ManualImportRequired: with ``critical`` priority on any mismatch.
+        """
+        if entry.get("ep_approval"):
+            return
+
+        recid = entry["record"].recid
+        restricted_keys = [
+            file_data["key"]
+            for version_data in entry.get("versions", {}).values()
+            for file_data in version_data.get("files", {}).values()
+            if file_data.get("metadata", {}).get("status")
+        ]
+        if not restricted_keys:
+            return
+
+        for published in records:
+            record = published._record
+            files_protection = record.access.protection.files
+            if files_protection != "restricted":
+                raise ManualImportRequired(
+                    message=(
+                        "Record has files with legacy restrictions but the "
+                        f"migrated record {record.pid.pid_value} has files "
+                        f"access '{files_protection}' instead of 'restricted'"
+                    ),
+                    field="access",
+                    subfield="files",
+                    stage="load",
+                    recid=recid,
+                    priority="critical",
+                    value=restricted_keys,
+                )
+
     @staticmethod
     def _have_migrated_recid(recid):
         """Check if we have minted `lrecid` pid."""
@@ -298,6 +343,9 @@ def _load(self, entry: MigrationEntry):
                         self.request_load_cls(entry).load(
                             records, self.create_inclusion_request, uow
                         )
+                        # final step of the load: nothing is committed if
+                        # restricted files ended up with a wrong access
+                        self._verify_restricted_files(entry, records)
                     uow.commit()
                 if recid_state_after_load:
                     # only log to disk once the unit of work has actually

cds_migrator_kit/rdm/records/transform/entities/version.py

@@ -97,6 +97,7 @@ def __init__(
         migration_logger,
         raw_file_dumps=None,
         publication_date=None,
+        has_ep_approval=False,
     ):
         """Constructor.
 
@@ -110,12 +111,16 @@ def __init__(
         :param publication_date: Arrow instance (when derived from a file's
             creation date by the caller) or plain ISO date string (metadata-
             only fallback copied from the record's publication date).
+        :param has_ep_approval: the record has an EP approval history, so it
+            is split into a public and a restricted record at load time
+            (by per-file access); a public/restricted file mix is expected.
         """
         self.record_access = record_access
         self.files_dump_dir = files_dump_dir
         self.migration_logger = migration_logger
         self.raw_file_dumps = raw_file_dumps or []
         self.publication_date = publication_date
+        self.has_ep_approval = has_ep_approval
         self.files = None
         self.access = None
 
@@ -152,6 +157,14 @@ def compute_access(self):
         # statuses — can't be represented, so hard-stop for manual review.
         recid = str(file["recid"])
         distinct_statuses = {f["status"] for f in self.raw_file_dumps}
+        restrictions = distinct_statuses - {""}
+        if self.has_ep_approval and restrictions:
+            # EP approval records are split into public/restricted records at
+            # load (ep_split), which sorts files by their own status - so a
+            # public + single-restriction mix is excused. The restriction is
+            # kept as `meta` for the restricted split's grants.
+            if len(restrictions) == 1:
+                distinct_statuses = restrictions
         if len(distinct_statuses) > 1:
             raise ManualImportRequired(
                 message=(

cds_migrator_kit/rdm/records/transform/transform.py

@@ -184,7 +184,7 @@ def _transform(self, raw_dump_entry) -> Optional[MigrationEntry]:
             record = self._record(raw_dump_entry, dump)
 
             if record:
-                versions = self._versions(raw_dump_entry, record)
+                versions = self._versions(raw_dump_entry, record, bool(ep_approval))
                 # RecordParent is the last entity to touch dojson_entry -
                 # it pops submitter/communities/access_grants - so the
                 # forgotten-keys check must run after it.
@@ -242,13 +242,14 @@ def _record(self, raw_dump_entry, dump) -> RecordEntry:
         entry_builder.build()
         return entry_builder
 
-    def _versions(self, raw_dump_entry, record: RecordEntry):
+    def _versions(self, raw_dump_entry, record: RecordEntry, has_ep_approval=False):
         return RecordVersionsTransform(
             raw_dump_entry=raw_dump_entry,
             record=record,
             files_dump_dir=self.files_dump_dir,
             plots=self.plots,
             migration_logger=self.migration_logger,
+            has_ep_approval=has_ep_approval,
         ).build()
 
     def _load_migrated_recids(self):

cds_migrator_kit/rdm/records/transform/transform_versions.py

@@ -33,7 +33,15 @@ class RecordVersionsTransform:
     legacy file revision - see ``_is_external_doi()``.
     """
 
-    def __init__(self, raw_dump_entry, record, files_dump_dir, plots, migration_logger):
+    def __init__(
+        self,
+        raw_dump_entry,
+        record,
+        files_dump_dir,
+        plots,
+        migration_logger,
+        has_ep_approval=False,
+    ):
         """Constructor.
 
         :param raw_dump_entry: the original harvested legacy entry (needs
@@ -43,12 +51,15 @@ def __init__(self, raw_dump_entry, record, files_dump_dir, plots, migration_logg
         :param files_dump_dir: local EOS mirror root for file content.
         :param plots: whether to keep Plot-type files.
         :param migration_logger: for skip/restriction logging.
+        :param has_ep_approval: record is split at load by EP approval, so
+            mixed public/restricted files in a version are allowed.
         """
         self.raw_dump_entry = raw_dump_entry
         self.record = record
         self.files_dump_dir = files_dump_dir
         self.plots = plots
         self.migration_logger = migration_logger
+        self.has_ep_approval = has_ep_approval
 
     def build(self):
         """Group legacy files by version, build + carry files forward, return."""
@@ -114,6 +125,7 @@ def build(self):
                 migration_logger=self.migration_logger,
                 publication_date=publication_dates[version_number],
                 raw_file_dumps=list(cumulative_raw_dumps.values()),
+                has_ep_approval=self.has_ep_approval,
             ).build()
 
         if not versions: