Merge pull request #1454 from ScilifelabDataCentre/DDS-1623-new-super…

…admin-command-to-get-all-unitusers

Endpoint to get all unit user emails

SPRINTLOG.md

@@ -280,3 +280,4 @@ _Nothing merged in CLI during this sprint_
 - Rename storage-related columns in `Unit` table ([#1447](https://github.com/ScilifelabDataCentre/dds_web/pull/1447))
 - Dependency: Bump `cryptography` to 41.0.3 due to security vulnerability alerts(s) ([#1451](https://github.com/ScilifelabDataCentre/dds_web/pull/1451))
 - Allow for change of storage location ([#1448](https://github.com/ScilifelabDataCentre/dds_web/pull/1448))
+- Endpoint: `UnitUserEmails`; Return primary emails for Unit Personnel- and Admins ([#1454](https://github.com/ScilifelabDataCentre/dds_web/pull/1454))

dds_web/api/__init__.py

@@ -91,6 +91,7 @@ def output_json(data, code, headers=None):
 )
 api.add_resource(superadmin_only.AnyProjectsBusy, "/proj/busy/any", endpoint="projects_busy_any")
 api.add_resource(superadmin_only.Statistics, "/stats", endpoint="stats")
+api.add_resource(superadmin_only.UnitUserEmails, "/user/emails", endpoint="user_emails")
 
 # Invoicing ############################################################################ Invoicing #
 api.add_resource(user.ShowUsage, "/usage", endpoint="usage")

dds_web/api/superadmin_only.py

@@ -345,3 +345,23 @@ def get(self):
                 if stat_rows
             ]
         }
+
+
+class UnitUserEmails(flask_restful.Resource):
+    """Get emails for Unit Admins and Unit Personnel."""
+
+    @auth.login_required(role=["Super Admin"])
+    @logging_bind_request
+    @handle_db_error
+    def get(self):
+        """Collect the user emails and return a list."""
+        # Get all emails connected to a Unit Admin or Personnel account
+        user_emails = [user.primary_email for user in models.UnitUser.query.all()]
+
+        # Return empty if no emails
+        if not user_emails:
+            flask.current_app.logger.info("There are no primary emails to return.")
+            return {"empty": True}
+
+        # Return emails
+        return {"emails": user_emails}

tests/__init__.py

@@ -230,5 +230,6 @@ class DDSEndpoint:
     USER_FIND = BASE_ENDPOINT + "/user/find"
     TOTP_DEACTIVATE = BASE_ENDPOINT + "/user/totp/deactivate"
     STATS = BASE_ENDPOINT + "/stats"
+    USER_EMAILS = BASE_ENDPOINT + "/user/emails"
 
     TIMEOUT = 5

tests/api/test_superadmin_only.py

@@ -9,7 +9,8 @@
 import unittest
 from datetime import datetime, timedelta
 from unittest import mock
-
+from unittest.mock import patch
+from unittest.mock import PropertyMock
 
 # Installed
 import flask
@@ -898,3 +899,79 @@ def add_row_to_reporting_table(time):
         "TBHours Last Month": reporting_row.tbhours,
         "TBHours Total": reporting_row.tbhours_since_start,
     }
+
+
+# UnitUserEmails
+
+
+def test_unituseremails_accessdenied(client: flask.testing.FlaskClient) -> None:
+    """Only Super Admins can get the emails."""
+    no_access_users: typing.Dict = users.copy()
+    no_access_users.pop("Super Admin")
+
+    for u in no_access_users:
+        token: typing.Dict = get_token(username=users[u], client=client)
+        response: werkzeug.test.WrapperTestResponse = client.get(
+            tests.DDSEndpoint.USER_EMAILS, headers=token
+        )
+        assert response.status_code == http.HTTPStatus.FORBIDDEN
+
+
+def test_unituseremails_no_emails(client: flask.testing.FlaskClient) -> None:
+    """Empty should be returned if no emails."""
+    # No users returned from query
+    with patch("dds_web.database.models.UnitUser.query") as mock_users:
+        mock_users.return_value = []
+
+        # Authenticate
+        token: typing.Dict = get_token(username=users["Super Admin"], client=client)
+
+        # Call endpoint
+        response: werkzeug.test.WrapperTestResponse = client.get(
+            tests.DDSEndpoint.USER_EMAILS, headers=token
+        )
+        assert response.status_code == http.HTTPStatus.OK
+
+        # Verify response
+        assert response.json and response.json.get("empty") == True
+
+
+def test_unituseremails_ok(client: flask.testing.FlaskClient) -> None:
+    """Return user emails for unit users only."""
+    # Emails that should be returned
+    unituser_emails = [user.primary_email for user in models.UnitUser.query.all()]
+
+    # Emails that should not be returned
+    researcher_emails = [user.primary_email for user in models.ResearchUser.query.all()]
+    superadmin_emails = [user.primary_email for user in models.SuperAdmin.query.all()]
+    non_primary_emails = [
+        email.email for email in models.Email.query.filter_by(primary=False).all()
+    ]
+
+    # Authenticate
+    token: typing.Dict = get_token(username=users["Super Admin"], client=client)
+
+    # Call endpoint
+    response: werkzeug.test.WrapperTestResponse = client.get(
+        tests.DDSEndpoint.USER_EMAILS, headers=token
+    )
+    assert response.status_code == http.HTTPStatus.OK
+
+    # Verify response -------------------------------
+
+    # There should be a json response
+    json_response = response.json
+    assert json_response
+
+    # There should be emails in response
+    emails = json_response.get("emails")
+    assert emails
+
+    # The list of emails should contain all unit user primary emails
+    assert len(emails) == len(unituser_emails)
+    for e in unituser_emails:
+        assert e in emails
+
+    # The list of should not contain any of the other emails
+    for e in researcher_emails + superadmin_emails + non_primary_emails:
+        assert e not in emails