19430749134

Committed 17 Nov 2025 01:10PM UTC coverage: 86.399% (-0.04%) from 86.437%

Build # 19430749134

Build Type

Pull #1115

github

Committed by

web-flow

Commit Message

Merge 06a5cf42a into 88fbe3ee5

Pull Request Pull Request #1115: fix: send indentify() call only when necessary

Run Details

80 of 95 new or added lines in 6 files covered. (84.21%)

6 existing lines in 5 files now uncovered.

23103 of 26740 relevant lines covered (86.4%)

1.52 hits per line

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

83.54

/components/renku_data_services/users/models.py

"""Base models for users."""

from __future__ import annotations

import json
import re
from collections.abc import Iterable
from dataclasses import dataclass
from datetime import UTC, datetime
from enum import Enum
from typing import Any, NamedTuple

from pydantic import BaseModel, Field

from renku_data_services.app_config import logging
from renku_data_services.base_models import errors
from renku_data_services.namespace.models import UserNamespace

logger = logging.getLogger(__name__)


class KeycloakEvent(Enum):
    """The Keycloak user events that result from the user registering or updating their personal information."""

    REGISTER = "REGISTER"
    UPDATE_PROFILE = "UPDATE_PROFILE"


class KeycloakAdminEvent(Enum):
    """The Keycloak admin events used to keep users up to date."""

    DELETE = "DELETE"
    UPDATE = "UPDATE"
    CREATE = "CREATE"


@dataclass
class UserInfoFieldUpdate:
    """An update of a specific field of user information."""

    user_id: str
    timestamp_utc: datetime
    field_name: str
    new_value: str
    old_value: str | None = None

    @classmethod
    def from_json_user_events(cls, val: Iterable[dict[str, Any]]) -> list[UserInfoFieldUpdate]:
        """Generate a list of updates from a json response from Keycloak."""
        output: list[UserInfoFieldUpdate] = []
        for event in val:
            details = event.get("details")
            user_id = event.get("userId")
            timestamp_epoch = event.get("time")
            if not timestamp_epoch:
                logger.warning("Expected response from keycloak events to have a time field.")
                continue
            timestamp_utc = datetime.utcfromtimestamp(timestamp_epoch / 1000)
            if not details:
                logger.warning("Expected response from keycloak events to have a details field.")
                continue
            if not user_id:
                logger.warning("Expected response from keycloak events to have a userId field.")
                continue
            match event.get("type"):
                case KeycloakEvent.REGISTER.value:
                    first_name = details.get("first_name")
                    last_name = details.get("last_name")
                    email = details.get("email")
                    if email:
                        output.append(
                            UserInfoFieldUpdate(
                                field_name="email",
                                new_value=email,
                                timestamp_utc=timestamp_utc,
                                user_id=user_id,
                            )
                        )
                    if first_name:
                        output.append(
                            UserInfoFieldUpdate(
                                field_name="first_name",
                                new_value=first_name,
                                timestamp_utc=timestamp_utc,
                                user_id=user_id,
                            )
                        )
                    if last_name:
                        output.append(
                            UserInfoFieldUpdate(
                                field_name="last_name",
                                new_value=last_name,
                                timestamp_utc=timestamp_utc,
                                user_id=user_id,
                            )
                        )
                case KeycloakEvent.UPDATE_PROFILE.value:
                    first_name = details.get("updated_first_name")
                    last_name = details.get("updated_last_name")
                    email = details.get("updated_email")
                    if first_name:
                        old_value = details.get("previous_first_name")
                        output.append(
                            UserInfoFieldUpdate(
                                field_name="first_name",
                                new_value=first_name,
                                old_value=old_value,
                                timestamp_utc=timestamp_utc,
                                user_id=user_id,
                            )
                        )
                    if last_name:
                        old_value = details.get("previous_last_name")
                        output.append(
                            UserInfoFieldUpdate(
                                field_name="last_name",
                                new_value=last_name,
                                old_value=old_value,
                                timestamp_utc=timestamp_utc,
                                user_id=user_id,
                            )
                        )
                    if email:
                        old_value = details.get("previous_email")
                        output.append(
                            UserInfoFieldUpdate(
                                field_name="email",
                                new_value=email,
                                old_value=old_value,
                                timestamp_utc=timestamp_utc,
                                user_id=user_id,
                            )
                        )
                case _:
                    logger.warning(f"Skipping unknown event when parsing Keycloak user events: {event.get('type')}")
        return output

    @classmethod
    def from_json_admin_events(cls, val: Iterable[dict[str, Any]]) -> list[UserInfoFieldUpdate]:
        """Generate a list of updates from a json response from Keycloak."""
        output: list[UserInfoFieldUpdate] = []
        for event in val:
            timestamp_epoch = event.get("time")
            if not timestamp_epoch:
                logger.warning("Expected response from keycloak events to have a time field.")
                continue
            timestamp_utc = datetime.utcfromtimestamp(timestamp_epoch / 1000)
            resource = event.get("resourceType")
            if resource != "USER":
                continue
            operation = event.get("operationType")
            if not operation:
                logger.warning(f"Skipping unknown operation {operation}")
                continue
            resource_path = event.get("resourcePath")
            if not resource_path:
                logger.warning("Cannot find resource path in events response")
                continue
            user_id_match = re.match(r"^users/(.+)", resource_path)
            if not user_id_match:
                logger.warning("No match for user ID in resource path")
                continue
            user_id = user_id_match.group(1)
            if not isinstance(user_id, str) or user_id is None or len(user_id) == 0:
                logger.warning("Could not extract user ID from match in resource path")
                continue
            match operation:
                case KeycloakAdminEvent.CREATE.value | KeycloakAdminEvent.UPDATE.value:
                    payload = json.loads(event.get("representation", "{}"))
                    first_name = payload.get("firstName")
                    if first_name:
                        output.append(
                            UserInfoFieldUpdate(
                                field_name="first_name",
                                new_value=first_name,
                                timestamp_utc=timestamp_utc,
                                user_id=user_id,
                            )
                        )
                    last_name = payload.get("lastName")
                    if last_name:
                        output.append(
                            UserInfoFieldUpdate(
                                field_name="last_name",
                                new_value=last_name,
                                timestamp_utc=timestamp_utc,
                                user_id=user_id,
                            )
                        )
                    email = payload.get("email")
                    if email:
                        output.append(
                            UserInfoFieldUpdate(
                                field_name="email",
                                new_value=email,
                                timestamp_utc=timestamp_utc,
                                user_id=user_id,
                            )
                        )
                case KeycloakAdminEvent.DELETE.value:
                    output.append(
                        UserInfoFieldUpdate(
                            field_name="email",
                            new_value="",
                            timestamp_utc=timestamp_utc,
                            user_id=user_id,
                        )
                    )
                case _:
                    logger.warning(f"Skipping unknown admin event operation when parsing Keycloak events: {operation}")
        return output


