4571350728

Build Type

github

Committed by GitHub

Commit Message

Merge e0b4c1d0e into 8cacbeb40

Pull Request Pull Request #831: Develop

Run Details

38 of 38 new or added lines in 10 files covered. (100.0%)

4514 of 8004 relevant lines covered (56.4%)

0.56 hits per line

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

0.0

/sis_provisioner/events/enrollment.py

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


from sis_provisioner.events import SISProvisionerProcessor
from sis_provisioner.models.events import EnrollmentLog
from sis_provisioner.dao.canvas import (
    get_student_sis_import_role, ENROLLMENT_ACTIVE, ENROLLMENT_DELETED)
from sis_provisioner.dao.user import valid_reg_id
from sis_provisioner.exceptions import (
    InvalidLoginIdException, UnhandledActionCodeException)
from uw_sws.models import Term, Section
from dateutil.parser import parse as date_parse

log_prefix = 'ENROLLMENT:'
QUEUE_SETTINGS_NAME = 'ENROLLMENT_V2'


class EnrollmentProcessor(SISProvisionerProcessor):
    """
    Collects enrollment event described by
    https://wiki.cac.washington.edu/display/StudentEvents/UW+Course+Enrollment+v2
    """
    _logModel = EnrollmentLog

    # What we expect in a v2 enrollment message
    _eventMessageType = 'uw-student-registration-v2'
    _eventMessageVersion = '2'

    def __init__(self):
        super(EnrollmentProcessor, self).__init__(
            queue_settings_name=QUEUE_SETTINGS_NAME, is_encrypted=True)

    def process_message_body(self, json_data):
        enrollments = []
        for event in json_data.get('Events', []):
            section_data = event['Section']
            course_data = section_data['Course']

            section = Section()
            section.term = Term(quarter=course_data['Quarter'],
                                year=course_data['Year'])
            section.curriculum_abbr = course_data['CurriculumAbbreviation']
            section.course_number = course_data['CourseNumber']
            section.section_id = section_data['SectionID']
            section.is_primary_section = True
            section.linked_section_urls = []

            if ('PrimarySection' in event and
                    'Course' in event['PrimarySection']):
                primary_course = event['PrimarySection']['Course']
                if primary_course:
                    section.is_primary_section = False
                    section.primary_section_curriculum_abbr = (
                        primary_course['CurriculumAbbreviation'])
                    section.primary_section_course_number = (
                        primary_course['CourseNumber'])
                    section.primary_section_id = (
                        event['PrimarySection']['SectionID'])

            try:
                valid_reg_id(event['Person']['UWRegID'])
                data = {
                    'Section': section,
                    'Role': get_student_sis_import_role(),
                    'UWRegID': event['Person']['UWRegID'],
                    'Status': self._enrollment_status(event, section),
                    'LastModified': date_parse(event['LastModified']),
                    'InstructorUWRegID': event['Instructor']['UWRegID'] if (
                        'Instructor' in event and event['Instructor'] and
                        'UWRegID' in event['Instructor']) else None
                }

                if 'Auditor' in event and event['Auditor']:
                    data['Role'] = 'Auditor'

                if 'RequestDate' in event:
                    data['RequestDate'] = date_parse(event['RequestDate'])

                self.logger.debug('{} LOAD {} for {} at {}'.format(
                    log_prefix,
                    event['Action']['Code'],
                    event['Person']['UWRegID'],
                    event['LastModified']))
                enrollments.append(data)
            except UnhandledActionCodeException:
                self.logger.warning('{} UNKNOWN {} for {} at {}'.format(
                    log_prefix,
                    event['Action']['Code'],
                    event['Person']['UWRegID'],
                    event['LastModified']))
                pass
            except InvalidLoginIdException:
                self.logger.warning('{} INVALID UWRegID {}, Href: {}'.format(
                    log_prefix, event['Person']['UWRegID'],
                    event['Person']['Href']))

        self.load_enrollments(enrollments)

    def _enrollment_status(self, event, section):
        # Canvas "active" corresponds to Action codes:
        #   "A" == ADDED and
        #   "S" == STANDBY (EO only status)
        action_code = event['Action']['Code'].upper()

        if action_code == 'A':
            return ENROLLMENT_ACTIVE

        if action_code == 'S':
            self.logger.debug('{} ADD standby {} to {}'.format(
                log_prefix,
                event['Person']['UWRegID'],
                section.canvas_section_sis_id()))
            return ENROLLMENT_ACTIVE

        if action_code == 'D':
            return ENROLLMENT_DELETED

        raise UnhandledActionCodeException()

