16203281716

Committed 10 Jul 2025 06:40PM UTC coverage: 28.102% (-46.6%) from 74.666%

Build # 16203281716

Build Type

push

github

Committed by

paulineribeyre

Commit Message

only run test_login_user_with_idp_already_in_db

Run Details

3116 of 11088 relevant lines covered (28.1%)

0.28 hits per line

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

18.48

fence/resources/admin/admin_groups.py

from fence.resources.admin import admin_users as au
from fence.errors import NotFound, UserError
from fence.resources import group as gp, project as pj, user as us


__all__ = [
    "create_group",
    "delete_group",
    "update_group",
    "get_group_info",
    "get_all_groups",
    "get_group_users",
    "connect_project_to_group",
    "update_group_users_projects",
    "add_projects_to_group",
    "disconnect_project_from_group",
    "update_user_projects_within_group",
    "remove_projects_from_group",
    "get_group_projects",
]


def create_group(current_session, groupname, description):
    """
    Creates a group and returns it
    """
    return gp.create_group(current_session, groupname, description)


def delete_group(current_session, groupname):
    """
    Deletes a group
    """
    projects_to_purge = gp.get_group_projects(current_session, groupname)
    remove_projects_from_group(current_session, groupname, projects_to_purge)
    gp.clear_users_in_group(current_session, groupname)
    gp.clear_projects_in_group(current_session, groupname)
    gp.delete_group(current_session, groupname)
    return {"result": "success"}


def update_group(current_session, groupname, description, new_name=None):
    gp.update_group(current_session, groupname, description, new_name)
    name = new_name or groupname
    return gp.get_group_info(current_session, name)


def get_group_info(current_session, groupname):
    return gp.get_group_info(current_session, groupname)


def get_all_groups(current_session):
    groups = gp.get_all_groups(current_session)
    groups_list = [get_group_info(current_session, group.name) for group in groups]
    return {"groups": groups_list}


def get_group_users(current_session, groupname):
    users = gp.get_group_users(current_session, groupname)
    get_user_info = lambda user: us.get_user_info(current_session, user.username)
    users_names = [
        {"name": the_user["name"], "role": the_user["role"]}
        for the_user in map(get_user_info, users)
    ]
    return {"users": users_names}


def connect_project_to_group(current_session, grp, project=None):
    prj = pj.get_project(current_session, project)
    if not prj:
        raise UserError(("Project {0} doesn't exist".format(project)))
    return gp.connect_project_to_group(current_session, grp, prj)


def update_group_users_projects(current_session, group, project, users):
    proj = pj.get_project(current_session, project)
    for user in users:
        try:
            user_projects = list(user.project_access.keys())
            if project not in user_projects:
                project_info = {"auth_id": proj.auth_id, "privilege": ["read"]}
                au.connect_user_to_project(
                    current_session,
                    us.get_user(current_session, user.username),
                    project_info,
                )
        except NotFound:
            pass
    return {
        "success": "users {0} connected to project {1}".format(
            [user.username for user in users], project
        )
    }


def add_projects_to_group(current_session, groupname, projects=None):
    if not projects:
        projects = []
    grp = gp.get_group(current_session, groupname)
    usrs = gp.get_group_users(current_session, groupname)
    if not grp:
        raise UserError("Error: group does not exist")
    responses = []
    for proj in projects:
        try:
            response = connect_project_to_group(current_session, grp, proj)
            responses.append(response)
            update_group_users_projects(current_session, grp, proj, usrs)
        except Exception as e:
            current_session.rollback()
            raise e
    return {"result": responses}


def disconnect_project_from_group(current_session, grp, projectname):
    prj = pj.get_project(current_session, projectname)
    if not prj:
        return {"warning": ("Project {0} doesn't exist".format(projectname))}
    return gp.remove_project_from_group(current_session, grp, prj)


def update_user_projects_within_group(
    current_session, username, groupname, projectname
):
    user_groups = us.get_user_groups(current_session, username)
    """
    Simplified version for awg:
      Users only have read permission, so just checking the
      presence of the project in any of their other groups
      suffices to keep the projec in the list.
    In real life we should check permissions coming from all groups
    and remove the specific ones comiing from groupname
    """
    group_projects = [
        gp.get_group_projects(current_session, group)
        for group in user_groups["groups"]
        if group != groupname
    ]

    projects_to_keep = [item for sublist in group_projects for item in sublist]

    if projectname not in projects_to_keep:
        try:
            us.remove_user_from_project(
                current_session,
                us.get_user(current_session, username),
                pj.get_project(current_session, projectname),
            )
        except NotFound:
            # somehow the user was not linked to that project
            pass


def remove_projects_from_group(current_session, groupname, projects=None):
    if not projects:
        projects = []
    grp = gp.get_group(current_session, groupname)
    usrs = get_group_users(current_session, groupname)
    users_names = [x["name"] for x in usrs["users"]]
    if not grp:
        raise UserError("Error: group does not exist")
    responses = []
    for proj in projects:
        for usr in users_names:
            update_user_projects_within_group(current_session, usr, groupname, proj)
        response = disconnect_project_from_group(current_session, grp, proj)
        responses.append(response)
    return {"result": responses}


def get_group_projects(current_session, groupname):
    return gp.get_group_projects(current_session, groupname)

