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

29.41

fence/resources/admin/admin_projects.py

from fence.resources import project as pj
from flask import current_app

__all__ = [
    "get_project_info",
    "get_all_projects",
    "create_project",
    "delete_project",
    "create_bucket_on_project",
    "delete_bucket_on_project",
    "list_buckets_on_project",
]


def get_project_info(current_session, project_name):
    """
    Return the information associated with a project
    Returns a dictionary.
    """
    return pj.get_project_info(current_session, project_name)


def get_all_projects(current_session):
    """
    Return the information associated with a project
    Returns a dictionary.
    """
    return pj.get_all_projects(current_session)


def create_project(current_session, projectname, authid, storageaccesses):
    """
    Create a project with the specified auth_id and
    storage access.
    Returns a dictionary.
    """
    if pj.create_project(current_session, projectname, authid, storageaccesses):
        return {"result": "success"}


def delete_project(current_session, project_name):
    """
    Remove a project. All buckets must be deleted
    before this oepration can be called.
    Returns a dictionary.
    """
    response = pj.delete_project(current_session, project_name)
    if response["result"] == "success":
        for user in response["users_to_remove"]:
            current_app.storage_manager.delete_user(user[0].backend, user[1])
        return {"result": "success"}


def create_bucket_on_project(current_session, project_name, bucket_name, provider_name):
    """
    Create a bucket on userdatamodel database and
    on the cloud provider and associate it with the project.
    Returns a dictionary.
    """
    response = pj.create_bucket_on_project(
        current_session, project_name, bucket_name, provider_name
    )
    project = pj.get_project(current_session, project_name)
    if response["result"] == "success":
        current_app.storage_manager.create_bucket(
            response["provider"].name, current_session, response["bucket"].name, project
        )
        for user_pair in response["users_to_update"]:
            current_app.storage_manager.update_bucket_acl(
                response["provider"].name,
                response["bucket"],
                (user_pair[0], user_pair[1]),
            )
        return {"result": "success"}
    else:
        return response


def delete_bucket_on_project(current_session, project_name, bucket_name):
    """
    Remove a bucket from a project, both on the userdatamodel
    and on the storage associated with that bucket.
    Returns a dictionary.
    """
    response = pj.delete_bucket_on_project(current_session, project_name, bucket_name)
    if response["result"] == "success":
        current_app.storage_manager.delete_bucket(
            response["provider"].name, bucket_name
        )
        return {"result": "success"}
    else:
        current_app.storage_manager.delete_bucket(
            response["provider"].name, bucket_name
        )
        return {"result": response["result"]}


def list_buckets_on_project(current_session, project_name):
    """
    Retrieve the buckets associated with a project.
    Returns a dictionary.
    """
    return pj.list_buckets_on_project(current_session, project_name)

1	from fence.resources import project as pj	1✔
2	from flask import current_app	1✔
3
4	__all__ = [	1✔
5	"get_project_info",
6	"get_all_projects",
7	"create_project",
8	"delete_project",
9	"create_bucket_on_project",
10	"delete_bucket_on_project",
11	"list_buckets_on_project",
12	]
13
14
15	def get_project_info(current_session, project_name):	1✔
16	"""
17	Return the information associated with a project
18	Returns a dictionary.
19	"""
20	return pj.get_project_info(current_session, project_name)	×
21
22
23	def get_all_projects(current_session):	1✔
24	"""
25	Return the information associated with a project
26	Returns a dictionary.
27	"""
28	return pj.get_all_projects(current_session)	×
29
30
31	def create_project(current_session, projectname, authid, storageaccesses):	1✔
32	"""
33	Create a project with the specified auth_id and
34	storage access.
35	Returns a dictionary.
36	"""
37	if pj.create_project(current_session, projectname, authid, storageaccesses):	×
38	return {"result": "success"}	×
39
40
41	def delete_project(current_session, project_name):	1✔
42	"""
43	Remove a project. All buckets must be deleted
44	before this oepration can be called.
45	Returns a dictionary.
46	"""
47	response = pj.delete_project(current_session, project_name)	×
48	if response["result"] == "success":	×
49	for user in response["users_to_remove"]:	×
50	current_app.storage_manager.delete_user(user[0].backend, user[1])	×
51	return {"result": "success"}	×
52
53
54	def create_bucket_on_project(current_session, project_name, bucket_name, provider_name):	1✔
55	"""
56	Create a bucket on userdatamodel database and
57	on the cloud provider and associate it with the project.
58	Returns a dictionary.
59	"""
60	response = pj.create_bucket_on_project(	×
61	current_session, project_name, bucket_name, provider_name
62	)
63	project = pj.get_project(current_session, project_name)	×
64	if response["result"] == "success":	×
65	current_app.storage_manager.create_bucket(	×
66	response["provider"].name, current_session, response["bucket"].name, project
67	)
68	for user_pair in response["users_to_update"]:	×
69	current_app.storage_manager.update_bucket_acl(	×
70	response["provider"].name,
71	response["bucket"],
72	(user_pair[0], user_pair[1]),
73	)
74	return {"result": "success"}	×
75	else:
76	return response	×
77
78
79	def delete_bucket_on_project(current_session, project_name, bucket_name):	1✔
80	"""
81	Remove a bucket from a project, both on the userdatamodel
82	and on the storage associated with that bucket.
83	Returns a dictionary.
84	"""
85	response = pj.delete_bucket_on_project(current_session, project_name, bucket_name)	×
86	if response["result"] == "success":	×
87	current_app.storage_manager.delete_bucket(	×
88	response["provider"].name, bucket_name
89	)
90	return {"result": "success"}	×
91	else:
92	current_app.storage_manager.delete_bucket(	×
93	response["provider"].name, bucket_name
94	)
95	return {"result": response["result"]}	×
96
97
98	def list_buckets_on_project(current_session, project_name):	1✔
99	"""
100	Retrieve the buckets associated with a project.
101	Returns a dictionary.
102	"""
103	return pj.list_buckets_on_project(current_session, project_name)	×

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