diff --git a/api/main.py b/api/main.py index 497b5705..6d23f6b7 100644 --- a/api/main.py +++ b/api/main.py @@ -28,6 +28,7 @@ Header, Query, Body, + Response, ) from fastapi.encoders import jsonable_encoder from fastapi.responses import ( @@ -70,6 +71,7 @@ UserUpdate, UserUpdateRequest, UserGroup, + UserGroupCreateRequest, InviteAcceptRequest, InviteUrlResponse, ) @@ -585,8 +587,11 @@ async def update_me(request: Request, user: UserUpdateRequest, if not group: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, - detail=f"User group does not exist with name: \ - {group_name}") + detail=( + "User group does not exist with name: " + f"{group_name}" + ), + ) groups.append(group) user_update = UserUpdate(**(user.model_dump( exclude={'groups', 'is_superuser'}, exclude_none=True))) @@ -624,8 +629,11 @@ async def update_user(user_id: str, request: Request, user: UserUpdateRequest, if not group: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, - detail=f"User group does not exist with name: \ - {group_name}") + detail=( + "User group does not exist with name: " + f"{group_name}" + ), + ) groups.append(group) user_update = UserUpdate(**(user.model_dump( exclude={'groups'}, exclude_none=True))) @@ -644,6 +652,77 @@ async def update_user(user_id: str, request: Request, user: UserUpdateRequest, return updated_user +@app.get("/user-groups", response_model=PageModel, tags=["user"]) +async def get_user_groups(request: Request, + current_user: User = Depends(get_current_superuser)): + """List user groups (admin-only).""" + metrics.add('http_requests_total', 1) + query_params = dict(request.query_params) + for pg_key in ['limit', 'offset']: + query_params.pop(pg_key, None) + paginated_resp = await db.find_by_attributes(UserGroup, query_params) + paginated_resp.items = serialize_paginated_data( + UserGroup, paginated_resp.items) + return paginated_resp + + +@app.get("/user-groups/{group_id}", response_model=UserGroup, tags=["user"], + response_model_by_alias=False) +async def get_user_group(group_id: str, + current_user: User = Depends(get_current_superuser)): + """Get a user group by id (admin-only).""" + metrics.add('http_requests_total', 1) + group = await db.find_by_id(UserGroup, group_id) + if not group: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"User group not found with id: {group_id}", + ) + return group + + +@app.post("/user-groups", response_model=UserGroup, tags=["user"], + response_model_by_alias=False) +async def create_user_group(group: UserGroupCreateRequest, + current_user: User = Depends( + get_current_superuser)): + """Create a user group (admin-only).""" + metrics.add('http_requests_total', 1) + existing = await db.find_one(UserGroup, name=group.name) + if existing: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"User group already exists with name: {group.name}", + ) + return await db.create(UserGroup(name=group.name)) + + +@app.delete("/user-groups/{group_id}", status_code=status.HTTP_204_NO_CONTENT, + tags=["user"]) +async def delete_user_group(group_id: str, + current_user: User = Depends( + get_current_superuser)): + """Delete a user group (admin-only).""" + metrics.add('http_requests_total', 1) + group = await db.find_by_id(UserGroup, group_id) + if not group: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"User group not found with id: {group_id}", + ) + assigned_count = await db.count(User, {"groups.name": group.name}) + if assigned_count: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=( + "User group is assigned to users and cannot be deleted. " + "Remove it from users first." + ), + ) + await db.delete_by_id(UserGroup, group_id) + return Response(status_code=status.HTTP_204_NO_CONTENT) + + def _get_node_runtime(node: Node) -> Optional[str]: """Best-effort runtime lookup from node data.""" data = getattr(node, 'data', None) diff --git a/api/models.py b/api/models.py index a9e02757..9e8ad8c8 100644 --- a/api/models.py +++ b/api/models.py @@ -115,6 +115,11 @@ def get_indexes(cls): ] +class UserGroupCreateRequest(BaseModel): + """Create user group request schema for API router""" + name: str = Field(description="User group name") + + class User(BeanieBaseUser, Document, # pylint: disable=too-many-ancestors DatabaseModel): """API User model""" diff --git a/doc/api-details.md b/doc/api-details.md index d721e366..5d79cd20 100644 --- a/doc/api-details.md +++ b/doc/api-details.md @@ -372,14 +372,22 @@ User groups are plain name strings stored in the `usergroup` collection. Group names must already exist before they can be assigned to users; otherwise the API returns `400`. -There is currently no REST endpoint for creating or deleting user groups. Use -MongoDB tooling to manage them. Example with `mongosh`: +User groups are plain name strings stored in the `usergroup` collection. You +can manage them via the API endpoints below or directly with MongoDB tooling. +Example with `mongosh`: ``` $ mongosh "mongodb://db:27017/kernelci" > db.usergroup.insertOne({name: "runtime:lava-collabora:node-editor"}) ``` +Admin-only user group management endpoints are available: + +- `GET /user-groups` (list; supports `name` filter) +- `GET /user-groups/` +- `POST /user-groups` with `{"name": "runtime:lava-collabora:node-editor"}` +- `DELETE /user-groups/` (fails with `409` if assigned to users) + Admin users can assign or remove groups via: - `POST /user/invite` with a `groups` list @@ -393,12 +401,133 @@ Example using the helper script: ``` $ ./scripts/usermanager.py list-users +$ ./scripts/usermanager.py list-groups +$ ./scripts/usermanager.py create-group runtime:lava-collabora:node-editor $ ./scripts/usermanager.py update-user 615f30020eb7c3c6616e5ac3 \ --data '{"groups": ["runtime:lava-collabora:node-editor"]}' ``` Users cannot update their own groups; admin access is required. +### Usermanager workflows (examples) + +These examples use `scripts/usermanager.py`. It reads `./usermanager.toml` or +`~/.config/kernelci/usermanager.toml` by default, and you can override with +`--api-url`/`--token` or `KCI_API_URL`/`KCI_API_TOKEN`. + +Common admin workflows: + +- List users and capture IDs: + +``` +$ ./scripts/usermanager.py list-users +$ ./scripts/usermanager.py get-user +``` + +- Invite a user (optionally add groups): + +``` +$ ./scripts/usermanager.py invite \ + --username alice \ + --email alice@example.org \ + --groups runtime:pull-labs-demo:node-editor \ + --return-token +``` + +- Accept an invite manually (useful for service accounts or testing): + +``` +$ ./scripts/usermanager.py accept-invite --token "" +``` + +- Login to get a bearer token: + +``` +$ ./scripts/usermanager.py login --username alice +``` + +- Deactivate or reactivate a user: + +``` +$ ./scripts/usermanager.py update-user --inactive +$ ./scripts/usermanager.py update-user --active +``` + +- Grant or revoke superuser: + +``` +$ ./scripts/usermanager.py update-user --superuser +$ ./scripts/usermanager.py update-user --no-superuser +``` + +- Mark a user verified or unverified (admin only): + +``` +$ ./scripts/usermanager.py update-user --verified +$ ./scripts/usermanager.py update-user --unverified +``` + +- Assign or remove groups: + +``` +$ ./scripts/usermanager.py update-user \ + --add-group runtime:pull-labs-demo:node-editor +$ ./scripts/usermanager.py update-user \ + --remove-group runtime:pull-labs-demo:node-editor +$ ./scripts/usermanager.py update-user \ + --set-groups runtime:pull-labs-demo:node-editor,team-a +``` + +- Set a password (admin only, useful for service accounts): + +``` +$ ./scripts/usermanager.py update-user --password "" +``` + +- Manage user groups: + +``` +$ ./scripts/usermanager.py list-groups +$ ./scripts/usermanager.py create-group runtime:pull-labs-demo:node-editor +$ ./scripts/usermanager.py delete-group runtime:pull-labs-demo:node-editor +``` + +- Delete a user: + +``` +$ ./scripts/usermanager.py delete-user +``` + +### Permissions and node update rules + +Node update permissions are determined by the user and the node being edited: + +- Superusers can update any node. +- The node owner can update their own nodes. +- Users with group `node:edit:any` can update any node. +- Users with a group listed in the node's `user_groups` can update that node. +- Users with `runtime::node-editor` or `runtime::node-admin` + can update nodes whose `data.runtime` matches ``. + +Example: allow updates only for runtime `pull-labs-demo`: + +``` +$ mongosh "mongodb://db:27017/kernelci" +> db.usergroup.insertOne({name: "runtime:pull-labs-demo:node-editor"}) +``` + +``` +$ ./scripts/usermanager.py update-user \ + --add-group runtime:pull-labs-demo:node-editor +``` + +To remove a user group definition entirely, delete it in MongoDB: + +``` +$ mongosh "mongodb://db:27017/kernelci" +> db.usergroup.deleteOne({name: "runtime:pull-labs-demo:node-editor"}) +``` + ### Delete user matching user ID (Admin only) diff --git a/scripts/usermanager.py b/scripts/usermanager.py index 2c467a33..653c3b7f 100755 --- a/scripts/usermanager.py +++ b/scripts/usermanager.py @@ -3,6 +3,7 @@ import getpass import json import os +import re import sys import urllib.error import urllib.parse @@ -124,7 +125,7 @@ def _apply_group_changes(current, add_groups, remove_groups): def _resolve_user_id(user_id, api_url, token): - if "@" not in user_id: + if _looks_like_object_id(user_id): return user_id status, body = _request_json("GET", f"{api_url}/users", token=token) if status >= 400: @@ -134,23 +135,102 @@ def _resolve_user_id(user_id, api_url, token): payload = json.loads(body) if body else [] except json.JSONDecodeError as exc: raise SystemExit("Failed to parse users response") from exc - if not isinstance(payload, list): - raise SystemExit("Unexpected users response") + items = _parse_paginated_items(payload) + matches = [] + for user in items: + if not isinstance(user, dict): + continue + if user.get("email") == user_id or user.get("username") == user_id: + matches.append(user) + if not matches: + raise SystemExit(f"No user found with email/username: {user_id}") + if len(matches) > 1: + raise SystemExit(f"Multiple users found with email/username: {user_id}") + resolved_id = matches[0].get("id") + if not resolved_id: + raise SystemExit(f"User with {user_id} has no id") + return resolved_id + + +def _parse_paginated_items(payload): + if isinstance(payload, dict) and "items" in payload: + return payload.get("items") or [] + if isinstance(payload, list): + return payload + return [] + + +def _looks_like_object_id(value): + return bool(re.fullmatch(r"[0-9a-fA-F]{24}", value)) + + +def _resolve_group_id(group_id, api_url, token): + if _looks_like_object_id(group_id): + return group_id + query = urllib.parse.urlencode({"name": group_id}) + status, body = _request_json("GET", f"{api_url}/user-groups?{query}", token=token) + if status >= 400: + _print_response(status, body) + raise SystemExit(1) + try: + payload = json.loads(body) if body else {} + except json.JSONDecodeError as exc: + raise SystemExit("Failed to parse user-groups response") from exc + items = _parse_paginated_items(payload) matches = [ - user - for user in payload - if isinstance(user, dict) and user.get("email") == user_id + group + for group in items + if isinstance(group, dict) and group.get("name") == group_id ] if not matches: - raise SystemExit(f"No user found with email: {user_id}") + raise SystemExit(f"No group found with name: {group_id}") if len(matches) > 1: - raise SystemExit(f"Multiple users found with email: {user_id}") + raise SystemExit(f"Multiple groups found with name: {group_id}") resolved_id = matches[0].get("id") if not resolved_id: - raise SystemExit(f"User with email {user_id} has no id") + raise SystemExit(f"Group {group_id} has no id") return resolved_id +def _resolve_group_name(group_name, api_url, token): + if not _looks_like_object_id(group_name): + return group_name + status, body = _request_json( + "GET", f"{api_url}/user-groups/{group_name}", token=token + ) + if status >= 400: + _print_response(status, body) + raise SystemExit(1) + try: + payload = json.loads(body) if body else {} + except json.JSONDecodeError as exc: + raise SystemExit("Failed to parse user-group response") from exc + resolved_name = payload.get("name") + if not resolved_name: + raise SystemExit(f"Group {group_name} has no name") + return resolved_name + + +def _resolve_group_names(group_names, api_url, token): + return _dedupe([_resolve_group_name(name, api_url, token) for name in group_names]) + + +def _update_user_groups(resolved_id, add_groups, remove_groups, api_url, token): + status, body = _request_json("GET", f"{api_url}/user/{resolved_id}", token=token) + if status >= 400: + _print_response(status, body) + raise SystemExit(1) + try: + payload = json.loads(body) if body else {} + except json.JSONDecodeError as exc: + raise SystemExit("Failed to parse user response") from exc + current_groups = _extract_group_names(payload) + data = { + "groups": _apply_group_changes(current_groups, add_groups, remove_groups), + } + return _request_json("PATCH", f"{api_url}/user/{resolved_id}", data, token=token) + + def _request_json(method, url, data=None, token=None, form=False): headers = {"accept": "application/json"} body = None @@ -195,9 +275,36 @@ def _require_token(token, args): def main(): + command_help = [ + ("accept-invite", "Accept an invite"), + ("assign-group", "Assign group(s) to a user"), + ("config-example", "Print a sample usermanager.toml"), + ("create-group", "Create user group"), + ("deassign-group", "Remove group(s) from a user"), + ("delete-group", "Delete user group"), + ("delete-user", "Delete user by id/email/username"), + ("generate-api-token", "Print just the access token for a user"), + ("get-group", "Get user group by id or name"), + ("get-user", "Get user by id/email/username"), + ("invite", "Invite a new user"), + ("invite-url", "Preview invite URL base"), + ("list-groups", "List user groups"), + ("list-users", "List users"), + ("login", "Get a full auth token response"), + ("update-user", "Patch user by id/email/username"), + ("whoami", "Show current user"), + ] + command_list = "\n".join(f" {name:<18} {desc}" for name, desc in command_help) default_paths = "\n".join(f" - {path}" for path in DEFAULT_CONFIG_PATHS) parser = argparse.ArgumentParser( description="KernelCI API user management helper", + usage=( + "usermanager.py [-h] [--config CONFIG] [--api-url API_URL] " + "[--token TOKEN] [--instance INSTANCE] [--token-label TOKEN_LABEL]\n" + " []\n\n" + "Commands:\n" + f"{command_list}" + ), epilog=( "Examples:\n" " ./scripts/usermanager.py invite --username alice --email " @@ -206,10 +313,12 @@ def main(): " ./scripts/usermanager.py login --username alice\n" " ./scripts/usermanager.py whoami\n" " ./scripts/usermanager.py list-users --instance staging\n" - " ./scripts/usermanager.py print-config-example\n" + " ./scripts/usermanager.py config-example\n" "\n" "Default config lookup (first match wins):\n" f"{default_paths}\n" + "\n" + "Run ' -h' for command-specific help.\n" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) @@ -228,6 +337,55 @@ def main(): subparsers = parser.add_subparsers(dest="command", required=True) + accept = subparsers.add_parser("accept-invite", help="Accept an invite") + accept.add_argument("--token") + accept.add_argument("--password") + + assign_group = subparsers.add_parser( + "assign-group", help="Assign group(s) to a user" + ) + assign_group.add_argument("user_id") + assign_group.add_argument( + "--group", + action="append", + default=[], + help="Group name or id; can be used multiple times or with commas", + ) + + subparsers.add_parser("config-example", help="Print a sample usermanager.toml") + + create_group = subparsers.add_parser("create-group", help="Create user group") + create_group.add_argument("name") + + deassign_group = subparsers.add_parser( + "deassign-group", help="Remove group(s) from a user" + ) + deassign_group.add_argument("user_id") + deassign_group.add_argument( + "--group", + action="append", + default=[], + help="Group name or id; can be used multiple times or with commas", + ) + + delete_group = subparsers.add_parser("delete-group", help="Delete user group") + delete_group.add_argument("group_id") + + delete_user = subparsers.add_parser("delete-user", help="Delete user by id") + delete_user.add_argument("user_id") + + generate_token = subparsers.add_parser( + "generate-api-token", help="Print just the access token for a user" + ) + generate_token.add_argument("--username", required=True) + generate_token.add_argument("--password") + + get_group = subparsers.add_parser("get-group", help="Get user group by id or name") + get_group.add_argument("group_id") + + get_user = subparsers.add_parser("get-user", help="Get user by id") + get_user.add_argument("user_id") + invite = subparsers.add_parser("invite", help="Invite a new user") invite.add_argument("--username", required=True) invite.add_argument("--email", required=True) @@ -240,21 +398,14 @@ def main(): invite_url = subparsers.add_parser("invite-url", help="Preview invite URL base") - accept = subparsers.add_parser("accept-invite", help="Accept an invite") - accept.add_argument("--token") - accept.add_argument("--password") + list_groups = subparsers.add_parser("list-groups", help="List user groups") + + list_users = subparsers.add_parser("list-users", help="List users") login = subparsers.add_parser("login", help="Get an auth token") login.add_argument("--username", required=True) login.add_argument("--password") - whoami = subparsers.add_parser("whoami", help="Show current user") - - list_users = subparsers.add_parser("list-users", help="List users") - - get_user = subparsers.add_parser("get-user", help="Get user by id") - get_user.add_argument("user_id") - update_user = subparsers.add_parser("update-user", help="Patch user by id") update_user.add_argument("user_id") update_user.add_argument("--data", help="JSON object with fields to update") @@ -306,16 +457,11 @@ def main(): help="Remove group(s); can be used multiple times or with commas", ) - delete_user = subparsers.add_parser("delete-user", help="Delete user by id") - delete_user.add_argument("user_id") - - subparsers.add_parser( - "print-config-example", help="Print a sample usermanager.toml" - ) + whoami = subparsers.add_parser("whoami", help="Show current user") args = parser.parse_args() - if args.command == "print-config-example": + if args.command == "config-example": print( 'default_instance = "local"\n\n' "[instances.local]\n" @@ -362,6 +508,12 @@ def main(): "get-user", "update-user", "delete-user", + "assign-group", + "deassign-group", + "list-groups", + "get-group", + "create-group", + "delete-group", }: token = _require_token(token, args) @@ -407,6 +559,29 @@ def main(): payload, form=True, ) + elif args.command == "generate-api-token": + password = _prompt_if_missing( + args.password, + "Password: ", + secret=True, + ) + payload = {"username": args.username, "password": password} + status, body = _request_json( + "POST", + f"{api_url}/user/login", + payload, + form=True, + ) + if status < 400: + try: + payload = json.loads(body) if body else {} + except json.JSONDecodeError as exc: + raise SystemExit("Failed to parse login response") from exc + token = payload.get("access_token") + if not token: + raise SystemExit("Login response missing access_token") + print(token) + return elif args.command == "whoami": status, body = _request_json("GET", f"{api_url}/whoami", token=token) elif args.command == "list-users": @@ -442,6 +617,12 @@ def main(): set_groups = _parse_group_list(args.set_groups) add_groups = _parse_group_list(args.add_group) remove_groups = _parse_group_list(args.remove_group) + if set_groups: + set_groups = _resolve_group_names(set_groups, api_url, token) + if add_groups: + add_groups = _resolve_group_names(add_groups, api_url, token) + if remove_groups: + remove_groups = _resolve_group_names(remove_groups, api_url, token) if set_groups or add_groups or remove_groups: if set_groups: current_groups = set_groups @@ -471,6 +652,39 @@ def main(): status, body = _request_json( "DELETE", f"{api_url}/user/{resolved_id}", token=token ) + elif args.command == "assign-group": + resolved_id = _resolve_user_id(args.user_id, api_url, token) + add_groups = _parse_group_list(args.group) + if not add_groups: + raise SystemExit("No groups specified. Use --group.") + add_groups = _resolve_group_names(add_groups, api_url, token) + status, body = _update_user_groups(resolved_id, add_groups, [], api_url, token) + elif args.command == "deassign-group": + resolved_id = _resolve_user_id(args.user_id, api_url, token) + remove_groups = _parse_group_list(args.group) + if not remove_groups: + raise SystemExit("No groups specified. Use --group.") + remove_groups = _resolve_group_names(remove_groups, api_url, token) + status, body = _update_user_groups( + resolved_id, [], remove_groups, api_url, token + ) + elif args.command == "list-groups": + status, body = _request_json("GET", f"{api_url}/user-groups", token=token) + elif args.command == "get-group": + resolved_id = _resolve_group_id(args.group_id, api_url, token) + status, body = _request_json( + "GET", f"{api_url}/user-groups/{resolved_id}", token=token + ) + elif args.command == "create-group": + payload = {"name": args.name} + status, body = _request_json( + "POST", f"{api_url}/user-groups", payload, token=token + ) + elif args.command == "delete-group": + resolved_id = _resolve_group_id(args.group_id, api_url, token) + status, body = _request_json( + "DELETE", f"{api_url}/user-groups/{resolved_id}", token=token + ) else: raise SystemExit("Unknown command") diff --git a/tests/unit_tests/conftest.py b/tests/unit_tests/conftest.py index d0f7f6ab..47fd4448 100644 --- a/tests/unit_tests/conftest.py +++ b/tests/unit_tests/conftest.py @@ -153,6 +153,15 @@ def mock_db_find_by_id(mocker): return async_mock +@pytest.fixture +def mock_db_delete_by_id(mocker): + """Mocks async call to Database class method used to delete an object""" + async_mock = AsyncMock() + mocker.patch('api.db.Database.delete_by_id', + side_effect=async_mock) + return async_mock + + @pytest.fixture def mock_db_find_one(mocker): """Mocks async call to database method used to find one object""" diff --git a/tests/unit_tests/test_user_group_handler.py b/tests/unit_tests/test_user_group_handler.py index c262736d..5aa8db5e 100644 --- a/tests/unit_tests/test_user_group_handler.py +++ b/tests/unit_tests/test_user_group_handler.py @@ -1,108 +1,86 @@ # SPDX-License-Identifier: LGPL-2.1-or-later # -# Copyright (C) 2023 Collabora Limited -# Author: Jeny Sadadia - -# pylint: disable=unused-argument +# Copyright (C) 2025 Collabora Limited """Unit test functions for KernelCI API user group handler""" import json -from tests.unit_tests.conftest import ( - ADMIN_BEARER_TOKEN, - BEARER_TOKEN, -) -from api.models import UserGroup, PageModel +from api.models import PageModel, UserGroup +from tests.unit_tests.conftest import ADMIN_BEARER_TOKEN -def test_create_user_group(mock_db_create, mock_publish_cloudevent, - test_client): - """ - Test Case : Test KernelCI API /group endpoint to create user group - when requested with admin user's bearer token - Expected Result : - HTTP Response Code 200 OK - JSON with 'id' and 'name' keys - """ - mock_db_create.return_value = UserGroup( - id='61bda8f2eb1a63d2b7152422', - name='kernelci') +def test_list_user_groups(mock_db_find_by_attributes, test_client): + """GET /user-groups returns a paginated list of user groups.""" + group_1 = {"id": "65265305c74695807499037f", "name": "team-a"} + group_2 = {"id": "65265305c746958074990370", "name": "team-b"} + mock_db_find_by_attributes.return_value = PageModel( + items=[group_1, group_2], + total=2, + limit=50, + offset=0, + ) - response = test_client.post( - "group", + response = test_client.get( + "user-groups", headers={ "Accept": "application/json", - "Authorization": ADMIN_BEARER_TOKEN + "Authorization": ADMIN_BEARER_TOKEN, }, - data=json.dumps({"name": "kernelci"}) ) - print(response.json()) assert response.status_code == 200 - assert ('id', 'name') == tuple(response.json().keys()) + assert response.json()["total"] == 2 -def test_create_group_endpoint_negative(mock_publish_cloudevent, - test_client): - """ - Test Case : Test KernelCI API /group endpoint when requested - with regular user's bearer token - Expected Result : - HTTP Response Code 403 Forbidden - JSON with 'detail' key denoting 'Forbidden' error - """ +def test_create_user_group(mock_db_find_one, mock_db_create, test_client): + """POST /user-groups creates a new user group.""" + mock_db_find_one.return_value = None + mock_db_create.return_value = UserGroup(name="runtime:pull-labs-demo:node-editor") + response = test_client.post( - "group", + "user-groups", headers={ "Accept": "application/json", - "Authorization": BEARER_TOKEN + "Authorization": ADMIN_BEARER_TOKEN, }, - data=json.dumps({"name": "kernelci"}) + data=json.dumps({"name": "runtime:pull-labs-demo:node-editor"}), ) - print(response.json()) - assert response.status_code == 403 - assert response.json() == {'detail': 'Forbidden'} + assert response.status_code == 200 + assert response.json()["name"] == "runtime:pull-labs-demo:node-editor" -def test_get_groups(mock_db_find_by_attributes, - test_client): - """ - Test Case : Test KernelCI API GET /groups endpoint - Expected Result : - HTTP Response Code 200 OK - List of all the user group objects - """ - user_group_1 = { - "id": "61bda8f2eb1a63d2b7152421", - "name": "admin"} - user_group_2 = { - "id": "61bda8f2eb1a63d2b7152422", - "name": "kernelci"} - mock_db_find_by_attributes.return_value = PageModel( - items=[user_group_1, user_group_2], - total=2, - limit=50, - offset=0 +def test_delete_user_group(mock_db_find_by_id, mock_db_count, + mock_db_delete_by_id, test_client): + """DELETE /user-groups/{id} removes an unused user group.""" + mock_db_find_by_id.return_value = UserGroup(name="team-a") + mock_db_count.return_value = 0 + + response = test_client.delete( + "user-groups/65265305c74695807499037f", + headers={ + "Accept": "application/json", + "Authorization": ADMIN_BEARER_TOKEN, + }, + ) + assert response.status_code == 204 + mock_db_delete_by_id.assert_called_once_with( + UserGroup, + "65265305c74695807499037f", ) - response = test_client.get("groups") - print("response.json()", response.json()) - assert response.status_code == 200 - assert ('items', 'total', 'limit', - 'offset') == tuple(response.json().keys()) -def test_get_group_by_id(mock_db_find_by_id, - test_client): - """ - Test Case : Test KernelCI API GET /group/{group_id} endpoint - Expected Result : - HTTP Response Code 200 OK - JSON with UserGroup object - """ - mock_db_find_by_id.return_value = UserGroup(id='61bda8f2eb1a63d2b7152422', - name='kernelci') +def test_delete_user_group_when_assigned(mock_db_find_by_id, mock_db_count, + test_client): + """DELETE /user-groups/{id} rejects when group is assigned to users.""" + mock_db_find_by_id.return_value = UserGroup(name="team-a") + mock_db_count.return_value = 2 - response = test_client.get("group/61bda8f2eb1a63d2b7152422") - print("response.json()", response.json()) - assert response.status_code == 200 - assert response.json().keys() == {'id', 'name'} + response = test_client.delete( + "user-groups/65265305c74695807499037f", + headers={ + "Accept": "application/json", + "Authorization": ADMIN_BEARER_TOKEN, + }, + ) + assert response.status_code == 409 + assert response.json()["detail"].startswith("User group is assigned")