1	from fence.resources.admin import admin_users as au	1✔
2	from fence.errors import NotFound, UserError	1✔
3	from fence.resources import group as gp, project as pj, user as us	1✔
4
5
6	__all__ = [	1✔
7	"create_group",
8	"delete_group",
9	"update_group",
10	"get_group_info",
11	"get_all_groups",
12	"get_group_users",
13	"connect_project_to_group",
14	"update_group_users_projects",
15	"add_projects_to_group",
16	"disconnect_project_from_group",
17	"update_user_projects_within_group",
18	"remove_projects_from_group",
19	"get_group_projects",
20	]
21
22
23	def create_group(current_session, groupname, description):	1✔
24	"""
25	Creates a group and returns it
26	"""
27	return gp.create_group(current_session, groupname, description)	×
28
29
30	def delete_group(current_session, groupname):	1✔
31	"""
32	Deletes a group
33	"""
34	projects_to_purge = gp.get_group_projects(current_session, groupname)	×
35	remove_projects_from_group(current_session, groupname, projects_to_purge)	×
36	gp.clear_users_in_group(current_session, groupname)	×
37	gp.clear_projects_in_group(current_session, groupname)	×
38	gp.delete_group(current_session, groupname)	×
39	return {"result": "success"}	×
40
41
42	def update_group(current_session, groupname, description, new_name=None):	1✔
43	gp.update_group(current_session, groupname, description, new_name)	×
44	name = new_name or groupname	×
45	return gp.get_group_info(current_session, name)	×
46
47
48	def get_group_info(current_session, groupname):	1✔
49	return gp.get_group_info(current_session, groupname)	×
50
51
52	def get_all_groups(current_session):	1✔
53	groups = gp.get_all_groups(current_session)	×
54	groups_list = [get_group_info(current_session, group.name) for group in groups]	×
55	return {"groups": groups_list}	×
56
57
58	def get_group_users(current_session, groupname):	1✔
59	users = gp.get_group_users(current_session, groupname)	×
60	get_user_info = lambda user: us.get_user_info(current_session, user.username)	×
61	users_names = [	×
62	{"name": the_user["name"], "role": the_user["role"]}
63	for the_user in map(get_user_info, users)
64	]
65	return {"users": users_names}	×
66
67
68	def connect_project_to_group(current_session, grp, project=None):	1✔
69	prj = pj.get_project(current_session, project)	×
70	if not prj:	×
71	raise UserError(("Project {0} doesn't exist".format(project)))	×
72	return gp.connect_project_to_group(current_session, grp, prj)	×
73
74
75	def update_group_users_projects(current_session, group, project, users):	1✔
76	proj = pj.get_project(current_session, project)	×
77	for user in users:	×
78	try:	×
79	user_projects = list(user.project_access.keys())	×
80	if project not in user_projects:	×
81	project_info = {"auth_id": proj.auth_id, "privilege": ["read"]}	×
82	au.connect_user_to_project(	×
83	current_session,
84	us.get_user(current_session, user.username),
85	project_info,
86	)
87	except NotFound:	×
88	pass	×
89	return {	×
90	"success": "users {0} connected to project {1}".format(
91	[user.username for user in users], project
92	)
93	}
94
95
96	def add_projects_to_group(current_session, groupname, projects=None):	1✔
97	if not projects:	×
98	projects = []	×
99	grp = gp.get_group(current_session, groupname)	×
100	usrs = gp.get_group_users(current_session, groupname)	×
101	if not grp:	×
102	raise UserError("Error: group does not exist")	×
103	responses = []	×
104	for proj in projects:	×
105	try:	×
106	response = connect_project_to_group(current_session, grp, proj)	×
107	responses.append(response)	×
108	update_group_users_projects(current_session, grp, proj, usrs)	×
109	except Exception as e:	×
110	current_session.rollback()	×
111	raise e	×
112	return {"result": responses}	×
113
114
115	def disconnect_project_from_group(current_session, grp, projectname):	1✔
116	prj = pj.get_project(current_session, projectname)	×
117	if not prj:	×
118	return {"warning": ("Project {0} doesn't exist".format(projectname))}	×
119	return gp.remove_project_from_group(current_session, grp, prj)	×
120
121
122	def update_user_projects_within_group(	1✔
123	current_session, username, groupname, projectname
124	):
125	user_groups = us.get_user_groups(current_session, username)	×
126	"""
127	Simplified version for awg:
128	Users only have read permission, so just checking the
129	presence of the project in any of their other groups
130	suffices to keep the projec in the list.
131	In real life we should check permissions coming from all groups
132	and remove the specific ones comiing from groupname
133	"""
134	group_projects = [	×
135	gp.get_group_projects(current_session, group)
136	for group in user_groups["groups"]
137	if group != groupname
138	]
139
140	projects_to_keep = [item for sublist in group_projects for item in sublist]	×
141
142	if projectname not in projects_to_keep:	×
143	try:	×
144	us.remove_user_from_project(	×
145	current_session,
146	us.get_user(current_session, username),
147	pj.get_project(current_session, projectname),
148	)
149	except NotFound:	×
150	# somehow the user was not linked to that project
151	pass	×
152
153
154	def remove_projects_from_group(current_session, groupname, projects=None):	1✔
155	if not projects:	×
156	projects = []	×
157	grp = gp.get_group(current_session, groupname)	×
158	usrs = get_group_users(current_session, groupname)	×
159	users_names = [x["name"] for x in usrs["users"]]	×
160	if not grp:	×
161	raise UserError("Error: group does not exist")	×
162	responses = []	×
163	for proj in projects:	×
164	for usr in users_names:	×
165	update_user_projects_within_group(current_session, usr, groupname, proj)	×
166	response = disconnect_project_from_group(current_session, grp, proj)	×
167	responses.append(response)	×
168	return {"result": responses}	×
169
170
171	def get_group_projects(current_session, groupname):	1✔
172	return gp.get_group_projects(current_session, groupname)	×

uc-cdis / fence / 16203281716

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