Add Consent to ODM export (repeating Common event, whitelist-only)

src/edc_cdisc/constants.py

@@ -94,4 +94,27 @@
 UNSCHEDULED_TYPE = "Unscheduled"
 COMMON_TYPE = "Common"
 
+SCREENING_EVENT_CATEGORY = "screening"
+CONSENT_EVENT_CATEGORY = "consent"
+
+# Models whose export field list is an explicit whitelist instead of the admin
+# fieldsets.  Used for documents that hold sensitive data (e.g. consent): only
+# the named, non-PII fields are exported.  The encrypted-field and
+# EXCLUDED_FIELD_NAMES floors STILL apply on top (see iter_whitelist_fields), so
+# an encrypted/excluded field added here cannot be re-exposed — it is dropped
+# and a warning is emitted.  Plaintext-sensitive additions are caught by review;
+# keep these lists tiny and explicit.
+CONSENT_EXPORT_FIELDS: tuple[str, ...] = (
+    "subject_identifier",
+    "consent_datetime",
+    "model_name",
+    "version",
+    "consent_definition_name",
+)
+
+# Static per-model whitelists (model label_lower -> field names).  The consent
+# model is resolved dynamically from settings.SUBJECT_CONSENT_MODEL in
+# get_whitelist_fields, so it is not listed here.
+WHITELIST_FIELDS: dict[str, tuple[str, ...]] = {}
+
 ODM_SCHEMA_PATH = Path(__file__).parent / "odm_schema" / "cdisc-odm-1.3.1" / "ODM1-3-1.xsd"

src/edc_cdisc/serializers/clinical_data_serializer.py

