Merge branch 'release/0.3.21' into main

edc_reference/model_mixins.py

@@ -11,17 +11,17 @@ class ReferenceModelMixin(models.Model):
     reference_deleter_cls = ReferenceDeleter
     reference_updater_cls = ReferenceUpdater
 
-    def update_reference_on_save(self):
+    def update_reference_on_save(self) -> None:
         # see also signal in edc-metadata
         self.model_reference_validate()
         if self.reference_updater_cls:
             self.reference_updater_cls(model_obj=self)
 
     @property
-    def reference_name(self):
+    def reference_name(self) -> str:
         return self._meta.label_lower
 
-    def model_reference_validate(self):
+    def model_reference_validate(self) -> None:
         if "panel" in [f.name for f in self._meta.get_fields()]:
             raise ReferenceModelMixinError(
                 "Detected field panel. Is this a requisition?. "
@@ -35,10 +35,10 @@ class Meta:
 
 class RequisitionReferenceModelMixin(ReferenceModelMixin, models.Model):
     @property
-    def reference_name(self):
+    def reference_name(self) -> str:
         return f"{self._meta.label_lower}.{self.panel.name}"
 
-    def model_reference_validate(self):
+    def model_reference_validate(self) -> None:
         if "panel" not in [f.name for f in self._meta.get_fields()]:
             raise ReferenceModelMixinError(
                 "Did not detect field panel. Is this a CRF?. "

edc_reference/reference/reference_updater.py

@@ -1,6 +1,24 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
 from ..site_reference import site_reference_configs
 from .reference_getter import ReferenceGetter
 
+if TYPE_CHECKING:
+    from edc_model.models import BaseUuidModel
+    from edc_visit_tracking.model_mixins import VisitTrackingCrfModelMixin
+
+    from ..model_mixins import ReferenceModelMixin
+
+    class AnyNonCrfModel(ReferenceModelMixin, BaseUuidModel):
+        pass
+
+    class AnyCrfModel(VisitTrackingCrfModelMixin, ReferenceModelMixin, BaseUuidModel):
+        """Any CRF or Requisition model"""
+
+        pass
+
 
 class ReferenceFieldNotFound(Exception):
     pass
@@ -13,47 +31,60 @@ class ReferenceUpdaterModelError(Exception):
 class ReferenceUpdater:
     """Updates or creates each reference model instance; one for
     each field in `edc_reference` for this model_obj.
+
+    Will fail with a proxy model.
     """
 
     getter_cls = ReferenceGetter
 
-    def __init__(self, model_obj=None):
-        reference_fields = site_reference_configs.get_fields(name=model_obj.reference_name)
+    def __init__(self, model_obj: AnyNonCrfModel | AnyCrfModel | ReferenceModelMixin = None):
+        self.model_obj = model_obj
+        if self.model_obj._meta.proxy:
+            raise ReferenceUpdaterModelError(
+                "Not allowed. ReferenceUpdater does not accept proxy models. "
+                f"Got `{self.model_obj.reference_name}`. "
+            )
+        reference_fields = site_reference_configs.get_fields(
+            name=self.model_obj.reference_name
+        )
         # loop through fields and update or create each
         # reference model instance
         for field_name in reference_fields:
             try:
                 field_obj = [
-                    fld for fld in model_obj._meta.get_fields() if fld.name == field_name
+                    fld for fld in self.model_obj._meta.get_fields() if fld.name == field_name
                 ][0]
             except IndexError:
                 raise ReferenceFieldNotFound(
                     f"Reference field not found on model. Got '{field_name}'. "
-                    f"See reference config for {model_obj.reference_name}. "
+                    f"See reference config for {self.model_obj.reference_name}. "
                     f"Model fields are "
-                    f"{[fld.name for fld in model_obj._meta.get_fields()]}"
+                    f"{[fld.name for fld in self.model_obj._meta.get_fields()]}"
                 )
             reference_getter = self.getter_cls(
-                model_obj=model_obj, field_name=field_name, create=True
+                model_obj=self.model_obj, field_name=field_name, create=True
             )
             if field_obj.name == "report_datetime":
                 try:
-                    value = getattr(model_obj.related_visit, field_name)
+                    value = getattr(self.model_obj.related_visit, field_name)
                 except AttributeError:
-                    value = getattr(model_obj, field_name)
+                    value = getattr(self.model_obj, field_name)
             else:
-                value = getattr(model_obj, field_name)
+                value = getattr(self.model_obj, field_name)
             try:
                 value = value.pk
             except AttributeError:
                 internal_type = field_obj.get_internal_type()
                 related_name = None
             else:
                 internal_type = "UUIDField"
-                related_name = getattr(model_obj, field_name)._meta.label_lower
+                related_name = getattr(self.model_obj, field_name)._meta.label_lower
             reference_getter.reference_obj.update_value(
                 internal_type=internal_type,
                 value=value,
                 related_name=related_name,
             )
             reference_getter.reference_obj.save()
+
+    def __repr__(self):
+        return f"ReferenceUpdater({self.model_obj.__class__})"

edc_reference/reference_model_config.py

@@ -20,26 +20,27 @@ class ReferenceFieldAlreadyAdded(Exception):
 class ReferenceModelConfig:
     reference_model = "edc_reference.reference"
 
-    def __init__(self, name=None, fields=None):
+    def __init__(self, name: str = None, fields: list[str] = None):
         """
         Keywords:
             name = app_label.model_name for CRFs
             name = app_label.model_name.panel for Requisitions
+
+        Note: `app_label.model_name` may not be a proxy model
         """
 
         if not fields:
             raise ReferenceFieldValidationError("No fields declared.")
-        self.field_names = list(set(fields))
+        self.field_names: list[str] = list(set(fields))
         self.field_names.sort()
-        self.name = name.lower()
-        self.model = ".".join(name.split(".")[:2])
-
+        self.name: str = name.lower()
+        self.model: str = ".".join(name.split(".")[:2])
         if len(fields) != len(self.field_names):
             raise ReferenceDuplicateField(
                 f"Duplicate field detected. Got {fields}. See '{self.name}'"
             )
 
-    def add_fields(self, fields=None):
+    def add_fields(self, fields: list[str] = None) -> None:
         for field_name in fields:
             if field_name in self.field_names:
                 raise ReferenceFieldAlreadyAdded(
@@ -49,7 +50,7 @@ def add_fields(self, fields=None):
         self.field_names = list(set(self.field_names))
         self.field_names.sort()
 
-    def remove_fields(self, fields=None):
+    def remove_fields(self, fields=None) -> None:
         for field in fields:
             self.field_names.remove(field)
         self.field_names = list(set(self.field_names))
@@ -68,6 +69,11 @@ def check(self):
             raise ReferenceModelValidationError(
                 f"Invalid app label or model name. Got {self.model}. See {repr(self)}."
             )
+        if model_cls._meta.proxy:
+            raise ReferenceModelValidationError(
+                f"Not allowed. Registered model is a proxy model. Got {self.model}. "
+                f"See {repr(self)}."
+            )
         model_field_names = [fld.name for fld in model_cls._meta.get_fields()]
         for field_name in self.field_names:
             if field_name not in model_field_names:

edc_reference/site_reference.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import copy
 import sys
 
@@ -53,12 +55,12 @@ class SiteReference:
     reference_updater = ReferenceUpdater()
 
     def __init__(self):
-        self.registry = {}
-        self.loaded = False
-        self.registered_from_visit_schedules = False
-        self.registered_visit_model = False
+        self.registry: dict[str, ReferenceModelConfig] = {}
+        self.loaded: bool = False
+        self.registered_from_visit_schedules: bool = False
+        self.registered_visit_model: bool = False
 
-    def register(self, reference_config=None):
+    def register(self, reference_config: ReferenceModelConfig = None) -> None:
         if reference_config.name in self.registry:
             raise AlreadyRegistered(
                 f"Reference fields have already been registered. "
@@ -67,23 +69,23 @@ def register(self, reference_config=None):
         self.registry.update({reference_config.name: reference_config})
         self.loaded = True
 
-    def reregister(self, reference=None):
+    def reregister(self, reference: ReferenceModelConfig = None) -> None:
         if reference.name not in self.registry:
             raise ReferenceConfigNotRegistered(
                 f"Reregister failed. Reference model configuration has not "
                 f"been registered. Got {reference.name}"
             )
         self.registry.update({reference.name: reference})
 
-    def unregister(self, name=None):
+    def unregister(self, name: str = None) -> None:
         if name not in self.registry:
             raise ReferenceConfigNotRegistered(
                 f"Unregister failed. Reference model configuration has not "
                 f"been registered. Got {name}"
             )
         self.registry = {k: v for k, v in self.registry.items() if k != name}
 
-    def get_config(self, name=None):
+    def get_config(self, name: str = None) -> ReferenceModelConfig:
         try:
             reference_config = self.registry.get(name.lower())
         except AttributeError:
@@ -95,19 +97,19 @@ def get_config(self, name=None):
             )
         return reference_config
 
-    def get_fields(self, name=None):
+    def get_fields(self, name=None) -> list[str]:
         """Returns a list of fields associated with the
         reference configuration of "model".
         """
         return self.get_config(name=name).field_names
 
-    def get_reference_model(self, name=None):
+    def get_reference_model(self, name=None) -> str:
         """Returns the reference model associated with the
         reference configuration of "model".
         """
         return self.get_config(name=name).reference_model
 
-    def check(self):
+    def check(self) -> dict:
         """Validates the reference data for all classes in the
         registry.
         """

edc_reference/tests/tests/test_case.py

@@ -1,6 +1,7 @@
 from dateutil.relativedelta import relativedelta
 from django.contrib.sites.models import Site
 from django.test import TestCase as BaseTestCase
+from edc_appointment.constants import INCOMPLETE_APPT
 from edc_appointment.models import Appointment
 from edc_facility import import_holidays
 from edc_utils import get_utcnow
@@ -43,3 +44,5 @@ def setUp(self):
                     schedule_name=appointment.schedule_name,
                 )
             )
+            appointment.appt_status = INCOMPLETE_APPT
+            appointment.save()

edc_reference/tests/tests/test_populater.py

@@ -4,6 +4,7 @@
 from django.contrib.sites.models import Site
 from django.core.exceptions import ObjectDoesNotExist
 from django.test import TestCase
+from edc_appointment.constants import INCOMPLETE_APPT
 from edc_appointment.models import Appointment
 from edc_utils import get_utcnow
 from edc_visit_tracking.constants import SCHEDULED
@@ -66,6 +67,9 @@ def setUp(self):
             site=Site.objects.get_current(),
         )
 
