6698520699

Committed 30 Oct 2023 08:45PM UTC coverage: 56.318% (+0.02%) from 56.295%

Build # 6698520699

Build Type

Pull #911

github-actions

Committed by

web-flow

Commit Message

Merge branch 'main' into develop

Pull Request Pull Request #911: Develop

Run Details

25 of 25 new or added lines in 5 files covered. (100.0%)

4573 of 8120 relevant lines covered (56.32%)

0.56 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

23.53

/sis_provisioner/builders/enrollments.py

# Copyright 2023 UW-IT, University of Washington
# SPDX-License-Identifier: Apache-2.0


from sis_provisioner.builders import Builder
from sis_provisioner.csv.format import CourseCSV, SectionCSV, EnrollmentCSV
from sis_provisioner.dao.user import get_person_by_regid
from sis_provisioner.dao.course import is_active_section, section_id_from_url
from sis_provisioner.dao.canvas import ENROLLMENT_ACTIVE, ENROLLMENT_DELETED
from sis_provisioner.exceptions import (
    UserPolicyException, MissingLoginIdException)
from uw_sws.models import Registration
from uw_sws.exceptions import InvalidCanvasIndependentStudyCourse
from restclients_core.exceptions import DataFailureException
from datetime import datetime, timedelta
from django.conf import settings
from django.utils.timezone import utc


class EnrollmentBuilder(Builder):
    """
    Generates import data for each of the passed Enrollment models.
    """
    def _process(self, enrollment):
        if enrollment.queue_id is not None:
            self.queue_id = enrollment.queue_id

        try:
            enrollment.person = get_person_by_regid(enrollment.reg_id)

            section = self.get_section_resource_by_id(enrollment.course_id)
            section.independent_study_instructor_regid = (
                enrollment.instructor_reg_id)

            if not is_active_section(section):
                return

            enrollment.section = section

        except MissingLoginIdException as err:
            if enrollment.last_modified > self.retry_missing_id:
                self._requeue_enrollment_event(enrollment, err)
            else:
                self._skip_enrollment_event(enrollment, err)
            return
        except (UserPolicyException,
                InvalidCanvasIndependentStudyCourse) as err:
            self._skip_enrollment_event(enrollment, err)
            return
        except DataFailureException as err:
            if err.status == 404:
                self._skip_enrollment_event(enrollment, err)
            else:
                self._requeue_enrollment_event(enrollment, err)
            return

        if enrollment.is_instructor():
            self._process_instructor_enrollment(enrollment)
        else:  # student/auditor
            if len(section.linked_section_urls):
                # Don't enroll students into primary sections
                self._skip_enrollment_event(
                    enrollment, 'Section has linked sections')
            else:
                self._process_student_enrollment(enrollment)

    def _process_instructor_enrollment(self, enrollment):
        if enrollment.section.is_independent_study:
            if enrollment.is_active():
                self.data.add(SectionCSV(section=enrollment.section))
                self.add_teacher_enrollment_data(enrollment.section,
                                                 enrollment.person,
                                                 enrollment.status)
            else:
                enrollment.section.delete_flag = section.DELETE_FLAG_WITHDRAWN

            # Add or remove independent study course
            self.data.add(CourseCSV(section=enrollment.section))

        elif len(enrollment.section.linked_section_urls):
            # Add/remove primary instructor for each linked section
            for url in enrollment.section.linked_section_urls:
                try:
                    linked_course_id = section_id_from_url(url)
                    linked_section = self.get_section_resource_by_id(
                        linked_course_id)
                    self.data.add(SectionCSV(section=linked_section))
                    self.add_teacher_enrollment_data(linked_section,
                                                     enrollment.person,
                                                     enrollment.status)
                except Exception:
                    continue

        else:
            self.data.add(SectionCSV(section=enrollment.section))
            self.add_teacher_enrollment_data(
                enrollment.section, enrollment.person, enrollment.status)

    def _process_student_enrollment(self, enrollment):
        request_date = enrollment.request_date if (
            enrollment.request_date is not None) else enrollment.last_modified

        registration = Registration(section=enrollment.section,
                                    person=enrollment.person,
                                    is_active=enrollment.is_active(),
                                    request_date=request_date.date(),
                                    request_status=enrollment.status)
        self.add_student_enrollment_data(registration)
        self.data.add(SectionCSV(section=enrollment.section))

    def _requeue_enrollment_event(self, enrollment, err):
        enrollment.queue_id = None
        enrollment.priority = enrollment.PRIORITY_DEFAULT
        enrollment.save()
        self.logger.info("Requeue enrollment {} in {}: {}".format(
            enrollment.reg_id, enrollment.course_id, err))

    def _skip_enrollment_event(self, enrollment, err):
        enrollment.queue_id = None
        enrollment.priority = enrollment.PRIORITY_NONE
        enrollment.save()
        self.logger.info("Skip enrollment {} in {}: {}".format(
            enrollment.reg_id, enrollment.course_id, err))

    def _init_build(self, **kwargs):
        now = datetime.utcnow().replace(tzinfo=utc)
        timeout = getattr(settings, 'MISSING_LOGIN_ID_RETRY_TIMEOUT', 48)
        self.retry_missing_id = now - timedelta(hours=timeout)


