From 2c30f7875f7457877e92fff692ee247b29855064 Mon Sep 17 00:00:00 2001 From: ivis-kosaka Date: Wed, 11 Mar 2026 13:18:07 +0000 Subject: [PATCH] test: add unit test for cov --- src/server/entities/bulk.py | 32 --- src/server/services/history.py | 1 + src/server/services/users.py | 44 +++-- src/server/services/utils/transformers.py | 16 +- tests/unit/services/test_history.py | 36 +++- tests/unit/services/test_history_table.py | 27 +++ tests/unit/services/test_transformers.py | 229 ++++++++++++++++++++++ tests/unit/services/test_users.py | 91 ++++++++- 8 files changed, 419 insertions(+), 57 deletions(-) diff --git a/src/server/entities/bulk.py b/src/server/entities/bulk.py index acf10cb..86788c0 100644 --- a/src/server/entities/bulk.py +++ b/src/server/entities/bulk.py @@ -15,16 +15,6 @@ from server.entities.user_detail import UserDetail -CSV_TO_FIELDS = { - "user_name": "user_name", - "groups[].id": "groups_ids", - "groups[].name": "groups_names", - "edu_person_principal_names[]": "eppns", - "emails[]": "emails", - "preferred_language": "preferred_language", -} - - class RepositoryMember(BaseModel): """Model for members of a repository.""" @@ -174,25 +164,3 @@ class FileContent(t.TypedDict): users: dict[str, str] """Dictionary of users.""" - - -class FileUserDict(t.TypedDict, total=False): - """Model for user data in file as dictionary.""" - - user_name: list[str] - """List of usernames.""" - - groups_ids: list[str] - """List of group IDs.""" - - groups_names: list[str] - """List of group names.""" - - eppns: list[str] - """List of eduPersonPrincipalNames.""" - - emails: list[str] - """List of e-mails.""" - - preferred_language: list[str] - """List of preferred languages.""" diff --git a/src/server/services/history.py b/src/server/services/history.py index 7401128..30071b2 100644 --- a/src/server/services/history.py +++ b/src/server/services/history.py @@ -272,6 +272,7 @@ def get_filter_items( stmt = stmt.limit(page_size).offset(offset) try: results = db.session.execute(stmt).all() + current_app.logger.info("results: %s", results) except SQLAlchemyError as exc: current_app.logger.error(str(exc)) raise DatabaseError(E.FAILED_GET_HISTORY_RECORDS % {"table": tab}) from exc diff --git a/src/server/services/users.py b/src/server/services/users.py index 656d789..dce03f4 100644 --- a/src/server/services/users.py +++ b/src/server/services/users.py @@ -776,9 +776,9 @@ def make_export_file( user_list, delimiter, file_path, permitted_repository_ids ) file_content = { - "repositories": list(file_repositories), - "groups": list(file_groups), - "users": list(file_users), + "repositories": file_repositories, + "groups": file_groups, + "users": file_users, } history_table.create_download_history( file_id, str(file_path), file_content, operator_id, operator_name @@ -792,7 +792,7 @@ def _wite_user( delimiter: str, file_path: Path, permitted_repository_ids: set[str], -) -> tuple[set[dict[str, str]], set[dict[str, str]], set[dict[str, str]]]: +) -> tuple[list[dict[str, str]], list[dict[str, str]], list[dict[str, str]]]: """Write user details to file. Args: @@ -802,37 +802,44 @@ def _wite_user( permitted_repository_ids (list[str]): A list of permitted repository IDs. Returns: - tuple[set[dict[str, str]], set[dict[str, str]], set[dict[str, str]]]: - A tuple containing sets of file repositories, file groups, and file users. + tuple[list[dict[str, str]], list[dict[str, str]], list[dict[str, str]]]: + A tuple containing lists of file repositories, file groups, and file users. Raises: InvalidExportError: If the user cannot be exported due to insufficient permissions. """ - file_repositories = set[dict[str, str]]() - file_groups = set[dict[str, str]]() - file_users = set[dict[str, str]]() + file_repositories_dict: dict[str, dict[str, str]] = {} + file_groups_dict: dict[str, dict[str, str]] = {} + file_users_dict: dict[str, dict[str, str]] = {} for map_user in user_list: roles, groups = detect_affiliations([g.value for g in map_user.groups or []]) if not is_current_user_system_admin() and any( - role.role == USER_ROLES.SYSTEM_ADMIN for role in roles + role_group.role == USER_ROLES.SYSTEM_ADMIN for role_group in roles ): - error = E.USER_CANNOT_EXPORT_SYSTEM_ADMIN - raise InvalidExportError(error) + raise InvalidExportError(E.USER_CANNOT_EXPORT_SYSTEM_ADMIN) if not is_current_user_system_admin() and not any( group.repository_id in permitted_repository_ids for group in groups ): - error = E.USER_FORBIDDEN_EXPORT - raise InvalidExportError(error) + raise InvalidExportError(E.USER_FORBIDDEN_EXPORT) - file_users.add({"id": map_user.id or "", "user_name": map_user.user_name or ""}) + file_users_dict[map_user.id or ""] = { + "id": map_user.id or "", + "user_name": map_user.user_name or "", + } group_ids = [] for group in groups: if group.repository_id not in permitted_repository_ids: continue - file_groups.add({"id": group.group_id or "", "display_name": ""}) - file_repositories.add({"id": group.repository_id or "", "display_name": ""}) + file_groups_dict[group.group_id or ""] = { + "id": group.group_id or "", + "display_name": "", + } + file_repositories_dict[group.repository_id or ""] = { + "id": group.repository_id or "", + "service_name": "", + } group_ids.append(group.group_id or "") roles_list = [ r.role.value for r in roles if r.repository_id in permitted_repository_ids @@ -859,4 +866,7 @@ def _wite_user( delimiter.join(row) + "\n", encoding="utf-8", ) + file_repositories = list(file_repositories_dict.values()) + file_groups = list(file_groups_dict.values()) + file_users = list(file_users_dict.values()) return file_repositories, file_groups, file_users diff --git a/src/server/services/utils/transformers.py b/src/server/services/utils/transformers.py index 7fc3951..65ca82f 100644 --- a/src/server/services/utils/transformers.py +++ b/src/server/services/utils/transformers.py @@ -8,6 +8,8 @@ import typing as t +from flask import current_app + from server.config import config from server.const import ( GROUP_DEFAULT_MEMBER_LIST_VISIBILITY, @@ -357,7 +359,7 @@ def validate_group_to_map_group( ) -> MapGroup: ... -def validate_group_to_map_group( # noqa: C901, PLR0912 +def validate_group_to_map_group( # noqa: C901 group: GroupDetail, *, mode: t.Literal["create", "update"] ) -> tuple[MapGroup, str] | MapGroup: """Validate the GroupDetail instance and convert it to a MapGroup instance. @@ -383,6 +385,7 @@ def validate_group_to_map_group( # noqa: C901, PLR0912 raise InvalidFormError(error) detected = detect_affiliation(group.id) + current_app.logger.error("Detected affiliation: %s", detected) if not detected: # out of this service's scope. error = E.GROUP_INVALID_ID_PATTERN @@ -414,11 +417,10 @@ def validate_group_to_map_group( # noqa: C901, PLR0912 error = E.GROUP_REQUIRES_USER_DEFINED_ID raise InvalidFormError(error) - if user_defined_id: - max_id_length = config.GROUPS.max_id_length - len(repository_id) - if len(user_defined_id) > max_id_length: - error = E.GROUP_TOO_LONG_ID % {"rid": repository_id, "max": max_id_length} - raise InvalidFormError(error) + max_id_length = config.GROUPS.max_id_length - len(repository_id) + if len(user_defined_id) > max_id_length: + error = E.GROUP_TOO_LONG_ID % {"rid": repository_id, "max": max_id_length} + raise InvalidFormError(error) id_pattern = config.GROUPS.id_patterns.user_defined group.id = id_pattern.format( @@ -696,6 +698,7 @@ def validate_user_groups(user: UserDetail, permitted: set[str]) -> list[str]: return [] specified = [group.id for group in user.groups if group.id] + current_app.logger.error("Specified group IDs: %s", specified) _, detected = detect_affiliations(specified) group_query = make_criteria_object( "groups", i=[group.group_id for group in detected], l=-1 @@ -704,6 +707,7 @@ def validate_user_groups(user: UserDetail, permitted: set[str]) -> list[str]: from server.services import groups # noqa: PLC0415 existed = {g.id for g in groups.search(criteria=group_query).resources} + current_app.logger.error("Existed group IDs: %s", existed) if non_existent := set(specified) - existed: error = E.USER_REQUIRES_EXISTING_GROUP % {"id": ", ".join(non_existent)} diff --git a/tests/unit/services/test_history.py b/tests/unit/services/test_history.py index b304c07..0c589e8 100644 --- a/tests/unit/services/test_history.py +++ b/tests/unit/services/test_history.py @@ -8,11 +8,12 @@ from sqlalchemy.dialects import postgresql from sqlalchemy.exc import SQLAlchemyError +from server.api.schemas import OperatorQuery from server.db.history import DownloadHistory, Files, UploadHistory, _FileContent, _ResultData from server.entities.history_detail import DownloadHistoryData, HistoryQuery, UploadHistoryData from server.entities.search_request import SearchResult from server.entities.summaries import UserSummary -from server.exc import DatabaseError, RecordNotFound +from server.exc import DatabaseError, InvalidQueryError, RecordNotFound from server.messages import E from server.services import history @@ -330,6 +331,39 @@ def test__build_filters_for_history( repoadmin_filter.assert_not_called() if is_system_admin else repoadmin_filter.assert_called_once() +def test_get_filter_items(app, mocker: MockerFixture): + expected = SearchResult[UserSummary]( + total=0, + page_size=20, + offset=0, + resources=[ + UserSummary( + id="operator_1", user_name="Operator 1", role=None, emails=None, eppns=None, last_modified=None + ), + UserSummary( + id="operator_2", user_name="Operator 2", role=None, emails=None, eppns=None, last_modified=None + ), + ], + ) + mock_data = [("operator_1", "Operator 1"), ("operator_2", "Operator 2")] + mocker.patch("server.db.db.session.execute", return_value=mocker.MagicMock(all=lambda: mock_data)) + result = history.get_filter_items("download", "o", OperatorQuery()) + assert result == expected + + +def test_get_filter_items_with_exception(app, mocker: MockerFixture): + mocker.patch("server.db.db.session.execute", side_effect=SQLAlchemyError()) + with pytest.raises(DatabaseError) as exc: + history.get_filter_items("download", "o", OperatorQuery()) + assert str(exc.value) == str(E.FAILED_GET_HISTORY_RECORDS % {"table": "download"}) + + +def test_get_filter_items_invalid_query(app, mocker: MockerFixture): + with pytest.raises(InvalidQueryError) as exc: + history.get_filter_items("download", "O", OperatorQuery()) + assert str(exc.value) == str(InvalidQueryError(E.FAILED_GET_FILTER_ITEMS % {"key": "O"})) + + def test_update_public_status_not_found(app, mocker: MockerFixture): db = mocker.MagicMock() mocker.patch("server.services.history.db", db) diff --git a/tests/unit/services/test_history_table.py b/tests/unit/services/test_history_table.py index fd2f7eb..d47a6b2 100644 --- a/tests/unit/services/test_history_table.py +++ b/tests/unit/services/test_history_table.py @@ -276,3 +276,30 @@ def test_create_file_without_id(app, mocker: MockerFixture): assert result.id == file_id assert result.file_path == file_path assert result.file_content == file_content + + +def test_create_download_history(app, mocker: MockerFixture): + file_id = uuid7() + file_path = "test/path" + file_content = {} + operator_id = "test_user_1" + operator_name = "Test user" + mocker.patch("server.services.history_table.create_file", return_value=None) + mock_add = mocker.patch("server.db.db.session.add") + result = history_table.create_download_history(file_id, file_path, file_content, operator_id, operator_name) + assert result.file_id == file_id + assert result.operator_id == operator_id + assert result.operator_name == operator_name + mock_add.assert_called_once() + + +def test_create_download_history_with_exception(app, mocker: MockerFixture): + file_id = uuid7() + file_path = "test/path" + file_content = {} + operator_id = "test_user_1" + operator_name = "Test user" + mocker.patch("server.services.history_table.create_file", side_effect=SQLAlchemyError) + with pytest.raises(DatabaseError) as exc: + history_table.create_download_history(file_id, file_path, file_content, operator_id, operator_name) + assert str(exc.value) == str(E.FAILED_CREATE_DOWNLOAD_HISTORY_RECORD % {"file_id": file_id}) diff --git a/tests/unit/services/test_transformers.py b/tests/unit/services/test_transformers.py index bfd0892..964c0c4 100644 --- a/tests/unit/services/test_transformers.py +++ b/tests/unit/services/test_transformers.py @@ -542,6 +542,36 @@ def test_validate_group_to_map_group(app, mocker: MockerFixture, group, mode, ex mock_make_map_group.assert_called_once_with(expectedarg) +def test_validate_group_to_map_group_non_repository_id(app, mocker: MockerFixture): + group = GroupDetail( + id="jc_repo1_gr_test_group_test", + display_name="Test Group", + repository=GroupRepository(id="repo1"), + user_defined_id="test_group", + public=True, + member_list_visibility="Hidden", + type="group", + ) + expectedarg = GroupDetail( + id="jc_repo1_gr_test_group_test", + display_name="Test Group", + repository=GroupRepository(id="repo1"), + user_defined_id="test_group", + public=True, + member_list_visibility="Hidden", + type="group", + ) + mocker.patch("server.services.repositories.get_by_id", return_value=RepositoryDetail(id="repo1")) + mocker.patch("server.services.utils.transformers.get_permitted_repository_ids", return_value={"repo1"}) + mock_make_map_group = mocker.patch("server.services.utils.transformers.make_map_group", return_value=MapGroup()) + mocker.patch( + "server.services.utils.transformers.detect_affiliation", + return_value=_Group(repository_id="", group_id="jc__gr_test_1", user_defined_id="test_1"), + ) + assert transformers.validate_group_to_map_group(group=group, mode="update") == MapGroup() + mock_make_map_group.assert_called_once_with(expectedarg) + + @pytest.mark.parametrize( ("group", "mode", "repository_exist", "expected"), [ @@ -583,6 +613,29 @@ def test_validate_group_to_map_group(app, mocker: MockerFixture, group, mode, ex True, E.GROUP_TOO_LONG_ID % {"rid": "repo1", "max": 50 - len("jc_") - len("_gr_") - len("repo1")}, ), + ( + GroupDetail( + id="invalid_id_pattern", + display_name="Test Group1", + repository=None, + user_defined_id="id", + type="group", + ), + "update", + True, + E.GROUP_INVALID_ID_PATTERN, + ), + ( + GroupDetail( + display_name="Test Group", + repository=GroupRepository(id="repo2"), + user_defined_id="group_id", + type="group", + ), + "create", + True, + E.GROUP_FORBIDDEN_REPOSITORY % {"rid": "repo2"}, + ), ], ) def test_validate_group_to_map_group_error(app, mocker: MockerFixture, group, mode, repository_exist, expected): @@ -941,6 +994,182 @@ def test_validate_user_to_map_user_update( return_fnc.assert_called_once_with(expectedarg) +def test_validate_user_to_map_user_not_promote(app, mocker: MockerFixture): + user_detail = UserDetail( + id="user1", + user_name="Test User", + eppns=["test_eppn"], + emails=["test@email.com"], + groups=[], + is_system_admin=True, + ) + mocker.patch( + "server.services.repositories.get_by_id", + return_value=RepositoryDetail(id="repo1"), + ) + mocker.patch("server.services.utils.transformers.get_permitted_repository_ids", return_value={"repo1"}) + mocker.patch( + "server.services.utils.transformers.validate_user_roles", + return_value=[], + ) + mocker.patch( + "server.services.utils.transformers.is_super", + return_value=False, + ) + mocker.patch("server.services.utils.transformers.validate_user_groups", return_value=["jc_repo1_gr_test_group"]) + with pytest.raises(InvalidFormError) as exc: + transformers.validate_user_to_map_user(user_detail, mode="update") + assert str(exc.value) == str(E.USER_NO_PROMOTE_SYSTEM_ADMIN) + + +def test_validate_user_roles(app, mocker: MockerFixture): + user_detail = UserDetail( + user_name="Test User", + repository_roles=[ + RepositoryRole(id="repo1", user_role=USER_ROLES.REPOSITORY_ADMIN), + RepositoryRole(id="repo2", user_role=None), + RepositoryRole(id="", user_role=USER_ROLES.REPOSITORY_ADMIN), + ], + ) + permitted = {"repo1", "repo2"} + mocker.patch( + "server.services.repositories.search", + return_value=SearchResult[RepositorySummary]( + total=2, page_size=20, offset=1, resources=[RepositorySummary(id="repo1"), RepositorySummary(id="repo2")] + ), + ) + assert transformers.validate_user_roles(user_detail, permitted) == ["jc_repo1_ro_radm_test"] + + +def test_validate_user_roles_no_existed(app, mocker: MockerFixture): + user_detail = UserDetail( + user_name="Test User", + repository_roles=[ + RepositoryRole(id="repo1", user_role=USER_ROLES.REPOSITORY_ADMIN), + ], + ) + permitted = {"repo1", "repo2"} + mocker.patch( + "server.services.repositories.search", + return_value=SearchResult[RepositorySummary](total=0, page_size=20, offset=1, resources=[]), + ) + with pytest.raises(InvalidFormError) as exc: + transformers.validate_user_roles(user_detail, permitted) + assert str(exc.value) == str(E.USER_REQUIRES_EXISTING_REPOSITORY % {"id": "repo1"}) + + +def test_validate_user_roles_forbidden(app, mocker: MockerFixture): + user_detail = UserDetail( + user_name="Test User", + repository_roles=[ + RepositoryRole(id="repo1", user_role=USER_ROLES.REPOSITORY_ADMIN), + RepositoryRole(id="repo2", user_role=None), + ], + ) + permitted = set() + mocker.patch( + "server.services.repositories.search", + return_value=SearchResult[RepositorySummary]( + total=2, + page_size=20, + offset=1, + resources=[RepositorySummary(id="repo1"), RepositorySummary(id="repo2")], + ), + ) + with pytest.raises(InvalidFormError) as exc: + transformers.validate_user_roles(user_detail, permitted) + assert str(exc.value) == str(E.USER_FORBIDDEN_REPOSITORY % {"id": "repo1"}) + + +def test_validate_user_groups(app, mocker: MockerFixture): + user_detail = UserDetail( + user_name="Test User", + groups=[GroupSummary(id="jc_repo1_gr_test_group")], + ) + permitted = {"repo1"} + mocker.patch( + "server.services.utils.transformers.detect_affiliations", + return_value=( + None, + [ + _Group(repository_id="repo1", group_id="jc_repo1_gr_test_group", user_defined_id="test_group"), + ], + ), + ) + mocker.patch( + "server.services.groups.search", + return_value=SearchResult[GroupSummary]( + total=1, page_size=20, offset=1, resources=[GroupSummary(id="jc_repo1_gr_test_group")] + ), + ) + assert transformers.validate_user_groups(user_detail, permitted) == ["jc_repo1_gr_test_group"] + + +def test_validate_user_groups_non_existent(app, mocker: MockerFixture): + user_detail = UserDetail( + user_name="Test User", + groups=[GroupSummary(id="jc_repo1_gr_test_group"), GroupSummary(id="jc_repo2_gr_test_group")], + ) + permitted = {"repo1"} + mocker.patch( + "server.services.utils.transformers.detect_affiliations", + return_value=( + None, + [ + _Group(repository_id="repo1", group_id="jc_repo1_gr_test_group", user_defined_id="test_group"), + _Group(repository_id="repo2", group_id="jc_repo2_gr_test_group", user_defined_id="test_group"), + ], + ), + ) + mocker.patch( + "server.services.groups.search", + return_value=SearchResult[GroupSummary]( + total=1, page_size=20, offset=1, resources=[GroupSummary(id="jc_repo1_gr_test_group")] + ), + ) + with pytest.raises(InvalidFormError) as exc: + transformers.validate_user_groups(user_detail, permitted) + assert str(exc.value) == str(E.USER_REQUIRES_EXISTING_GROUP % {"id": "jc_repo2_gr_test_group"}) + + +def test_validate_user_groups_forbidden(app, mocker: MockerFixture): + user_detail = UserDetail( + user_name="Test User", + groups=[GroupSummary(id="jc_repo1_gr_test_group"), GroupSummary(id="jc_repo2_gr_test_group")], + ) + permitted = {"repo1"} + mocker.patch( + "server.services.utils.transformers.detect_affiliations", + return_value=( + None, + [ + _Group(repository_id="repo1", group_id="jc_repo1_gr_test_group", user_defined_id="test_group"), + _Group(repository_id="repo2", group_id="jc_repo2_gr_test_group", user_defined_id="test_group"), + ], + ), + ) + mocker.patch( + "server.services.groups.search", + return_value=SearchResult[GroupSummary]( + total=2, + page_size=20, + offset=1, + resources=[GroupSummary(id="jc_repo1_gr_test_group"), GroupSummary(id="jc_repo2_gr_test_group")], + ), + ) + with pytest.raises(InvalidFormError) as exc: + transformers.validate_user_groups(user_detail, permitted) + assert str(exc.value) == str(E.USER_FORBIDDEN_GROUP % {"id": "jc_repo2_gr_test_group"}) + + +def test_validate_user_groups_non(app, mocker: MockerFixture): + user_detail = UserDetail( + user_name="Test User", + ) + permitted = {"repo1"} + assert transformers.validate_user_groups(user_detail, permitted) == [] + + @pytest.mark.parametrize( ("user_detail", "expected"), [ diff --git a/tests/unit/services/test_users.py b/tests/unit/services/test_users.py index 90d239e..c0c70c8 100644 --- a/tests/unit/services/test_users.py +++ b/tests/unit/services/test_users.py @@ -1,6 +1,7 @@ import typing as t from http import HTTPStatus +from pathlib import Path from unittest.mock import MagicMock import pytest @@ -10,15 +11,16 @@ from pytest_mock import MockerFixture from requests import Response -from server.clients.users import MapUser from server.const import USER_ROLES from server.entities.map_error import MapError +from server.entities.map_user import EPPN, Email, Group, MapUser from server.entities.search_request import SearchRequestParameter, SearchResponse, SearchResult from server.entities.summaries import UserSummary from server.entities.user_detail import RepositoryRole, UserDetail from server.exc import ( ApiClientError, CredentialsError, + InvalidExportError, InvalidFormError, InvalidQueryError, OAuthTokenError, @@ -26,12 +28,14 @@ ResourceNotFound, UnexpectedResponseError, ) +from server.messages import E from server.services import users from server.services.users import update, update_affiliations, update_put from server.services.utils import ( UsersCriteria, make_criteria_object, ) +from server.services.utils.affiliations import Affiliations, _Group, _RoleGroup from tests.helpers import load_json_data @@ -1416,6 +1420,91 @@ def test_handle_user_updated_eppns_false(mocker: MockerFixture) -> None: mock_clear_eppn.assert_not_called() +def test_make_export_file(app, mocker: MockerFixture) -> None: + operator_id = "test_user_1" + operator_name = "Test User 1" + mocker.patch("pathlib.Path.exists", return_value=True) + mocker.patch("pathlib.Path.mkdir", return_value=None) + mocker.patch("pathlib.Path.write_text", return_value=None) + mocker.patch("server.services.users.search", return_value=mocker.MagicMock(resources=[])) + mocker.patch("server.services.users.get_permitted_repository_ids", return_value={"repo1", "repo2"}) + mocker.patch( + "server.services.users._wite_user", + return_value=( + [{"id": "jc_repo1_test", "service_name": ""}], + [{"id": "jc_repo1_gr_test_1", "group_name": ""}, {"id": "jc_repo1_gr_test_2", "group_name": ""}], + [{"id": "user1", "user_name": "User 1"}, {"id": "user2", "user_name": "User 2"}], + ), + ) + mocker.patch("server.services.history_table.create_download_history", return_value=None) + mocker.patch("server.db.db.session.commit") + result = users.make_export_file(operator_id, operator_name) + assert isinstance(result, Path) + + +def test__wite_user(app, mocker: MockerFixture): + user_list = [ + MapUser( + id="user1", + user_name="User 1", + groups=[Group(value="jc_repo1_gr_test_1"), Group(value="jc_repo2_gr_test_1")], + emails=[Email(value="test@example.ac.jp")], + edu_person_principal_names=[EPPN(value="eppn1")], + ) + ] + delimiter = "," + file_path = Path("/test/path") + permitted_repository_ids = {"repo1"} + mocker.patch( + "server.services.users.detect_affiliations", + return_value=Affiliations( + roles=[], + groups=[ + _Group(repository_id="repo1", group_id="jc_repo1_gr_test_1", user_defined_id="test_1"), + _Group(repository_id="repo2", group_id="jc_repo2_gr_test_1", user_defined_id="test_1"), + ], + ), + ) + mocker.patch("server.services.users.is_current_user_system_admin", return_value=False) + mocker.patch("pathlib.Path.write_text", return_value=None) + file_repositories, file_groups, file_users = users._wite_user( # noqa: SLF001 + user_list, delimiter, file_path, permitted_repository_ids + ) + assert file_repositories == [{"id": "repo1", "service_name": ""}] + assert file_groups == [{"id": "jc_repo1_gr_test_1", "display_name": ""}] + assert file_users == [{"id": "user1", "user_name": "User 1"}] + + +@pytest.mark.parametrize( + ("affiliations", "expected"), + [ + ( + Affiliations(roles=[_RoleGroup(repository_id=None, role=USER_ROLES.SYSTEM_ADMIN)], groups=[]), + E.USER_CANNOT_EXPORT_SYSTEM_ADMIN, + ), + (Affiliations(roles=[], groups=[]), E.USER_FORBIDDEN_EXPORT), + ], +) +def test__wite_user_with_exception(app, mocker: MockerFixture, affiliations, expected) -> None: + user_list = [ + MapUser( + id="user1", + user_name="User 1", + groups=[Group(value="jc_repo1_gr_test_1")], + emails=[Email(value="test@example.ac.jp")], + edu_person_principal_names=[EPPN(value="eppn1")], + ) + ] + delimiter = "," + file_path = Path("/test/path") + permitted_repository_ids = {"repo1"} + mocker.patch("server.services.users.detect_affiliations", return_value=affiliations) + mocker.patch("server.services.users.is_current_user_system_admin", return_value=False) + with pytest.raises(InvalidExportError) as exc: + users._wite_user(user_list, delimiter, file_path, permitted_repository_ids) # noqa: SLF001 + assert str(exc.value) == str(expected) + + @pytest.fixture def user_data() -> tuple[dict[str, t.Any], MapUser]: json_data = load_json_data("data/map_user.json")