add unit test for cov and remove unuse function

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."""

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

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

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)}

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)

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})