class InvalidEnrollmentBuilder(Builder):
    """
    Generates import data for each of the passed InvalidEnrollment models.
    """
    def _process(self, inv_enrollment):
        now = datetime.utcnow().replace(tzinfo=utc)
        grace_dt = now - timedelta(days=getattr(
            settings, 'INVALID_ENROLLMENT_GRACE_DAYS', 90))
        status = None

        try:
            # Verify that the check conditions still exist
            if (inv_enrollment.user.is_affiliate_user() or
                    inv_enrollment.user.is_sponsored_user()):
                status = ENROLLMENT_ACTIVE
                if inv_enrollment.deleted_date is not None:
                    inv_enrollment.restored_date = now

            elif inv_enrollment.user.is_student_user():
                if inv_enrollment.found_date < grace_dt:
                    status = ENROLLMENT_DELETED
                    inv_enrollment.deleted_date = now
                    inv_enrollment.restored_date = None

            if status is not None:
                person = get_person_by_regid(inv_enrollment.user.reg_id)
                if self.add_user_data_for_person(person):
                    self.data.add(EnrollmentCSV(
                        section_id=inv_enrollment.section_id,
                        person=person,
                        role=inv_enrollment.role,
                        status=status))
                inv_enrollment.save()

        except DataFailureException as err:
            inv_enrollment.queue_id = None
            inv_enrollment.save()
            self.logger.info('Requeue invalid enrollment {} in {}: {}'.format(
                inv_enrollment.user.reg_id, inv_enrollment.section_id, err))