@dataclass(eq=True, frozen=True, kw_only=True)
class UnsavedUserInfo:
    """Keycloak user."""

    id: str
    first_name: str | None = None
    last_name: str | None = None
    email: str | None = None

    @classmethod
    def from_kc_user_payload(cls, payload: dict[str, Any]) -> UnsavedUserInfo:
        """Create a user object from the user payload from the Keycloak admin API."""
        return UnsavedUserInfo(
            id=payload["id"],
            first_name=payload.get("firstName"),
            last_name=payload.get("lastName"),
            email=payload.get("email"),
        )

    def to_keycloak_dict(self) -> dict[str, Any]:
        """Create a payload that would have been created by Keycloak for this user, used only for testing."""

        return {
            "id": self.id,
            "createdTimestamp": int(datetime.now(UTC).timestamp() * 1000),
            "username": self.email,
            "enabled": True,
            "emailVerified": False,
            "firstName": self.first_name,
            "lastName": self.last_name,
            "email": self.email,
            "access": {
                "manageGroupMembership": True,
                "view": True,
                "mapRoles": True,
                "impersonate": True,
                "manage": True,
            },
            "bruteForceStatus": {
                "numFailures": 0,
                "disabled": False,
                "lastIPFailure": "n/a",
                "lastFailure": 0,
            },
        }


