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

25.27

/sis_provisioner/views/user.py

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


import json
import datetime
from logging import getLogger
from django.conf import settings
from django.utils.decorators import method_decorator
from restclients_core.exceptions import (
    InvalidNetID, InvalidRegID, DataFailureException)
from uw_sws.enrollment import enrollment_search_url_prefix
from uw_pws import PERSON_PREFIX
from uw_saml.decorators import group_required
from sis_provisioner.exceptions import (
    UserPolicyException, InvalidLoginIdException)
from sis_provisioner.dao.canvas import (
    get_user_by_sis_id, create_user, terminate_user_sessions,
    get_all_users_for_person, merge_all_users_for_person)
from sis_provisioner.dao.user import (
    get_person_by_netid, get_person_by_regid, valid_reg_id, can_access_canvas)
from sis_provisioner.models.user import User
from sis_provisioner.views.admin import RESTDispatch

logger = getLogger(__name__)


class UserView(RESTDispatch):
    """ Performs actions on a Course at /api/v1/user/<reg id>.
        GET returns 200 with User model (augmented with person)
        PUT returns 200 and updates User model
    """
    def get(self, request, *args, **kwargs):
        user_id = ''
        if 'user_id' in request.GET:
            user_id = request.GET.get('user_id', '').strip()

        if not len(user_id):
            return self.error_response(400, "Missing User ID")

        try:
            try:
                valid_reg_id(user_id.upper())
                person = get_person_by_regid(user_id.upper())
            except InvalidLoginIdException:
                person = get_person_by_netid(user_id.lower())

            return self.response_for_person(person)

        except DataFailureException as err:
            data = json.loads(err.msg)
            return self.error_response(
                400, "{} {}".format(err.status, err.msg))
        except Exception as err:
            return self.error_response(400, err)

    def post(self, request, *args, **kwargs):
        try:
            rep = json.loads(request.read())
            net_id = self.netid_from_request(rep)
            person = get_person_by_netid(net_id)
            user = User.objects.add_user_by_netid(
                person.uwnetid, priority=User.PRIORITY_IMMEDIATE)
            return self.response_for_person(person)

        except DataFailureException as err:
            data = json.loads(err.msg)
            return self.error_response(
                400, "{} {}".format(err.status, err.msg))
        except Exception as err:
            return self.error_response(400, err)

    def response_for_person(self, person):
        can_view_source_data = self.can_view_source_data(self.request)
        can_terminate_user_sessions = self.can_terminate_user_sessions(
            self.request)

        response = {
            'is_valid': True,
            'display_name': person.display_name,
            'net_id': person.uwnetid,
            'reg_id': person.uwregid,
            'added_date': None,
            'provisioned_date': None,
            'priority': 'normal',
            'queue_id': None,
            'can_merge_users': False,
            'enrollment_url': '/restclients/view/sws{}{}'.format(
                enrollment_search_url_prefix, person.uwregid) if (
                    can_view_source_data) else None,
            'canvas_users': [],
        }

        # Add the provisioning information for this user
        user = User.objects._find_existing(person.uwnetid, person.uwregid)
        if user:
            response.update(user.json_data())

        # Get the Canvas data for this user
        for user in get_all_users_for_person(person):
            user_data = user.json_data()
            try:
                user_data['can_access_canvas'] = can_access_canvas(
                    user.login_id)
            except UserPolicyException:
                user_data['can_access_canvas'] = False

            if can_view_source_data and user.sis_user_id:
                user_data['person_url'] = (
                    '/restclients/view/pws{api_path}/{uwregid}/full.json'
                ).format(api_path=PERSON_PREFIX, uwregid=user.sis_user_id)

            user_data['can_update_sis_id'] = False
            user_data['can_terminate_user_sessions'] = (
                user_data['can_access_canvas'] and
                user_data['last_login'] is not None and
                can_terminate_user_sessions)

            response['canvas_users'].append(user_data)

        if not len(response['canvas_users']):
            try:
                response['can_access_canvas'] = can_access_canvas(
                    person.uwnetid)
            except UserPolicyException:
                response['can_access_canvas'] = False
        elif len(response['canvas_users']) == 1:
            if (response['canvas_users'][0]['sis_user_id'] != person.uwregid and  # noqa
                    self.can_merge_users(self.request)):
                response['canvas_users'][0]['can_update_sis_id'] = True
        else:
            response['can_merge_users'] = self.can_merge_users(self.request)

        return self.json_response(response)


class UserMergeView(UserView):
    def put(self, request, *args, **kwargs):
        reg_id = kwargs.get('reg_id')
        try:
            person = get_person_by_regid(reg_id)
            merge_all_users_for_person(person)
        except DataFailureException as ex:
            return self.error_response(ex.status, message=ex.msg)

        return self.response_for_person(person)


class UserSessionsView(UserView):
    def delete(self, request, *args, **kwargs):
        user_id = kwargs.get('user_id')
        try:
            terminate_user_sessions(user_id)
        except DataFailureException as ex:
            return self.error_response(ex.status, message=ex.msg)

        return self.json_response({'user_id': user_id})

