18123243513

Committed 30 Sep 2025 08:10AM UTC coverage: 86.702% (-0.01%) from 86.714%

Build # 18123243513

Build Type

Pull #1019

github

Committed by

web-flow

Commit Message

Merge e726c4543 into 0690bab65

Pull Request Pull Request #1019: feat: Attempt to support dockerhub private images

Run Details

70 of 101 new or added lines in 9 files covered. (69.31%)

106 existing lines in 6 files now uncovered.

22357 of 25786 relevant lines covered (86.7%)

1.52 hits per line

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

94.21

/components/renku_data_services/connected_services/models.py

"""Models for connected services."""

from dataclasses import dataclass
from datetime import UTC, datetime
from enum import StrEnum
from typing import Any

from ulid import ULID

from renku_data_services.users.db import APIUser


class ProviderKind(StrEnum):
    """The kind of platform we connnect to."""

    gitlab = "gitlab"
    github = "github"
    drive = "drive"
    onedrive = "onedrive"
    dropbox = "dropbox"
    generic_oidc = "generic_oidc"
    dockerhub = "dockerhub"


class ConnectionStatus(StrEnum):
    """The status of a connection."""

    connected = "connected"
    pending = "pending"


class RepositorySelection(StrEnum):
    """The repository selection for GitHub applications."""

    all = "all"
    selected = "selected"


@dataclass(frozen=True, eq=True, kw_only=True)
class UnsavedOAuth2Client:
    """OAuth2 Client model."""

    id: str
    app_slug: str
    kind: ProviderKind
    client_id: str
    client_secret: str | None
    display_name: str
    scope: str
    url: str
    use_pkce: bool
    image_registry_url: str | None = None
    oidc_issuer_url: str | None = None


@dataclass(frozen=True, eq=True, kw_only=True)
class OAuth2Client(UnsavedOAuth2Client):
    """OAuth2 Client model."""

    created_by_id: str
    creation_date: datetime
    updated_at: datetime


@dataclass(frozen=True, eq=True, kw_only=True)
class OAuth2ClientPatch:
    """Model for changes requested on a OAuth2 Client."""

    kind: ProviderKind | None
    app_slug: str | None
    client_id: str | None
    client_secret: str | None
    display_name: str | None
    scope: str | None
    url: str | None
    use_pkce: bool | None
    image_registry_url: str | None
    oidc_issuer_url: str | None


@dataclass(frozen=True, eq=True, kw_only=True)
class OAuth2Connection:
    """OAuth2 connection model."""

    id: ULID
    provider_id: str
    status: ConnectionStatus

    def is_connected(self) -> bool:
        """Returns whether this connection is in status 'connected'."""
        return self.status == ConnectionStatus.connected


@dataclass(frozen=True, eq=True, kw_only=True)
class ConnectedAccount:
    """OAuth2 connected account model."""

    username: str
    web_url: str


class OAuth2TokenSet(dict):
    """OAuth2 token set model."""

    @classmethod
    def from_dict(cls, token_set: dict[str, Any]) -> "OAuth2TokenSet":
        """Create an OAuth2 token set from a dictionary."""
        if isinstance(token_set, dict) and not isinstance(token_set, cls):
            return cls(token_set)
        return token_set

    def dump_for_api(self) -> dict[str, Any]:
        """Expose the access token and other token metadata for API consumption."""
        data = dict((k, v) for k, v in self.items() if k != "refresh_token")
        if self.expires_at_iso is not None:
            data["expires_at_iso"] = self.expires_at_iso
        return data

    @property
    def username(self) -> str | None:
        """Return the username property."""
        return self.get("username")

    @property
    def access_token(self) -> str | None:
        """Returns the access token."""
        return self.get("access_token")

    @property
    def refresh_token(self) -> str | None:
        """Returns the refresh token."""
        return self.get("refresh_token")

    @property
    def expires_at(self) -> int | None:
        """Returns the access token expiry date."""
        return self.get("expires_at")

    @property
    def expires_at_iso(self) -> str | None:
        """Returns the access token expiry date."""
        if self.expires_at is None:
            return None
        return datetime.fromtimestamp(self.expires_at, UTC).isoformat()


