Skip to content

Commit

Permalink
Add various checks around conflicting visit CRFs
Browse files Browse the repository at this point in the history
Checks include creating errors for the following conditions:
- visit CRF required + same model in PRNs
- proxy CRF required + same proxy in PRNs
- proxy root model appears in visit, includes:
    - proxy root CRF required in single visit (i.e. proxy_root, plus its
     child proxy as CRF/PRN)
    - required proxy CRF + proxy root PRN

Following allowed:
- concrete CRF not required + in PRNs (e.g. Chest Xray)
- proxy CRF not required + proxy in PRNs (e.g. Chest Xray)

Following allowed if explicitly flagged with `shares_proxy_root=True`:
- different proxys pointing to proxy root (parent) model in collection
(e.g. proxies for screening Part 1, 2 and 3 - all pointing to same
proxy root)

Also:
- add (and use throughout) get_duplicates() function
- standardise form/crf/requisition collection error messaged
  • Loading branch information
JonathanWillitts committed Jan 11, 2024
1 parent f07589e commit 2820508
Show file tree
Hide file tree
Showing 14 changed files with 1,578 additions and 56 deletions.
5 changes: 5 additions & 0 deletions .coveragerc
Original file line number Diff line number Diff line change
Expand Up @@ -2,3 +2,8 @@
include = edc_visit_schedule/*
omit = edc_visit_schedule/tests/*,edc_visit_schedule/migrations/*,edc_visit_schedule/wsgi.py
branch = 1

[report]
exclude_lines =
pragma: no cover
if TYPE_CHECKING:
2 changes: 1 addition & 1 deletion README.rst
Original file line number Diff line number Diff line change
Expand Up @@ -126,7 +126,7 @@ The ``site_visit_schedules`` has a number of methods to help query the visit sch
OnSchedule and OffSchedule models
=================================

Two models mixins are required for the the on-schedule and off-schedule models, ``OnScheduleModelMixin`` and ``OffScheduleModelMixin``. OnSchedule/OffSchedule models are specific to a ``schedule``. The ``visit_schedule_name`` and ``schedule_name`` are declared on the model's ``Meta`` class attribute ``visit_schedule_name``.
Two models mixins are required for the on-schedule and off-schedule models, ``OnScheduleModelMixin`` and ``OffScheduleModelMixin``. OnSchedule/OffSchedule models are specific to a ``schedule``. The ``visit_schedule_name`` and ``schedule_name`` are declared on the model's ``Meta`` class attribute ``visit_schedule_name``.

For example:

Expand Down
3 changes: 2 additions & 1 deletion edc_visit_schedule/apps.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
from django.db.models.signals import post_migrate

from .site_visit_schedules import site_visit_schedules
from .system_checks import visit_schedule_check
from .system_checks import check_form_collections, visit_schedule_check

style = color_style()

Expand All @@ -33,3 +33,4 @@ def ready(self):
site_visit_schedules.autodiscover()
sys.stdout.write(f" Done loading {self.verbose_name}.\n")
register(visit_schedule_check)
register(check_form_collections)
171 changes: 170 additions & 1 deletion edc_visit_schedule/system_checks.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,17 @@
from django.core.checks import Warning
from __future__ import annotations

from collections import defaultdict
from typing import TYPE_CHECKING

from django.core.checks import Error, Warning
from django.db import models

from .site_visit_schedules import site_visit_schedules
from .utils import get_duplicates
from .visit import CrfCollection

if TYPE_CHECKING:
from .visit import Visit


def visit_schedule_check(app_configs, **kwargs):
Expand All @@ -15,3 +26,161 @@ def visit_schedule_check(app_configs, **kwargs):
for result in results:
errors.append(Warning(result, id=f"edc_visit_schedule.{key}"))
return errors


def check_form_collections(app_configs, **kwargs):
errors = []
for visit_schedule in site_visit_schedules.visit_schedules.values():
for schedule in visit_schedule.schedules.values():
for visit in schedule.visits.values():
for visit_crf_collection, visit_type in [
(visit.crfs, "Scheduled"),
(visit.crfs_unscheduled, "Unscheduled"),
(visit.crfs_missed, "Missed"),
]:
if duplicate_required_models_err := check_duplicate_required_models(
visit=visit,
visit_crf_collection=visit_crf_collection,
visit_type=visit_type,
):
errors.append(duplicate_required_models_err)

if proxy_root_alongside_child_err := check_proxy_root_alongside_child(
visit=visit,
visit_crf_collection=visit_crf_collection,
visit_type=visit_type,
):
errors.append(proxy_root_alongside_child_err)

if same_proxy_root_err := check_multiple_proxies_same_proxy_root(
visit=visit,
visit_crf_collection=visit_crf_collection,
visit_type=visit_type,
):
errors.append(same_proxy_root_err)

return errors


def check_duplicate_required_models(
visit: Visit,
visit_crf_collection: CrfCollection,
visit_type: str,
) -> Error | None:
if duplicates := get_duplicates(
list_items=[
*get_required_models(collection=visit_crf_collection),
*get_models(collection=visit.crfs_prn),
]
):
return Error(
"Required model class appears more than once for a visit. "
f"Got '{visit}' '{visit_type}' visit Crf collection. "
f"Duplicates {[d._meta.label_lower for d in duplicates]}",
id="edc_visit_schedule.002",
)


def check_proxy_root_alongside_child(
visit: Visit,
visit_crf_collection: CrfCollection,
visit_type: str,
) -> Error | None:
all_models = get_models(collection=visit_crf_collection) + get_models(
collection=visit.crfs_prn
)
all_proxy_models = get_proxy_models(collection=visit_crf_collection) + get_proxy_models(
collection=visit.crfs_prn
)

if child_proxies_alongside_proxy_roots := [
m for m in all_proxy_models if get_proxy_root_model(m) in all_models
]:
proxy_root_child_pairs = [
(
f"proxy_root_model={get_proxy_root_model(proxy)._meta.label_lower}",
f"proxy_model={proxy._meta.label_lower}",
)
for proxy in child_proxies_alongside_proxy_roots
]
return Error(
"Proxy root model class appears alongside associated child "
"proxy for a visit. "
f"Got '{visit}' '{visit_type}' visit Crf collection. "
f"Proxy root/child models: {proxy_root_child_pairs=}",
id="edc_visit_schedule.003",
)


def check_multiple_proxies_same_proxy_root(
visit: Visit,
visit_crf_collection: CrfCollection,
visit_type: str,
) -> Error | None:
all_proxy_models = get_proxy_models(collection=visit_crf_collection) + get_proxy_models(
collection=visit.crfs_prn
)
proxy_roots = [get_proxy_root_model(m) for m in all_proxy_models]

if duplicate_proxy_roots := get_duplicates(proxy_roots):
# Determine if there is a clash with multiple proxies with the same proxy root model
non_shared_proxy_root_models = get_proxy_models(
collection=CrfCollection(
*[f for f in visit_crf_collection if not f.shares_proxy_root]
)
) + get_proxy_models(
collection=CrfCollection(*[f for f in visit.crfs_prn if not f.shares_proxy_root])
)

proxy_root_clashes = defaultdict(list)
for pm in non_shared_proxy_root_models:
proxy_root_model = get_proxy_root_model(pm)
if proxy_root_model in duplicate_proxy_roots:
proxy_root_clashes[proxy_root_model._meta.label_lower].append(
pm._meta.label_lower
)

# Prepare error message and add
if proxy_root_clashes:
shared_proxy_root_models = get_proxy_models(
collection=CrfCollection(
*[f for f in visit_crf_collection if f.shares_proxy_root]
)
) + get_proxy_models(
collection=CrfCollection(*[f for f in visit.crfs_prn if f.shares_proxy_root])
)
for pm in shared_proxy_root_models:
proxy_root_model = get_proxy_root_model(pm)
if proxy_root_model in duplicate_proxy_roots:
clash_desc = proxy_root_clashes[proxy_root_model._meta.label_lower]
clash_desc.append(pm._meta.label_lower)
clash_desc.sort()

return Error(
"Multiple proxies with same proxy root model appear for "
"a visit. If this is intentional, consider using "
"`shares_proxy_root` argument when defining Crf. "
f"Got '{visit}' '{visit_type}' visit Crf collection. "
f"Proxy root/child models: {proxy_root_clashes=}",
id="edc_visit_schedule.004",
)


def get_models(collection: CrfCollection) -> list[models.Model]:
return [f.model_cls for f in collection]


def get_required_models(collection: CrfCollection) -> list[models.Model]:
return [f.model_cls for f in collection if f.required]


def get_proxy_models(collection: CrfCollection) -> list[models.Model]:
return [f.model_cls for f in collection if f.model_cls._meta.proxy]


def get_proxy_root_model(proxy_model: models.Model) -> models.Model | None:
"""Returns proxy's root (concrete) model if `proxy_model` is a
proxy model, else returns None.
"""
if proxy_model._meta.proxy:
return proxy_model._meta.concrete_model
59 changes: 59 additions & 0 deletions edc_visit_schedule/tests/tests/test_crf.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,59 @@
from django.test import TestCase

from edc_visit_schedule.visit import Crf
from edc_visit_schedule.visit.crf import CrfModelNotProxyModelError


class TestCrfCollection(TestCase):
def test_crf_ok(self):
try:
Crf(show_order=1, model="visit_schedule_app.CrfOne")
except Exception as e:
self.fail(f"Exception unexpectedly raised. Got {e}")

try:
Crf(show_order=1, model="visit_schedule_app.CrfOneProxyOne")
except Exception as e:
self.fail(f"Exception unexpectedly raised. Got {e}")

try:
Crf(show_order=1, model="visit_schedule_app.CrfTwo")
except Exception as e:
self.fail(f"Exception unexpectedly raised. Got {e}")

try:
Crf(show_order=1, model="visit_schedule_app.CrfThree")
except Exception as e:
self.fail(f"Exception unexpectedly raised. Got {e}")

def test_proxy_child_crf_with_allow_proxy_parent_clash_ok(self):
try:
Crf(
show_order=1,
model="visit_schedule_app.CrfOneProxyOne",
shares_proxy_root=True,
)
except Exception as e:
self.fail(f"Exception unexpectedly raised. Got {e}")

def test_proxy_root_crf_with_allow_proxy_parent_clash_raises(self):
with self.assertRaises(CrfModelNotProxyModelError) as cm:
Crf(show_order=1, model="visit_schedule_app.CrfOne", shares_proxy_root=True)
self.assertIn(
"Invalid use of `shares_proxy_root=True`. CRF model is not a proxy model.",
str(cm.exception),
)
self.assertIn("visit_schedule_app.crfone", str(cm.exception))

def test_non_proxy_crf_with_allow_proxy_parent_clash_raises(self):
with self.assertRaises(CrfModelNotProxyModelError) as cm:
Crf(
show_order=1,
model="visit_schedule_app.CrfThree",
shares_proxy_root=True,
)
self.assertIn(
"Invalid use of `shares_proxy_root=True`. CRF model is not a proxy model.",
str(cm.exception),
)
self.assertIn("visit_schedule_app.crfthree", str(cm.exception))
41 changes: 41 additions & 0 deletions edc_visit_schedule/tests/tests/test_crf_collection.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,41 @@
from django.test import TestCase

from edc_visit_schedule.visit import Crf, CrfCollection, FormsCollectionError


class TestCrfCollection(TestCase):
def test_crf_collection_ok(self):
crfs = [
Crf(show_order=100, model="x.one"),
Crf(show_order=200, model="x.two"),
Crf(show_order=300, model="x.three"),
]
try:
CrfCollection(*crfs)
except FormsCollectionError:
self.fail("FormsCollectionError unexpectedly raised")

def test_crf_collection_with_duplicate_ordering_raises(self):
crfs = [
Crf(show_order=100, model="x.one"),
Crf(show_order=200, model="x.two"),
Crf(show_order=100, model="x.three"),
]
with self.assertRaises(FormsCollectionError) as cm:
CrfCollection(*crfs)
self.assertIn(
'CrfCollection "show order" must be a unique sequence.',
str(cm.exception),
)
self.assertIn("Duplicates [100].", str(cm.exception))

def test_crf_collection_with_duplicate_models_raises(self):
crfs = [
Crf(show_order=100, model="x.one"),
Crf(show_order=200, model="x.two"),
Crf(show_order=300, model="x.one"),
]
with self.assertRaises(FormsCollectionError) as cm:
CrfCollection(*crfs)
self.assertIn("Expected to be a unique sequence of crf/models.", str(cm.exception)),
self.assertIn(" Duplicates ['x.one'].", str(cm.exception))
Loading

0 comments on commit 2820508

Please sign in to comment.