1	# Copyright 2023 UW-IT, University of Washington
2	# SPDX-License-Identifier: Apache-2.0
3
4
5	from sis_provisioner.builders import Builder	1✔
6	from sis_provisioner.csv.format import CourseCSV, SectionCSV, EnrollmentCSV	1✔
7	from sis_provisioner.dao.user import get_person_by_regid	1✔
8	from sis_provisioner.dao.course import is_active_section, section_id_from_url	1✔
9	from sis_provisioner.dao.canvas import ENROLLMENT_ACTIVE, ENROLLMENT_DELETED	1✔
10	from sis_provisioner.exceptions import (	1✔
11	UserPolicyException, MissingLoginIdException)
12	from uw_sws.models import Registration	1✔
13	from uw_sws.exceptions import InvalidCanvasIndependentStudyCourse	1✔
14	from restclients_core.exceptions import DataFailureException	1✔
15	from datetime import datetime, timedelta	1✔
16	from django.conf import settings	1✔
17	from django.utils.timezone import utc	1✔
18
19
20	class EnrollmentBuilder(Builder):	1✔
21	"""
22	Generates import data for each of the passed Enrollment models.
23	"""
24	def _process(self, enrollment):	1✔
25	if enrollment.queue_id is not None:	×
26	self.queue_id = enrollment.queue_id	×
27
28	try:	×
29	enrollment.person = get_person_by_regid(enrollment.reg_id)	×
30
31	section = self.get_section_resource_by_id(enrollment.course_id)	×
32	section.independent_study_instructor_regid = (	×
33	enrollment.instructor_reg_id)
34
35	if not is_active_section(section):	×
36	return	×
37
38	enrollment.section = section	×
39
40	except MissingLoginIdException as err:	×
41	if enrollment.last_modified > self.retry_missing_id:	×
42	self._requeue_enrollment_event(enrollment, err)	×
43	else:
44	self._skip_enrollment_event(enrollment, err)	×
45	return	×
46	except (UserPolicyException,	×
47	InvalidCanvasIndependentStudyCourse) as err:
48	self._skip_enrollment_event(enrollment, err)	×
49	return	×
50	except DataFailureException as err:	×
51	if err.status == 404:	×
52	self._skip_enrollment_event(enrollment, err)	×
53	else:
54	self._requeue_enrollment_event(enrollment, err)	×
55	return	×
56
57	if enrollment.is_instructor():	×
58	self._process_instructor_enrollment(enrollment)	×
59	else: # student/auditor
60	if len(section.linked_section_urls):	×
61	# Don't enroll students into primary sections
62	self._skip_enrollment_event(	×
63	enrollment, 'Section has linked sections')
64	else:
65	self._process_student_enrollment(enrollment)	×
66
67	def _process_instructor_enrollment(self, enrollment):	1✔
68	if enrollment.section.is_independent_study:	×
69	if enrollment.is_active():	×
70	self.data.add(SectionCSV(section=enrollment.section))	×
71	self.add_teacher_enrollment_data(enrollment.section,	×
72	enrollment.person,
73	enrollment.status)
74	else:
75	enrollment.section.delete_flag = section.DELETE_FLAG_WITHDRAWN	×
76
77	# Add or remove independent study course
78	self.data.add(CourseCSV(section=enrollment.section))	×
79
80	elif len(enrollment.section.linked_section_urls):	×
81	# Add/remove primary instructor for each linked section
82	for url in enrollment.section.linked_section_urls:	×
83	try:	×
84	linked_course_id = section_id_from_url(url)	×
85	linked_section = self.get_section_resource_by_id(	×
86	linked_course_id)
87	self.data.add(SectionCSV(section=linked_section))	×
88	self.add_teacher_enrollment_data(linked_section,	×
89	enrollment.person,
90	enrollment.status)
91	except Exception:	×
92	continue	×
93
94	else:
95	self.data.add(SectionCSV(section=enrollment.section))	×
96	self.add_teacher_enrollment_data(	×
97	enrollment.section, enrollment.person, enrollment.status)
98
99	def _process_student_enrollment(self, enrollment):	1✔
100	request_date = enrollment.request_date if (	×
101	enrollment.request_date is not None) else enrollment.last_modified
102
103	registration = Registration(section=enrollment.section,	×
104	person=enrollment.person,
105	is_active=enrollment.is_active(),
106	request_date=request_date.date(),
107	request_status=enrollment.status)
108	self.add_student_enrollment_data(registration)	×
109	self.data.add(SectionCSV(section=enrollment.section))	×
110
111	def _requeue_enrollment_event(self, enrollment, err):	1✔
112	enrollment.queue_id = None	×
113	enrollment.priority = enrollment.PRIORITY_DEFAULT	×
114	enrollment.save()	×
115	self.logger.info("Requeue enrollment {} in {}: {}".format(	×
116	enrollment.reg_id, enrollment.course_id, err))
117
118	def _skip_enrollment_event(self, enrollment, err):	1✔
119	enrollment.queue_id = None	×
120	enrollment.priority = enrollment.PRIORITY_NONE	×
121	enrollment.save()	×
122	self.logger.info("Skip enrollment {} in {}: {}".format(	×
123	enrollment.reg_id, enrollment.course_id, err))
124
125	def _init_build(self, **kwargs):	1✔
126	now = datetime.utcnow().replace(tzinfo=utc)	1✔
127	timeout = getattr(settings, 'MISSING_LOGIN_ID_RETRY_TIMEOUT', 48)	1✔
128	self.retry_missing_id = now - timedelta(hours=timeout)	1✔
129
130
131	class InvalidEnrollmentBuilder(Builder):	1✔
132	"""
133	Generates import data for each of the passed InvalidEnrollment models.
134	"""
135	def _process(self, inv_enrollment):	1✔
136	now = datetime.utcnow().replace(tzinfo=utc)	×
137	grace_dt = now - timedelta(days=getattr(	×
138	settings, 'INVALID_ENROLLMENT_GRACE_DAYS', 90))
139	status = None	×
140
141	try:	×
142	# Verify that the check conditions still exist
143	if (inv_enrollment.user.is_affiliate_user() or	×
144	inv_enrollment.user.is_sponsored_user()):
145	status = ENROLLMENT_ACTIVE	×
146	if inv_enrollment.deleted_date is not None:	×
147	inv_enrollment.restored_date = now	×
148
149	elif inv_enrollment.user.is_student_user():	×
150	if inv_enrollment.found_date < grace_dt:	×
151	status = ENROLLMENT_DELETED	×
152	inv_enrollment.deleted_date = now	×
153	inv_enrollment.restored_date = None	×
154
155	if status is not None:	×
156	person = get_person_by_regid(inv_enrollment.user.reg_id)	×
157	if self.add_user_data_for_person(person):	×
158	self.data.add(EnrollmentCSV(	×
159	section_id=inv_enrollment.section_id,
160	person=person,
161	role=inv_enrollment.role,
162	status=status))
163	inv_enrollment.save()	×
164
165	except DataFailureException as err:	×
166	inv_enrollment.queue_id = None	×
167	inv_enrollment.save()	×
168	self.logger.info('Requeue invalid enrollment {} in {}: {}'.format(	×
169	inv_enrollment.user.reg_id, inv_enrollment.section_id, err))

uw-it-aca / canvas-sis-provisioner / 6698520699

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous