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

81.52

/sis_provisioner/dao/group.py

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


from django.conf import settings
from django.utils.timezone import utc
from uw_gws import GWS
from uw_gws.exceptions import InvalidGroupID
from restclients_core.exceptions import DataFailureException
from sis_provisioner.dao.user import valid_net_id
from sis_provisioner.exceptions import (
    UserPolicyException, GroupPolicyException, GroupNotFoundException,
    GroupUnauthorizedException)
from logging import getLogger
import re

logger = getLogger(__name__)


def valid_group_id(group_id):
    try:
        GWS()._valid_group_id(group_id)
    except InvalidGroupID:
        raise GroupPolicyException("Invalid Group ID: {}".format(group_id))

    RE_GROUP_DISALLOWED = re.compile(r'^({}).*$'.format('|'.join(
        getattr(settings, 'DISALLOWED_UW_GROUPS', []))))
    if RE_GROUP_DISALLOWED.match(group_id):
        raise GroupPolicyException(
            "This group cannot be used in Canvas: {}".format(group_id))


def is_modified_group(group_id, changed_since_dt):
    try:
        group = GWS().get_group_by_id(group_id)
        member_mtime = group.membership_modified.replace(tzinfo=utc)
        return (member_mtime > changed_since_dt)
    except DataFailureException as err:
        if err.status == 404:
            raise GroupNotFoundException(
                "Group not found: {}".format(group_id))
        else:
            raise


def get_group(act_as, group_id):
    return GWS(act_as=act_as).get_group_by_id(group_id)


def search_groups(act_as, **kwargs):
    if not kwargs.get('scope'):
        kwargs['scope'] = 'all'

    if kwargs.get('name') and not kwargs['name'].endswith('*'):
        kwargs['name'] += '*'

    groups = []
    for group in GWS(act_as=act_as).search_groups(**kwargs):
        try:
            valid_group_id(group.name)
            groups.append(group)
        except GroupPolicyException:
            pass

    return groups


def get_sis_import_members():
    valid_members = {}
    group_id = getattr(settings, 'SIS_IMPORT_USERS')
    for member in GWS().get_effective_members(group_id):
        try:
            if member.is_uwnetid():
                valid_net_id(member.name)
                valid_members[member.name] = member
        except UserPolicyException:
            pass

    return list(valid_members.values())


def get_effective_members(group_id, act_as=None):
    gws = GWS(act_as=act_as)
    seen_group_ids = set()  # Prevent duplicate group processing

    def _get_members(group_id):
        valid_members = {}
        invalid_members = {}
        member_group_ids = []

        try:
            valid_group_id(group_id)

            if group_id in seen_group_ids:
                logger.info("Duplicate group: {}, Processed groups: {}".format(
                    group_id, list(seen_group_ids)))
                return (valid_members, invalid_members, member_group_ids)
            seen_group_ids.add(group_id)

            for member in gws.get_members(group_id):
                try:
                    if member.is_uwnetid():
                        valid_net_id(member.name)
                        valid_members[member.name] = member

                    elif member.is_eppn():
                        invalid_members[member.name] = member

                    elif member.is_group():
                        (valid_sub, invalid_sub,
                            member_subgroups) = _get_members(member.name)
                        valid_members.update(valid_sub)
                        invalid_members.update(invalid_sub)
                        member_group_ids += [member.name] + member_subgroups

                except (UserPolicyException, GroupPolicyException) as err:
                    member.error = err
                    invalid_members[member.name] = member

        except DataFailureException as err:
            # Group not found or access denied is ok
            if err.status == 404:
                raise GroupNotFoundException(
                    "Group not found: {}".format(group_id))
            elif err.status == 401:
                raise GroupUnauthorizedException(
                    "Group not permitted for {}: {}".format(
                        gws.act_as, group_id))
            else:
                raise

        except GroupPolicyException as err:
            raise

        return (valid_members, invalid_members, member_group_ids)

    (valid_members, invalid_members, member_group_ids) = _get_members(group_id)
    return (list(valid_members.values()),
            list(invalid_members.values()),
            member_group_ids)

1	# Copyright 2023 UW-IT, University of Washington
2	# SPDX-License-Identifier: Apache-2.0
3
4
5	from django.conf import settings	1✔
6	from django.utils.timezone import utc	1✔
7	from uw_gws import GWS	1✔
8	from uw_gws.exceptions import InvalidGroupID	1✔
9	from restclients_core.exceptions import DataFailureException	1✔
10	from sis_provisioner.dao.user import valid_net_id	1✔
11	from sis_provisioner.exceptions import (	1✔
12	UserPolicyException, GroupPolicyException, GroupNotFoundException,
13	GroupUnauthorizedException)
14	from logging import getLogger	1✔
15	import re	1✔
16
17	logger = getLogger(__name__)	1✔
18
19
20	def valid_group_id(group_id):	1✔
21	try:	1✔
22	GWS()._valid_group_id(group_id)	1✔
23	except InvalidGroupID:	1✔
24	raise GroupPolicyException("Invalid Group ID: {}".format(group_id))	1✔
25
26	RE_GROUP_DISALLOWED = re.compile(r'^({}).*$'.format('\|'.join(	1✔
27	getattr(settings, 'DISALLOWED_UW_GROUPS', []))))
28	if RE_GROUP_DISALLOWED.match(group_id):	1✔
29	raise GroupPolicyException(	1✔
30	"This group cannot be used in Canvas: {}".format(group_id))
31
32
33	def is_modified_group(group_id, changed_since_dt):	1✔
34	try:	1✔
35	group = GWS().get_group_by_id(group_id)	1✔
36	member_mtime = group.membership_modified.replace(tzinfo=utc)	1✔
37	return (member_mtime > changed_since_dt)	1✔
38	except DataFailureException as err:	1✔
39	if err.status == 404:	1✔
40	raise GroupNotFoundException(	1✔
41	"Group not found: {}".format(group_id))
42	else:
43	raise	×
44
45
46	def get_group(act_as, group_id):	1✔
47	return GWS(act_as=act_as).get_group_by_id(group_id)	1✔
48
49
50	def search_groups(act_as, **kwargs):	1✔
51	if not kwargs.get('scope'):	1✔
52	kwargs['scope'] = 'all'	1✔
53
54	if kwargs.get('name') and not kwargs['name'].endswith('*'):	1✔
55	kwargs['name'] += '*'	1✔
56
57	groups = []	1✔
58	for group in GWS(act_as=act_as).search_groups(**kwargs):	1✔
59	try:	×
60	valid_group_id(group.name)	×
61	groups.append(group)	×
62	except GroupPolicyException:	×
63	pass	×
64
65	return groups	1✔
66
67
68	def get_sis_import_members():	1✔
69	valid_members = {}	1✔
70	group_id = getattr(settings, 'SIS_IMPORT_USERS')	1✔
71	for member in GWS().get_effective_members(group_id):	1✔
72	try:	1✔
73	if member.is_uwnetid():	1✔
74	valid_net_id(member.name)	1✔
75	valid_members[member.name] = member	1✔
76	except UserPolicyException:	×
77	pass	×
78
79	return list(valid_members.values())	1✔
80
81
82	def get_effective_members(group_id, act_as=None):	1✔
83	gws = GWS(act_as=act_as)	1✔
84	seen_group_ids = set() # Prevent duplicate group processing	1✔
85
86	def _get_members(group_id):	1✔
87	valid_members = {}	1✔
88	invalid_members = {}	1✔
89	member_group_ids = []	1✔
90
91	try:	1✔
92	valid_group_id(group_id)	1✔
93
94	if group_id in seen_group_ids:	1✔
95	logger.info("Duplicate group: {}, Processed groups: {}".format(	1✔
96	group_id, list(seen_group_ids)))
97	return (valid_members, invalid_members, member_group_ids)	1✔
98	seen_group_ids.add(group_id)	1✔
99
100	for member in gws.get_members(group_id):	1✔
101	try:	1✔
102	if member.is_uwnetid():	1✔
103	valid_net_id(member.name)	1✔
104	valid_members[member.name] = member	1✔
105
106	elif member.is_eppn():	1✔
107	invalid_members[member.name] = member	×
108
109	elif member.is_group():	1✔
110	(valid_sub, invalid_sub,	1✔
111	member_subgroups) = _get_members(member.name)
112	valid_members.update(valid_sub)	1✔
113	invalid_members.update(invalid_sub)	1✔
114	member_group_ids += [member.name] + member_subgroups	1✔
115
116	except (UserPolicyException, GroupPolicyException) as err:	×
117	member.error = err	×
118	invalid_members[member.name] = member	×
119
120	except DataFailureException as err:	1✔
121	# Group not found or access denied is ok
122	if err.status == 404:	1✔
123	raise GroupNotFoundException(	1✔
124	"Group not found: {}".format(group_id))
125	elif err.status == 401:	×
126	raise GroupUnauthorizedException(	×
127	"Group not permitted for {}: {}".format(
128	gws.act_as, group_id))
129	else:
130	raise	×
131
132	except GroupPolicyException as err:	×
133	raise	×
134
135	return (valid_members, invalid_members, member_group_ids)	1✔
136
137	(valid_members, invalid_members, member_group_ids) = _get_members(group_id)	1✔
138	return (list(valid_members.values()),	1✔
139	list(invalid_members.values()),
140	member_group_ids)

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