+        appointment1.appt_status = INCOMPLETE_APPT
+        appointment1.save()
+
         self.subject_visit2 = SubjectVisit.objects.create(
             appointment=appointment2,
             report_datetime=appointment2.appt_datetime,

edc_reference/tests/tests/test_reference.py

@@ -1,14 +1,17 @@
-from datetime import date
-from decimal import Decimal
+from datetime import date, timedelta
 
 from dateutil.relativedelta import relativedelta
 from django.contrib.sites.models import Site
 from django.core.exceptions import ObjectDoesNotExist
 from django.test import TestCase
+from edc_appointment.constants import INCOMPLETE_APPT
 from edc_appointment.models import Appointment
 from edc_lab.models.panel import Panel
+from edc_registration.models import RegisteredSubject
 from edc_utils import get_utcnow
+from edc_visit_schedule import site_visit_schedules
 from edc_visit_tracking.constants import SCHEDULED
+from edc_visit_tracking.models import SubjectVisit
 
 from edc_reference.models import Reference, ReferenceFieldDatatypeNotFound
 from edc_reference.reference import (
@@ -28,9 +31,9 @@
 from reference_app.models import (
     CrfOne,
     CrfWithUnknownDatatype,
+    OnSchedule,
     SubjectConsent,
     SubjectRequisition,
-    SubjectVisit,
     TestModel,
 )
 from reference_app.reference_model_configs import register_configs
@@ -44,41 +47,51 @@ def setUp(self):
         self.panel_wb = Panel.objects.create(name="wb")
         self.subject_identifier = "123456789"
         self.site = Site.objects.get_current()
+        site_visit_schedules._registry = {}
+        site_visit_schedules.register(visit_schedule)
+        site_reference_configs.register_from_visit_schedule(
+            visit_models={"edc_appointment.appointment": "edc_visit_tracking.subjectvisit"}
+        )
 
         dte = get_utcnow()
 
+        RegisteredSubject.objects.create(subject_identifier=self.subject_identifier)
+
         SubjectConsent.objects.create(
             consent_datetime=dte,
             subject_identifier=self.subject_identifier,
             identity="012345678",
             confirm_identity="012345678",
             site=Site.objects.get_current(),
         )
-
-        appointment = Appointment.objects.create(
+        OnSchedule.objects.create(
             subject_identifier=self.subject_identifier,
-            appt_datetime=dte,
-            timepoint_datetime=dte,
-            visit_schedule_name=visit_schedule.name,
-            schedule_name="schedule",
-            visit_code="1000",
-            timepoint=Decimal("1.0"),
-            site=Site.objects.get_current(),
+            report_datetime=get_utcnow() - timedelta(days=15),
+            onschedule_datetime=get_utcnow() - timedelta(days=15),
         )
 
+        appointment = Appointment.objects.get(visit_code="1000")
+
         self.subject_visit = SubjectVisit.objects.create(
             appointment=appointment,
+            report_datetime=appointment.appt_datetime,
             reason=SCHEDULED,
             site=Site.objects.get_current(),
+            visit_schedule_name=appointment.visit_schedule_name,
+            schedule_name=appointment.schedule_name,
+            visit_code="1000",
+            visit_code_sequence=0,
         )
+        appointment.appt_status = INCOMPLETE_APPT
+        appointment.save()
 
     def reset_site_reference_configs(self):
         site_reference_configs.registry = {}
         site_reference_configs.loaded = False
         site_reference_configs.registered_from_visit_schedules = False
 
         subjectvisit_reference = ReferenceModelConfig(
-            name="reference_app.subjectvisit", fields=["report_datetime", "visit_code"]
+            name="edc_visist_tracking.subjectvisit", fields=["report_datetime", "visit_code"]
         )
 
         self.testmodel_reference = ReferenceModelConfig(

edc_reference/tests/tests/test_refset.py

@@ -2,10 +2,11 @@
 from django.contrib.sites.models import Site
 from edc_constants.constants import NEG, POS
 from edc_utils import get_utcnow
+from edc_visit_tracking.models import SubjectVisit
 
 from edc_reference.models import Reference
 from edc_reference.refsets import Refset, RefsetError
-from reference_app.models import CrfOne, SubjectVisit
+from reference_app.models import CrfOne
 
 from .test_case import TestCase