@dataclass(eq=True, frozen=True, kw_only=True)
class UserInfo(UnsavedUserInfo):
    """A tuple used to convey information about a user and their namespace."""

    namespace: UserNamespace

    def requires_update(self, current_user_info: UnsavedUserInfo) -> bool:
        """Returns true if the data self does not match the current_user_info."""
        if self.id != current_user_info.id:
            raise errors.ValidationError(message="Cannot check updates on two different users.")
        self_as_unsaved = UnsavedUserInfo(
            id=self.id,
            first_name=self.first_name,
            last_name=self.last_name,
            email=self.email,
        )
        return self_as_unsaved != current_user_info


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

    first_name: str | None = None
    last_name: str | None = None
    email: str | None = None

    @classmethod
    def from_unsaved_user_info(cls, user: UnsavedUserInfo) -> UserPatch:
        """Create a user patch from a UnsavedUserInfo instance."""
        return UserPatch(
            first_name=user.first_name,
            last_name=user.last_name,
            email=user.email,
        )


@dataclass
class DeletedUser:
    """A user that was deleted from the database."""

    id: str


class UserInfoUpdate(NamedTuple):
    """Used to convey information about an update of a user or their namespace."""

    old: UserInfo | None
    new: UserInfo


class PinnedProjects(BaseModel):
    """Pinned projects model."""

    project_slugs: list[str] | None = None

    @classmethod
    def from_dict(cls, data: dict) -> PinnedProjects:
        """Create model from a dict object."""
        return cls(project_slugs=data.get("project_slugs"))


class UserPreferences(BaseModel):
    """User preferences model."""

    user_id: str = Field(min_length=3)
    pinned_projects: PinnedProjects
    show_project_migration_banner: bool = True