1	# Copyright 2023 UW-IT, University of Washington
2	# SPDX-License-Identifier: Apache-2.0
3
4
5	import json	1✔
6	import datetime	1✔
7	from logging import getLogger	1✔
8	from django.conf import settings	1✔
9	from django.utils.decorators import method_decorator	1✔
10	from restclients_core.exceptions import (	1✔
11	InvalidNetID, InvalidRegID, DataFailureException)
12	from uw_sws.enrollment import enrollment_search_url_prefix	1✔
13	from uw_pws import PERSON_PREFIX	1✔
14	from uw_saml.decorators import group_required	1✔
15	from sis_provisioner.exceptions import (	1✔
16	UserPolicyException, InvalidLoginIdException)
17	from sis_provisioner.dao.canvas import (	1✔
18	get_user_by_sis_id, create_user, terminate_user_sessions,
19	get_all_users_for_person, merge_all_users_for_person)
20	from sis_provisioner.dao.user import (	1✔
21	get_person_by_netid, get_person_by_regid, valid_reg_id, can_access_canvas)
22	from sis_provisioner.models.user import User	1✔
23	from sis_provisioner.views.admin import RESTDispatch	1✔
24
25	logger = getLogger(__name__)	1✔
26
27
28	class UserView(RESTDispatch):	1✔
29	""" Performs actions on a Course at /api/v1/user/<reg id>.
30	GET returns 200 with User model (augmented with person)
31	PUT returns 200 and updates User model
32	"""
33	def get(self, request, args, *kwargs):	1✔
34	user_id = ''	×
35	if 'user_id' in request.GET:	×
36	user_id = request.GET.get('user_id', '').strip()	×
37
38	if not len(user_id):	×
39	return self.error_response(400, "Missing User ID")	×
40
41	try:	×
42	try:	×
43	valid_reg_id(user_id.upper())	×
44	person = get_person_by_regid(user_id.upper())	×
45	except InvalidLoginIdException:	×
46	person = get_person_by_netid(user_id.lower())	×
47
48	return self.response_for_person(person)	×
49
50	except DataFailureException as err:	×
51	data = json.loads(err.msg)	×
52	return self.error_response(	×
53	400, "{} {}".format(err.status, err.msg))
54	except Exception as err:	×
55	return self.error_response(400, err)	×
56
57	def post(self, request, args, *kwargs):	1✔
58	try:	×
59	rep = json.loads(request.read())	×
60	net_id = self.netid_from_request(rep)	×
61	person = get_person_by_netid(net_id)	×
62	user = User.objects.add_user_by_netid(	×
63	person.uwnetid, priority=User.PRIORITY_IMMEDIATE)
64	return self.response_for_person(person)	×
65
66	except DataFailureException as err:	×
67	data = json.loads(err.msg)	×
68	return self.error_response(	×
69	400, "{} {}".format(err.status, err.msg))
70	except Exception as err:	×
71	return self.error_response(400, err)	×
72
73	def response_for_person(self, person):	1✔
74	can_view_source_data = self.can_view_source_data(self.request)	×
75	can_terminate_user_sessions = self.can_terminate_user_sessions(	×
76	self.request)
77
78	response = {	×
79	'is_valid': True,
80	'display_name': person.display_name,
81	'net_id': person.uwnetid,
82	'reg_id': person.uwregid,
83	'added_date': None,
84	'provisioned_date': None,
85	'priority': 'normal',
86	'queue_id': None,
87	'can_merge_users': False,
88	'enrollment_url': '/restclients/view/sws{}{}'.format(
89	enrollment_search_url_prefix, person.uwregid) if (
90	can_view_source_data) else None,
91	'canvas_users': [],
92	}
93
94	# Add the provisioning information for this user
95	user = User.objects._find_existing(person.uwnetid, person.uwregid)	×
96	if user:	×
97	response.update(user.json_data())	×
98
99	# Get the Canvas data for this user
100	for user in get_all_users_for_person(person):	×
101	user_data = user.json_data()	×
102	try:	×
103	user_data['can_access_canvas'] = can_access_canvas(	×
104	user.login_id)
105	except UserPolicyException:	×
106	user_data['can_access_canvas'] = False	×
107
108	if can_view_source_data and user.sis_user_id:	×
109	user_data['person_url'] = (	×
110	'/restclients/view/pws{api_path}/{uwregid}/full.json'
111	).format(api_path=PERSON_PREFIX, uwregid=user.sis_user_id)
112
113	user_data['can_update_sis_id'] = False	×
114	user_data['can_terminate_user_sessions'] = (	×
115	user_data['can_access_canvas'] and
116	user_data['last_login'] is not None and
117	can_terminate_user_sessions)
118
119	response['canvas_users'].append(user_data)	×
120
121	if not len(response['canvas_users']):	×
122	try:	×
123	response['can_access_canvas'] = can_access_canvas(	×
124	person.uwnetid)
125	except UserPolicyException:	×
126	response['can_access_canvas'] = False	×
127	elif len(response['canvas_users']) == 1:	×
128	if (response['canvas_users'][0]['sis_user_id'] != person.uwregid and # noqa	×
129	self.can_merge_users(self.request)):
130	response['canvas_users'][0]['can_update_sis_id'] = True	×
131	else:
132	response['can_merge_users'] = self.can_merge_users(self.request)	×
133
134	return self.json_response(response)	×
135
136
137	class UserMergeView(UserView):	1✔
138	def put(self, request, args, *kwargs):	1✔
139	reg_id = kwargs.get('reg_id')	×
140	try:	×
141	person = get_person_by_regid(reg_id)	×
142	merge_all_users_for_person(person)	×
143	except DataFailureException as ex:	×
144	return self.error_response(ex.status, message=ex.msg)	×
145
146	return self.response_for_person(person)	×
147
148
149	class UserSessionsView(UserView):	1✔
150	def delete(self, request, args, *kwargs):	1✔
151	user_id = kwargs.get('user_id')	×
152	try:	×
153	terminate_user_sessions(user_id)	×
154	except DataFailureException as ex:	×
155	return self.error_response(ex.status, message=ex.msg)	×
156
157	return self.json_response({'user_id': user_id})	×

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