1	# Copyright 2023 UW-IT, University of Washington
2	# SPDX-License-Identifier: Apache-2.0
3
4
5	from sis_provisioner.events import SISProvisionerProcessor	×
6	from sis_provisioner.models.events import EnrollmentLog	×
7	from sis_provisioner.dao.canvas import (	×
8	get_student_sis_import_role, ENROLLMENT_ACTIVE, ENROLLMENT_DELETED)
9	from sis_provisioner.dao.user import valid_reg_id	×
10	from sis_provisioner.exceptions import (	×
11	InvalidLoginIdException, UnhandledActionCodeException)
12	from uw_sws.models import Term, Section	×
13	from dateutil.parser import parse as date_parse	×
14
15	log_prefix = 'ENROLLMENT:'	×
16	QUEUE_SETTINGS_NAME = 'ENROLLMENT_V2'	×
17
18
19	class EnrollmentProcessor(SISProvisionerProcessor):	×
20	"""
21	Collects enrollment event described by
22	https://wiki.cac.washington.edu/display/StudentEvents/UW+Course+Enrollment+v2
23	"""
24	_logModel = EnrollmentLog	×
25
26	# What we expect in a v2 enrollment message
27	_eventMessageType = 'uw-student-registration-v2'	×
28	_eventMessageVersion = '2'	×
29
30	def __init__(self):	×
31	super(EnrollmentProcessor, self).__init__(	×
32	queue_settings_name=QUEUE_SETTINGS_NAME, is_encrypted=True)
33
34	def process_message_body(self, json_data):	×
35	enrollments = []	×
36	for event in json_data.get('Events', []):	×
37	section_data = event['Section']	×
38	course_data = section_data['Course']	×
39
40	section = Section()	×
41	section.term = Term(quarter=course_data['Quarter'],	×
42	year=course_data['Year'])
43	section.curriculum_abbr = course_data['CurriculumAbbreviation']	×
44	section.course_number = course_data['CourseNumber']	×
45	section.section_id = section_data['SectionID']	×
46	section.is_primary_section = True	×
47	section.linked_section_urls = []	×
48
49	if ('PrimarySection' in event and	×
50	'Course' in event['PrimarySection']):
51	primary_course = event['PrimarySection']['Course']	×
52	if primary_course:	×
53	section.is_primary_section = False	×
54	section.primary_section_curriculum_abbr = (	×
55	primary_course['CurriculumAbbreviation'])
56	section.primary_section_course_number = (	×
57	primary_course['CourseNumber'])
58	section.primary_section_id = (	×
59	event['PrimarySection']['SectionID'])
60
61	try:	×
62	valid_reg_id(event['Person']['UWRegID'])	×
63	data = {	×
64	'Section': section,
65	'Role': get_student_sis_import_role(),
66	'UWRegID': event['Person']['UWRegID'],
67	'Status': self._enrollment_status(event, section),
68	'LastModified': date_parse(event['LastModified']),
69	'InstructorUWRegID': event['Instructor']['UWRegID'] if (
70	'Instructor' in event and event['Instructor'] and
71	'UWRegID' in event['Instructor']) else None
72	}
73
74	if 'Auditor' in event and event['Auditor']:	×
75	data['Role'] = 'Auditor'	×
76
77	if 'RequestDate' in event:	×
78	data['RequestDate'] = date_parse(event['RequestDate'])	×
79
80	self.logger.debug('{} LOAD {} for {} at {}'.format(	×
81	log_prefix,
82	event['Action']['Code'],
83	event['Person']['UWRegID'],
84	event['LastModified']))
85	enrollments.append(data)	×
86	except UnhandledActionCodeException:	×
87	self.logger.warning('{} UNKNOWN {} for {} at {}'.format(	×
88	log_prefix,
89	event['Action']['Code'],
90	event['Person']['UWRegID'],
91	event['LastModified']))
92	pass	×
93	except InvalidLoginIdException:	×
94	self.logger.warning('{} INVALID UWRegID {}, Href: {}'.format(	×
95	log_prefix, event['Person']['UWRegID'],
96	event['Person']['Href']))
97
98	self.load_enrollments(enrollments)	×
99
100	def _enrollment_status(self, event, section):	×
101	# Canvas "active" corresponds to Action codes:
102	# "A" == ADDED and
103	# "S" == STANDBY (EO only status)
104	action_code = event['Action']['Code'].upper()	×
105
106	if action_code == 'A':	×
107	return ENROLLMENT_ACTIVE	×
108
109	if action_code == 'S':	×
110	self.logger.debug('{} ADD standby {} to {}'.format(	×
111	log_prefix,
112	event['Person']['UWRegID'],
113	section.canvas_section_sis_id()))
114	return ENROLLMENT_ACTIVE	×
115
116	if action_code == 'D':	×
117	return ENROLLMENT_DELETED	×
118
119	raise UnhandledActionCodeException()	×

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

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