1	"""Base models for users."""
2
3	from __future__ import annotations	2✔
4
5	import json	2✔
6	import re	2✔
7	from collections.abc import Iterable	2✔
8	from dataclasses import dataclass	2✔
9	from datetime import UTC, datetime	2✔
10	from enum import Enum	2✔
11	from typing import Any, NamedTuple	2✔
12
13	from pydantic import BaseModel, Field	2✔
14
15	from renku_data_services.app_config import logging	2✔
16	from renku_data_services.base_models import errors	2✔
17	from renku_data_services.namespace.models import UserNamespace	2✔
18
19	logger = logging.getLogger(__name__)	2✔
20
21
22	class KeycloakEvent(Enum):	2✔
23	"""The Keycloak user events that result from the user registering or updating their personal information."""
24
25	REGISTER = "REGISTER"	2✔
26	UPDATE_PROFILE = "UPDATE_PROFILE"	2✔
27
28
29	class KeycloakAdminEvent(Enum):	2✔
30	"""The Keycloak admin events used to keep users up to date."""
31
32	DELETE = "DELETE"	2✔
33	UPDATE = "UPDATE"	2✔
34	CREATE = "CREATE"	2✔
35
36
37	@dataclass	2✔
38	class UserInfoFieldUpdate:	2✔
39	"""An update of a specific field of user information."""
40
41	user_id: str	2✔
42	timestamp_utc: datetime	2✔
43	field_name: str	2✔
44	new_value: str	2✔
45	old_value: str \| None = None	2✔
46
47	@classmethod	2✔
48	def from_json_user_events(cls, val: Iterable[dict[str, Any]]) -> list[UserInfoFieldUpdate]:	2✔
49	"""Generate a list of updates from a json response from Keycloak."""
50	output: list[UserInfoFieldUpdate] = []	1✔
51	for event in val:	1✔
52	details = event.get("details")	1✔
53	user_id = event.get("userId")	1✔
54	timestamp_epoch = event.get("time")	1✔
55	if not timestamp_epoch:	1✔
56	logger.warning("Expected response from keycloak events to have a time field.")	×
57	continue	×
58	timestamp_utc = datetime.utcfromtimestamp(timestamp_epoch / 1000)	1✔
59	if not details:	1✔
60	logger.warning("Expected response from keycloak events to have a details field.")	×
61	continue	×
62	if not user_id:	1✔
63	logger.warning("Expected response from keycloak events to have a userId field.")	×
64	continue	×
65	match event.get("type"):	1✔
66	case KeycloakEvent.REGISTER.value:	1✔
67	first_name = details.get("first_name")	1✔
68	last_name = details.get("last_name")	1✔
69	email = details.get("email")	1✔
70	if email:	1✔
71	output.append(	1✔
72	UserInfoFieldUpdate(
73	field_name="email",
74	new_value=email,
75	timestamp_utc=timestamp_utc,
76	user_id=user_id,
77	)
78	)
79	if first_name:	1✔
80	output.append(	1✔
81	UserInfoFieldUpdate(
82	field_name="first_name",
83	new_value=first_name,
84	timestamp_utc=timestamp_utc,
85	user_id=user_id,
86	)
87	)
88	if last_name:	1✔
89	output.append(	1✔
90	UserInfoFieldUpdate(
91	field_name="last_name",
92	new_value=last_name,
93	timestamp_utc=timestamp_utc,
94	user_id=user_id,
95	)
96	)
97	case KeycloakEvent.UPDATE_PROFILE.value:	1✔
98	first_name = details.get("updated_first_name")	1✔
99	last_name = details.get("updated_last_name")	1✔
100	email = details.get("updated_email")	1✔
101	if first_name:	1✔
102	old_value = details.get("previous_first_name")	1✔
103	output.append(	1✔
104	UserInfoFieldUpdate(
105	field_name="first_name",
106	new_value=first_name,
107	old_value=old_value,
108	timestamp_utc=timestamp_utc,
109	user_id=user_id,
110	)
111	)
112	if last_name:	1✔
113	old_value = details.get("previous_last_name")	×
114	output.append(	×
115	UserInfoFieldUpdate(
116	field_name="last_name",
117	new_value=last_name,
118	old_value=old_value,
119	timestamp_utc=timestamp_utc,
120	user_id=user_id,
121	)
122	)
123	if email:	1✔
124	old_value = details.get("previous_email")	×
125	output.append(	×
126	UserInfoFieldUpdate(
127	field_name="email",
128	new_value=email,
129	old_value=old_value,
130	timestamp_utc=timestamp_utc,
131	user_id=user_id,
132	)
133	)
134	case _:	×
135	logger.warning(f"Skipping unknown event when parsing Keycloak user events: {event.get('type')}")	×
136	return output	1✔
137
138	@classmethod	2✔
139	def from_json_admin_events(cls, val: Iterable[dict[str, Any]]) -> list[UserInfoFieldUpdate]:	2✔
140	"""Generate a list of updates from a json response from Keycloak."""
141	output: list[UserInfoFieldUpdate] = []	1✔
142	for event in val:	1✔
143	timestamp_epoch = event.get("time")	1✔
144	if not timestamp_epoch:	1✔
145	logger.warning("Expected response from keycloak events to have a time field.")	×
146	continue	×
147	timestamp_utc = datetime.utcfromtimestamp(timestamp_epoch / 1000)	1✔
148	resource = event.get("resourceType")	1✔
149	if resource != "USER":	1✔
150	continue	×
151	operation = event.get("operationType")	1✔
152	if not operation:	1✔
153	logger.warning(f"Skipping unknown operation {operation}")	×
154	continue	×
155	resource_path = event.get("resourcePath")	1✔
156	if not resource_path:	1✔
157	logger.warning("Cannot find resource path in events response")	×
158	continue	×
159	user_id_match = re.match(r"^users/(.+)", resource_path)	1✔
160	if not user_id_match:	1✔
161	logger.warning("No match for user ID in resource path")	×
162	continue	×
163	user_id = user_id_match.group(1)	1✔
164	if not isinstance(user_id, str) or user_id is None or len(user_id) == 0:	1✔
165	logger.warning("Could not extract user ID from match in resource path")	×
166	continue	×
167	match operation:	1✔
168	case KeycloakAdminEvent.CREATE.value \| KeycloakAdminEvent.UPDATE.value:	1✔
169	payload = json.loads(event.get("representation", "{}"))	1✔
170	first_name = payload.get("firstName")	1✔
171	if first_name:	1✔
172	output.append(	1✔
173	UserInfoFieldUpdate(
174	field_name="first_name",
175	new_value=first_name,
176	timestamp_utc=timestamp_utc,
177	user_id=user_id,
178	)
179	)
180	last_name = payload.get("lastName")	1✔
181	if last_name:	1✔
182	output.append(	1✔
183	UserInfoFieldUpdate(
184	field_name="last_name",
185	new_value=last_name,
186	timestamp_utc=timestamp_utc,
187	user_id=user_id,
188	)
189	)
190	email = payload.get("email")	1✔
191	if email:	1✔
192	output.append(	1✔
193	UserInfoFieldUpdate(
194	field_name="email",
195	new_value=email,
196	timestamp_utc=timestamp_utc,
197	user_id=user_id,
198	)
199	)
200	case KeycloakAdminEvent.DELETE.value:	1✔
201	output.append(	1✔
202	UserInfoFieldUpdate(
203	field_name="email",
204	new_value="",
205	timestamp_utc=timestamp_utc,
206	user_id=user_id,
207	)
208	)
209	case _:	×
210	logger.warning(f"Skipping unknown admin event operation when parsing Keycloak events: {operation}")	×
211	return output	1✔
212
213
214	@dataclass(eq=True, frozen=True, kw_only=True)	2✔
215	class UnsavedUserInfo:	2✔
216	"""Keycloak user."""
217
218	id: str	2✔
219	first_name: str \| None = None	2✔
220	last_name: str \| None = None	2✔
221	email: str \| None = None	2✔
222
223	@classmethod	2✔
224	def from_kc_user_payload(cls, payload: dict[str, Any]) -> UnsavedUserInfo:	2✔
225	"""Create a user object from the user payload from the Keycloak admin API."""
226	return UnsavedUserInfo(	2✔
227	id=payload["id"],
228	first_name=payload.get("firstName"),
229	last_name=payload.get("lastName"),
230	email=payload.get("email"),
231	)
232
233	def to_keycloak_dict(self) -> dict[str, Any]:	2✔
234	"""Create a payload that would have been created by Keycloak for this user, used only for testing."""
235
236	return {	2✔
237	"id": self.id,
238	"createdTimestamp": int(datetime.now(UTC).timestamp() * 1000),
239	"username": self.email,
240	"enabled": True,
241	"emailVerified": False,
242	"firstName": self.first_name,
243	"lastName": self.last_name,
244	"email": self.email,
245	"access": {
246	"manageGroupMembership": True,
247	"view": True,
248	"mapRoles": True,
249	"impersonate": True,
250	"manage": True,
251	},
252	"bruteForceStatus": {
253	"numFailures": 0,
254	"disabled": False,
255	"lastIPFailure": "n/a",
256	"lastFailure": 0,
257	},
258	}
259
260
261	@dataclass(eq=True, frozen=True, kw_only=True)	2✔
262	class UserInfo(UnsavedUserInfo):	2✔
263	"""A tuple used to convey information about a user and their namespace."""
264
265	namespace: UserNamespace	2✔
266
267	def requires_update(self, current_user_info: UnsavedUserInfo) -> bool:	2✔
268	"""Returns true if the data self does not match the current_user_info."""
269	if self.id != current_user_info.id:	1✔
NEW 270	raise errors.ValidationError(message="Cannot check updates on two different users.")	×
271	self_as_unsaved = UnsavedUserInfo(	1✔
272	id=self.id,
273	first_name=self.first_name,
274	last_name=self.last_name,
275	email=self.email,
276	)
277	return self_as_unsaved != current_user_info	1✔
278
279
280	@dataclass(frozen=True, eq=True, kw_only=True)	2✔
281	class UserPatch:	2✔
282	"""Model for changes requested on a user."""
283
284	first_name: str \| None = None	2✔
285	last_name: str \| None = None	2✔
286	email: str \| None = None	2✔
287
288	@classmethod	2✔
289	def from_unsaved_user_info(cls, user: UnsavedUserInfo) -> UserPatch:	2✔
290	"""Create a user patch from a UnsavedUserInfo instance."""
291	return UserPatch(	1✔
292	first_name=user.first_name,
293	last_name=user.last_name,
294	email=user.email,
295	)
296
297
298	@dataclass	2✔
299	class DeletedUser:	2✔
300	"""A user that was deleted from the database."""
301
302	id: str	2✔
303
304
305	class UserInfoUpdate(NamedTuple):	2✔
306	"""Used to convey information about an update of a user or their namespace."""
307
308	old: UserInfo \| None	2✔
309	new: UserInfo	2✔
310
311
312	class PinnedProjects(BaseModel):	2✔
313	"""Pinned projects model."""
314
315	project_slugs: list[str] \| None = None	2✔
316
317	@classmethod	2✔
318	def from_dict(cls, data: dict) -> PinnedProjects:	2✔
319	"""Create model from a dict object."""
320	return cls(project_slugs=data.get("project_slugs"))	2✔
321
322
323	class UserPreferences(BaseModel):	2✔
324	"""User preferences model."""
325
326	user_id: str = Field(min_length=3)	2✔
327	pinned_projects: PinnedProjects	2✔
328	show_project_migration_banner: bool = True	2✔

SwissDataScienceCenter / renku-data-services / 19430749134

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