8540651701

Committed 03 Apr 2024 02:38PM UTC coverage: 89.274% (+0.06%) from 89.212%

Build # 8540651701

Build Type

push

gihub-action

Committed by

web-flow

Commit Message

update to python 3.12 (#167)

Run Details

5718 of 6405 relevant lines covered (89.27%)

0.89 hits per line

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

84.62

/components/renku_data_services/git/gitlab.py

"""Gitlab API."""
from dataclasses import dataclass, field

import httpx

from renku_data_services.base_models import APIUser, GitlabAccessLevel
from renku_data_services.errors import errors
from renku_data_services.utils.core import get_ssl_context


@dataclass(kw_only=True)
class GitlabAPI:
    """Adapter for interacting with the gitlab API."""

    gitlab_url: str
    gitlab_graphql_url: str = field(init=False)

    def __post_init__(self):
        """Sets the graphql url for gitlab."""
        gitlab_url = self.gitlab_url

        if not gitlab_url.startswith("http") and "://" not in gitlab_url:
            raise errors.ConfigurationError(message=f"Gitlab URL should start with 'http(s)://', got: {gitlab_url}")

        gitlab_url = gitlab_url.rstrip("/")

        self.gitlab_graphql_url = f"{gitlab_url}/api/graphql"

    async def filter_projects_by_access_level(
        self, user: APIUser, project_ids: list[str], min_access_level: GitlabAccessLevel
    ) -> list[str]:
        """Filter projects this user can access in gitlab with at least access level."""

        if not user.access_token or not user.full_name:
            return []
        header = {"Authorization": f"Bearer {user.access_token}", "Content-Type": "application/json"}
        ids = ",".join(f'"gid://gitlab/Project/{id}"' for id in project_ids)
        query_body = f"""
                    pageInfo {{
                      hasNextPage
                    }}
                    nodes {{
                        id
                        projectMembers(search: "{user.full_name}") {{
                            nodes {{
                                user {{
                                    id
                                    name
                                }}
                                accessLevel {{
                                    integerValue
                                }}
                            }}
                        }}
                    }}
        """
        body = {
            "query": f"""{{
                projects(ids: [{ids}]) {{
                    {query_body}
                }}
            }}
            """
        }

        async def _query_gitlab_graphql(body, header):
            async with httpx.AsyncClient(verify=get_ssl_context()) as client:
                resp = await client.post(self.gitlab_graphql_url, json=body, headers=header, timeout=10)
            if resp.status_code != 200:
                raise errors.BaseError(message=f"Error querying Gitlab api {self.gitlab_graphql_url}: {resp.text}")
            result = resp.json()

            if "data" not in result or "projects" not in result["data"]:
                raise errors.BaseError(message=f"Got unexpected response from Gitlab: {result}")
            return result

        resp_body = await _query_gitlab_graphql(body, header)
        result: list[str] = []

        def _process_projects(resp_body, min_access_level, result):
            for project in resp_body["data"]["projects"]["nodes"]:
                if min_access_level != GitlabAccessLevel.PUBLIC:
                    if not project["projectMembers"]["nodes"]:
                        continue
                    if min_access_level == GitlabAccessLevel.ADMIN:
                        max_level = max(
                            n["accessLevel"]["integerValue"]
                            for n in project["projectMembers"]["nodes"]
                            if n["user"]["id"].rsplit("/", maxsplit=1)[-1] == user.id
                        )
                        if max_level < 30:
                            continue
                result.append(project["id"].rsplit("/", maxsplit=1)[-1])

        _process_projects(resp_body, min_access_level, result)
        page_info = resp_body["data"]["projects"]["pageInfo"]
        while page_info["hasNextPage"]:
            cursor = page_info["endCursor"]
            body = {
                "query": f"""{{
                    projects(ids: [{ids}], after: "{cursor}") {{
                        {query_body}
                    }}
                }}
                """
            }
            resp_body = await _query_gitlab_graphql(body, header)
            page_info = resp_body["data"]["projects"]["pageInfo"]
            _process_projects(resp_body, min_access_level, result)

        return result


@dataclass(kw_only=True)
class DummyGitlabAPI:
    """Dummy gitlab API.

    The user with name John Doe has admin access to project 123456 and member access to 999999.
    """

    _store = {
        "John Doe": {
            GitlabAccessLevel.MEMBER: ["999999", "123456"],
            GitlabAccessLevel.ADMIN: ["123456"],
        },
    }

    async def filter_projects_by_access_level(
        self, user: APIUser, project_ids: list[str], min_access_level: GitlabAccessLevel
    ) -> list[str]:
        """Filter projects this user can access in gitlab with at least access level."""
        if not user.access_token or not user.full_name:
            return []
        if min_access_level == GitlabAccessLevel.PUBLIC:
            return []
        user_projects = self._store.get(user.full_name, {}).get(min_access_level, [])
        return [p for p in project_ids if p in user_projects]

1	"""Gitlab API."""	1✔
2	from dataclasses import dataclass, field	1✔
3
4	import httpx	1✔
5
6	from renku_data_services.base_models import APIUser, GitlabAccessLevel	1✔
7	from renku_data_services.errors import errors	1✔
8	from renku_data_services.utils.core import get_ssl_context	1✔
9
10
11	@dataclass(kw_only=True)	1✔
12	class GitlabAPI:	1✔
13	"""Adapter for interacting with the gitlab API."""	1✔
14
15	gitlab_url: str	1✔
16	gitlab_graphql_url: str = field(init=False)	1✔
17
18	def __post_init__(self):	1✔
19	"""Sets the graphql url for gitlab."""
20	gitlab_url = self.gitlab_url	1✔
21
22	if not gitlab_url.startswith("http") and "://" not in gitlab_url:	1✔
23	raise errors.ConfigurationError(message=f"Gitlab URL should start with 'http(s)://', got: {gitlab_url}")	×
24
25	gitlab_url = gitlab_url.rstrip("/")	1✔
26
27	self.gitlab_graphql_url = f"{gitlab_url}/api/graphql"	1✔
28
29	async def filter_projects_by_access_level(	1✔
30	self, user: APIUser, project_ids: list[str], min_access_level: GitlabAccessLevel
31	) -> list[str]:
32	"""Filter projects this user can access in gitlab with at least access level."""
33
34	if not user.access_token or not user.full_name:	1✔
35	return []	×
36	header = {"Authorization": f"Bearer {user.access_token}", "Content-Type": "application/json"}	1✔
37	ids = ",".join(f'"gid://gitlab/Project/{id}"' for id in project_ids)	1✔
38	query_body = f"""	1✔
39	pageInfo {{
40	hasNextPage
41	}}
42	nodes {{
43	id
44	projectMembers(search: "{user.full_name}") {{
45	nodes {{
46	user {{
47	id
48	name
49	}}
50	accessLevel {{
51	integerValue
52	}}
53	}}
54	}}
55	}}
56	"""
57	body = {	1✔
58	"query": f"""{{
59	projects(ids: [{ids}]) {{
60	{query_body}
61	}}
62	}}
63	"""
64	}
65
66	async def _query_gitlab_graphql(body, header):	1✔
67	async with httpx.AsyncClient(verify=get_ssl_context()) as client:	1✔
68	resp = await client.post(self.gitlab_graphql_url, json=body, headers=header, timeout=10)	1✔
69	if resp.status_code != 200:	1✔
70	raise errors.BaseError(message=f"Error querying Gitlab api {self.gitlab_graphql_url}: {resp.text}")	×
71	result = resp.json()	1✔
72
73	if "data" not in result or "projects" not in result["data"]:	1✔
74	raise errors.BaseError(message=f"Got unexpected response from Gitlab: {result}")	×
75	return result	1✔
76
77	resp_body = await _query_gitlab_graphql(body, header)	1✔
78	result: list[str] = []	1✔
79
80	def _process_projects(resp_body, min_access_level, result):	1✔
81	for project in resp_body["data"]["projects"]["nodes"]:	1✔
82	if min_access_level != GitlabAccessLevel.PUBLIC:	1✔
83	if not project["projectMembers"]["nodes"]:	1✔
84	continue	1✔
85	if min_access_level == GitlabAccessLevel.ADMIN:	1✔
86	max_level = max(	1✔
87	n["accessLevel"]["integerValue"]
88	for n in project["projectMembers"]["nodes"]
89	if n["user"]["id"].rsplit("/", maxsplit=1)[-1] == user.id
90	)
91	if max_level < 30:	1✔
92	continue	1✔
93	result.append(project["id"].rsplit("/", maxsplit=1)[-1])	1✔
94
95	_process_projects(resp_body, min_access_level, result)	1✔
96	page_info = resp_body["data"]["projects"]["pageInfo"]	1✔
97	while page_info["hasNextPage"]:	1✔
98	cursor = page_info["endCursor"]	×
99	body = {	×
100	"query": f"""{{
101	projects(ids: [{ids}], after: "{cursor}") {{
102	{query_body}
103	}}
104	}}
105	"""
106	}
107	resp_body = await _query_gitlab_graphql(body, header)	×
108	page_info = resp_body["data"]["projects"]["pageInfo"]	×
109	_process_projects(resp_body, min_access_level, result)	×
110
111	return result	1✔
112
113
114	@dataclass(kw_only=True)	1✔
115	class DummyGitlabAPI:	1✔
116	"""Dummy gitlab API.	1✔
117
118	The user with name John Doe has admin access to project 123456 and member access to 999999.
119	"""
120
121	_store = {	1✔
122	"John Doe": {
123	GitlabAccessLevel.MEMBER: ["999999", "123456"],
124	GitlabAccessLevel.ADMIN: ["123456"],
125	},
126	}
127
128	async def filter_projects_by_access_level(	1✔
129	self, user: APIUser, project_ids: list[str], min_access_level: GitlabAccessLevel
130	) -> list[str]:
131	"""Filter projects this user can access in gitlab with at least access level."""
132	if not user.access_token or not user.full_name:	1✔
133	return []	1✔
134	if min_access_level == GitlabAccessLevel.PUBLIC:	1✔
135	return []	×
136	user_projects = self._store.get(user.full_name, {}).get(min_access_level, [])	1✔
137	return [p for p in project_ids if p in user_projects]	1✔

SwissDataScienceCenter / renku-data-services / 8540651701

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