@dataclass(frozen=True, eq=True, kw_only=True)
class AppInstallation:
    """A GitHub app installation."""

    id: int
    account_login: str
    account_web_url: str
    repository_selection: RepositorySelection
    suspended_at: datetime | None = None


@dataclass(frozen=True, eq=True, kw_only=True)
class AppInstallationList:
    """GitHub app installation list."""

    total_count: int
    installations: list[AppInstallation]


@dataclass(frozen=True, eq=True)
class ConnectedUser:
    """A user and the corresponding oauth2 connection."""

    connection: OAuth2Connection
    user: APIUser

    def is_connected(self) -> bool:
        """Returns whether the connection is in status 'connected'."""
        return self.connection.is_connected()


@dataclass(frozen=True, eq=True)
class ImageProvider:
    """Result when retrieving provider information for an image."""

    provider: OAuth2Client
    connected_user: ConnectedUser | None
    registry_url: str

    def is_connected(self) -> bool:
        """Returns whether the connection exists and is in status 'connected'."""
        return self.connected_user is not None and self.connected_user.is_connected()

    @property
    def connection(self) -> OAuth2Connection | None:
        """Return the connection if present."""
        if self.connected_user:
            return self.connected_user.connection
        else:
            return None

    def __str__(self) -> str:
        conn = f"connection={self.connection.id}" if self.connection else "connection=None"
        return f"ImageProvider(provider={self.provider.id}/{self.provider.kind}, {conn})"