@@ -103,6 +103,7 @@ def build_subject_data(
             if event_element is not None:
                 event_elements.append(event_element)
         event_elements += self.build_common_event_data(subject_identifier)
+        event_elements += self.build_consent_event_data(subject_identifier)
 
         if not event_elements:
             return None
@@ -187,6 +188,33 @@ def build_common_event_data(self, subject_identifier: str) -> list[etree._Elemen
             elements.append(element)
         return elements
 
+    def build_consent_event_data(self, subject_identifier: str) -> list[etree._Element]:
+        """Repeating Common event — one StudyEventData per consent row.
+
+        Consent is unique on subject_identifier + version, so a subject may have
+        several rows; each is emitted with StudyEventRepeatKey = version.  Only
+        the whitelist fields (CONSENT_EXPORT_FIELDS) are serialized — the consent
+        document holds sensitive data.
+        """
+        consent_model = self.get_consent_model()
+        if not consent_model:
+            return []
+        elements: list[etree._Element] = []
+        qs = (
+            django_apps.get_model(consent_model)
+            .objects.filter(subject_identifier=subject_identifier)
+            .order_by("version", "consent_datetime")
+        )
+        for instance in qs:
+            element = etree.Element(
+                "StudyEventData",
+                StudyEventOID=self.common_event_oid(consent_model),
+                StudyEventRepeatKey=str(instance.version),
+            )
+            element.append(self.build_form_data(consent_model, instance))
+            elements.append(element)
+        return elements
+
     @staticmethod
     def get_crfs_for_visit(
         subject_visit: RelatedVisitProtocol,

src/edc_cdisc/serializers/metadata_serializer.py

@@ -10,10 +10,12 @@
 from ..constants import (
     CODELIST,
     COMMON_TYPE,
+    CONSENT_EVENT_CATEGORY,
     FORM,
     ITEM,
     ITEM_GROUP,
     SCHEDULED_TYPE,
+    SCREENING_EVENT_CATEGORY,
     UNSCHEDULED_TYPE,
 )
 from ..exceptions import ProtocolSerializerError
@@ -70,14 +72,14 @@ def build_metadata_version(self) -> etree._Element:
         element = etree.Element("MetaDataVersion")
         element.append(self.get_protocol_element())
 
-        screening_model = self.get_screening_model()
+        subject_level_events = list(self.iter_subject_level_events())
 
         # StudyEventDefs (order among them is free; schedule carried by Alias)
-        for model in self.get_common_models():
-            element.append(self.get_common_event_def_element(model))
-        if screening_model:
+        for model, category, repeating in subject_level_events:
             element.append(
-                self.get_common_event_def_element(screening_model, event_category="screening")
+                self.get_common_event_def_element(
+                    model, event_category=category, repeating=repeating
+                )
             )
         for schedule in self.visit_schedule.schedules.values():
             for visit in schedule.visits.values():
@@ -95,9 +97,7 @@ def build_metadata_version(self) -> etree._Element:
             for visit in schedule.visits.values()
             for crf in [*visit.all_crfs, *visit.all_requisitions]
         }
-        common_models = [*self.get_common_models()]
-        if screening_model:
-            common_models.append(screening_model)
+        common_models = [model for model, _category, _repeating in subject_level_events]
         models = [
             m
             for m in dict.fromkeys([*common_models, *sorted(form_models)])
@@ -130,22 +130,30 @@ def get_protocol_element(self) -> etree._Element:
                 StudyEventOID=self.unscheduled_event_oid(visit_code),
                 Mandatory=NO,
             )
-        for model in self.get_common_models():
+        for model, _category, _repeating in self.iter_subject_level_events():
             etree.SubElement(
                 element,
                 "StudyEventRef",
                 StudyEventOID=self.common_event_oid(model),
                 Mandatory=NO,
             )
-        if screening_model := self.get_screening_model():
-            etree.SubElement(
-                element,
-                "StudyEventRef",
-                StudyEventOID=self.common_event_oid(screening_model),
-                Mandatory=NO,
-            )
         return element
 
+    def iter_subject_level_events(self) -> Iterator[tuple[str, str | None, bool]]:
+        """Yield ``(model, event_category, repeating)`` for each non-scheduled,
+        subject-level Common event: death/offstudy, then screening, then consent.
+
+        Single source for the StudyEventDefs, the definition catalog, and the
+        Protocol StudyEventRefs so they cannot drift.  Consent is repeating
+        (unique on subject_identifier + version); the rest are singletons.
+        """
+        for model in self.get_common_models():
+            yield model, None, False
+        if screening_model := self.get_screening_model():
+            yield screening_model, SCREENING_EVENT_CATEGORY, False
+        if consent_model := self.get_consent_model():
+            yield consent_model, CONSENT_EVENT_CATEGORY, True
+
     @staticmethod
     def append_schedule_alias(element: etree._Element, schedule_name: str) -> None:
         etree.SubElement(element, "Alias", Context="clinicedc.schedule", Name=schedule_name)
@@ -326,13 +334,14 @@ def get_common_event_def_element(
         self,
         model: str,
         event_category: str | None = None,
+        repeating: bool = False,
     ) -> etree._Element:
         verbose_name = str(django_apps.get_model(model)._meta.verbose_name)
         element = etree.Element(
             "StudyEventDef",
             OID=self.common_event_oid(model),
             Name=verbose_name,
-            Repeating=NO,
+            Repeating=YES if repeating else NO,
             Type=COMMON_TYPE,
         )
         etree.SubElement(

src/edc_cdisc/serializers/visit_schedule_serializer.py

@@ -91,6 +91,17 @@ def get_screening_model(self) -> str | None:
         model = getattr(settings, "SUBJECT_SCREENING_MODEL", None)
         return model if model and self._model_exists(model) else None
 
+    def get_consent_model(self) -> str | None:
+        """The consent model (settings.SUBJECT_CONSENT_MODEL).
+
+        A *repeating* Common event keyed by subject_identifier (the allocation
+        identifier consent assigns), with repeat key = ``version`` — one row per
+        consent version.  The consent document holds sensitive data, so only the
+        explicit whitelist (CONSENT_EXPORT_FIELDS) is exported.
+        """
+        model = getattr(settings, "SUBJECT_CONSENT_MODEL", None)
+        return model if model and self._model_exists(model) else None
+
     @staticmethod
     def _model_exists(model: str) -> bool:
         try:

src/edc_cdisc/tests/tests/test_validate_odm.py

@@ -4,7 +4,7 @@
 import time_machine
 from clinicedc_tests.consents import consent_v1
 from clinicedc_tests.helper import Helper
-from clinicedc_tests.models import CrfFour, SubjectScreening
+from clinicedc_tests.models import CrfFour, SubjectConsent, SubjectScreening
 from dateutil.relativedelta import relativedelta
 from django.contrib import admin
 from django.db import models
@@ -28,7 +28,9 @@
     MetadataSerializer,
     SnapshotSerializer,
 )
-from edc_cdisc.utils import is_encrypted_field, validate_odm
+from edc_cdisc.utils import is_encrypted_field, iter_whitelist_fields, validate_odm
+
+CONSENT_OID = "CE.clinicedc_tests.subjectconsent"
 
 NS = {"odm": ODM_NAMESPACE}
 
@@ -176,6 +178,56 @@ def test_screening_event_data_present(self) -> None:
         self.assertIsNotNone(sed)
         self.assertEqual(validate_odm(self._build(SnapshotSerializer)), [])
 
+    def test_consent_event_def_is_repeating_with_category_alias(self) -> None:
+        root = etree.fromstring(self._build(MetadataSerializer))
+        sed = root.find(f".//odm:StudyEventDef[@OID='{CONSENT_OID}']", NS)
+        self.assertIsNotNone(sed)
+        self.assertEqual(sed.get("Type"), "Common")
+        # consent is unique on subject_identifier + version → repeating
+        self.assertEqual(sed.get("Repeating"), "Yes")
+        alias = sed.find("odm:Alias[@Context='clinicedc.event_category']", NS)
+        self.assertIsNotNone(alias)
+        self.assertEqual(alias.get("Name"), "consent")
+
+    def test_consent_event_data_present_with_repeat_key(self) -> None:
+        # enroll_to_baseline created a consent for the enrolled subject
+        root = etree.fromstring(self._build(SnapshotSerializer))
+        sed = root.find(
+            f".//odm:ClinicalData//odm:StudyEventData[@StudyEventOID='{CONSENT_OID}']", NS
+        )
+        self.assertIsNotNone(sed)
+        # repeat key carries the consent version
+        self.assertTrue(sed.get("StudyEventRepeatKey"))
+        self.assertEqual(validate_odm(self._build(SnapshotSerializer)), [])
+
+    def test_consent_export_drops_encrypted_pii(self) -> None:
+        # only the whitelist is exported; encrypted PII (e.g. first_name) must
+        # never appear as an ItemDef or ItemData
+        meta = etree.fromstring(self._build(MetadataSerializer))
+        data = etree.fromstring(self._build(SnapshotSerializer))
+        for root in (meta, data):
+            oids = {
+                e.get("ItemOID") or e.get("OID")
+                for e in root.iter()
+                if (e.get("ItemOID") or e.get("OID"))
+            }
+            self.assertNotIn("I.clinicedc_tests.subjectconsent.first_name", oids)
+            self.assertIn("I.clinicedc_tests.subjectconsent.version", oids)
+
+
+class TestConsentWhitelistFloor(TestCase):
+    def test_floor_drops_encrypted_field_even_if_whitelisted(self) -> None:
+        # first_name is an encrypted PII field; whitelisting it must not export
+        # it — it is dropped (with a warning) while real fields pass through
+        with self.assertWarns(UserWarning):
+            fields = list(
+                iter_whitelist_fields(
+                    SubjectConsent, ["subject_identifier", "first_name", "version"]
+                )
+            )
+        names = {f.name for f in fields}
+        self.assertEqual(names, {"subject_identifier", "version"})
+
 
 class TestEncryptedFieldDetection(TestCase):
     def test_is_encrypted_field(self) -> None:

src/edc_cdisc/utils.py

@@ -5,17 +5,20 @@
 from datetime import date, datetime, time
 from typing import TYPE_CHECKING
 
+from django.conf import settings
 from django.contrib import admin
 from django.contrib.admin.sites import all_sites
 from django.db import models
 from django.http import HttpRequest
 from lxml import etree
 
 from .constants import (
+    CONSENT_EXPORT_FIELDS,
     DJANGO_TO_ODM_DATATYPE,
     EXCLUDED_FIELD_NAMES,
     EXCLUDED_FIELDSET_NAMES,
     ODM_SCHEMA_PATH,
+    WHITELIST_FIELDS,
 )
 from .exceptions import ModelAdminNotFoundError
 
@@ -129,6 +132,44 @@ def is_encrypted_field(field: models.Field) -> bool:
     return isinstance(field, BaseField)
 
 
+def get_whitelist_fields(model_label: str) -> tuple[str, ...] | None:
+    """Explicit export field list for ``model_label``, or ``None`` when the
+    model is driven by its admin fieldsets.
+
+    The consent model (``settings.SUBJECT_CONSENT_MODEL``) holds sensitive data,
+    so only the explicit, non-PII :data:`CONSENT_EXPORT_FIELDS` are exported.
+    Static whitelists for other models live in :data:`WHITELIST_FIELDS`.
+    """
+    if model_label == getattr(settings, "SUBJECT_CONSENT_MODEL", None):
+        return CONSENT_EXPORT_FIELDS
+    return WHITELIST_FIELDS.get(model_label)
+
+
+def iter_whitelist_fields(
+    model_cls: type[models.Model], field_names: Iterable[str]
+) -> Iterator[models.Field]:
+    """Yield the explicitly whitelisted fields of ``model_cls``.
+
+    The same PII floor as everywhere else still applies: an encrypted field or
+    an :data:`EXCLUDED_FIELD_NAMES` field is dropped *even if whitelisted*, with
+    a warning so the mistake is visible.  This is what stops a sensitive field
+    from being re-exposed by being added to a whitelist.  A name that is not a
+    real field raises ``FieldDoesNotExist`` (whitelists must be explicit).
+    """
+    label = model_cls._meta.label_lower
+    for field_name in field_names:
+        field = model_cls._meta.get_field(field_name)
+        if is_encrypted_field(field) or field.name in EXCLUDED_FIELD_NAMES:
+            warnings.warn(
+                f"Whitelisted field {label}.{field_name} is encrypted or "
+                "excluded; not exported.",
+                UserWarning,
+                stacklevel=2,
+            )
+            continue
+        yield field
+
+
 def iter_crf_sections(
     model_cls: type[models.Model],
 ) -> Iterator[tuple[int, str | None, list[models.Field]]]:
@@ -141,9 +182,20 @@ def iter_crf_sections(
     positional index and is kept stable (it is not renumbered when a section
     drops out) so ``ItemGroupOID``\\s do not shift.
 
+    Whitelisted models (see :func:`get_whitelist_fields`) bypass the admin
+    entirely and yield a single unnamed section of their explicit fields — so a
+    model holding sensitive data needs no admin registered and exports only the
+    vetted list.
+
     Note: relation fields (FK / O2O / M2M) are *included* for now (emitted as
     text); proper relation handling is a later step.
     """
+    whitelist = get_whitelist_fields(model_cls._meta.label_lower)
+    if whitelist is not None:
+        fields = list(iter_whitelist_fields(model_cls, whitelist))
+        if fields:
+            yield 1, None, fields
+        return
     for index, (name, opts) in enumerate(get_modeladmin_fieldsets(model_cls), start=1):
         fields = []
         for field_name in opts.get("fields"):