1	"""Models for connected services."""
2
3	from dataclasses import dataclass	2✔
4	from datetime import UTC, datetime	2✔
5	from enum import StrEnum	2✔
6	from typing import Any	2✔
7
8	from ulid import ULID	2✔
9
10	from renku_data_services.users.db import APIUser	2✔
11
12
13	class ProviderKind(StrEnum):	2✔
14	"""The kind of platform we connnect to."""
15
16	gitlab = "gitlab"	2✔
17	github = "github"	2✔
18	drive = "drive"	2✔
19	onedrive = "onedrive"	2✔
20	dropbox = "dropbox"	2✔
21	generic_oidc = "generic_oidc"	2✔
22	dockerhub = "dockerhub"	2✔
23
24
25	class ConnectionStatus(StrEnum):	2✔
26	"""The status of a connection."""
27
28	connected = "connected"	2✔
29	pending = "pending"	2✔
30
31
32	class RepositorySelection(StrEnum):	2✔
33	"""The repository selection for GitHub applications."""
34
35	all = "all"	2✔
36	selected = "selected"	2✔
37
38
39	@dataclass(frozen=True, eq=True, kw_only=True)	2✔
40	class UnsavedOAuth2Client:	2✔
41	"""OAuth2 Client model."""
42
43	id: str	2✔
44	app_slug: str	2✔
45	kind: ProviderKind	2✔
46	client_id: str	2✔
47	client_secret: str \| None	2✔
48	display_name: str	2✔
49	scope: str	2✔
50	url: str	2✔
51	use_pkce: bool	2✔
52	image_registry_url: str \| None = None	2✔
53	oidc_issuer_url: str \| None = None	2✔
54
55
56	@dataclass(frozen=True, eq=True, kw_only=True)	2✔
57	class OAuth2Client(UnsavedOAuth2Client):	2✔
58	"""OAuth2 Client model."""
59
60	created_by_id: str	2✔
61	creation_date: datetime	2✔
62	updated_at: datetime	2✔
63
64
65	@dataclass(frozen=True, eq=True, kw_only=True)	2✔
66	class OAuth2ClientPatch:	2✔
67	"""Model for changes requested on a OAuth2 Client."""
68
69	kind: ProviderKind \| None	2✔
70	app_slug: str \| None	2✔
71	client_id: str \| None	2✔
72	client_secret: str \| None	2✔
73	display_name: str \| None	2✔
74	scope: str \| None	2✔
75	url: str \| None	2✔
76	use_pkce: bool \| None	2✔
77	image_registry_url: str \| None	2✔
78	oidc_issuer_url: str \| None	2✔
79
80
81	@dataclass(frozen=True, eq=True, kw_only=True)	2✔
82	class OAuth2Connection:	2✔
83	"""OAuth2 connection model."""
84
85	id: ULID	2✔
86	provider_id: str	2✔
87	status: ConnectionStatus	2✔
88
89	def is_connected(self) -> bool:	2✔
90	"""Returns whether this connection is in status 'connected'."""
91	return self.status == ConnectionStatus.connected	1✔
92
93
94	@dataclass(frozen=True, eq=True, kw_only=True)	2✔
95	class ConnectedAccount:	2✔
96	"""OAuth2 connected account model."""
97
98	username: str	2✔
99	web_url: str	2✔
100
101
102	class OAuth2TokenSet(dict):	2✔
103	"""OAuth2 token set model."""
104
105	@classmethod	2✔
106	def from_dict(cls, token_set: dict[str, Any]) -> "OAuth2TokenSet":	2✔
107	"""Create an OAuth2 token set from a dictionary."""
108	if isinstance(token_set, dict) and not isinstance(token_set, cls):	2✔
109	return cls(token_set)	2✔
110	return token_set	1✔
111
112	def dump_for_api(self) -> dict[str, Any]:	2✔
113	"""Expose the access token and other token metadata for API consumption."""
114	data = dict((k, v) for k, v in self.items() if k != "refresh_token")	1✔
115	if self.expires_at_iso is not None:	1✔
116	data["expires_at_iso"] = self.expires_at_iso	1✔
117	return data	1✔
118
119	@property	2✔
120	def username(self) -> str \| None:	2✔
121	"""Return the username property."""
NEW 122	return self.get("username")	×
123
124	@property	2✔
125	def access_token(self) -> str \| None:	2✔
126	"""Returns the access token."""
127	return self.get("access_token")	2✔
128
129	@property	2✔
130	def refresh_token(self) -> str \| None:	2✔
131	"""Returns the refresh token."""
132	return self.get("refresh_token")	2✔
133
134	@property	2✔
135	def expires_at(self) -> int \| None:	2✔
136	"""Returns the access token expiry date."""
137	return self.get("expires_at")	1✔
138
139	@property	2✔
140	def expires_at_iso(self) -> str \| None:	2✔
141	"""Returns the access token expiry date."""
142	if self.expires_at is None:	1✔
143	return None	×
144	return datetime.fromtimestamp(self.expires_at, UTC).isoformat()	1✔
145
146
147	@dataclass(frozen=True, eq=True, kw_only=True)	2✔
148	class AppInstallation:	2✔
149	"""A GitHub app installation."""
150
151	id: int	2✔
152	account_login: str	2✔
153	account_web_url: str	2✔
154	repository_selection: RepositorySelection	2✔
155	suspended_at: datetime \| None = None	2✔
156
157
158	@dataclass(frozen=True, eq=True, kw_only=True)	2✔
159	class AppInstallationList:	2✔
160	"""GitHub app installation list."""
161
162	total_count: int	2✔
163	installations: list[AppInstallation]	2✔
164
165
166	@dataclass(frozen=True, eq=True)	2✔
167	class ConnectedUser:	2✔
168	"""A user and the corresponding oauth2 connection."""
169
170	connection: OAuth2Connection	2✔
171	user: APIUser	2✔
172
173	def is_connected(self) -> bool:	2✔
174	"""Returns whether the connection is in status 'connected'."""
175	return self.connection.is_connected()	1✔
176
177
178	@dataclass(frozen=True, eq=True)	2✔
179	class ImageProvider:	2✔
180	"""Result when retrieving provider information for an image."""
181
182	provider: OAuth2Client	2✔
183	connected_user: ConnectedUser \| None	2✔
184	registry_url: str	2✔
185
186	def is_connected(self) -> bool:	2✔
187	"""Returns whether the connection exists and is in status 'connected'."""
188	return self.connected_user is not None and self.connected_user.is_connected()	1✔
189
190	@property	2✔
191	def connection(self) -> OAuth2Connection \| None:	2✔
192	"""Return the connection if present."""
193	if self.connected_user:	×
194	return self.connected_user.connection	×
195	else:
196	return None	×
197
198	def __str__(self) -> str:	2✔
199	conn = f"connection={self.connection.id}" if self.connection else "connection=None"	×
200	return f"ImageProvider(provider={self.provider.id}/{self.provider.kind}, {conn})"	×

SwissDataScienceCenter / renku-data-services / 18123243513

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