Expired link is deleted from database when expired

SPRINTLOG.md

@@ -299,4 +299,5 @@ _Nothing merged in CLI during this sprint_
 - Set `sto2*` columns in `Unit` table to nullable ([#1456](https://github.com/ScilifelabDataCentre/dds_web/pull/1462))
 - Dependency: Bump `MariaDB` to LTS version 10.11.5 ([#1465](https://github.com/ScilifelabDataCentre/dds_web/pull/1465))
 - Bug fixed: Row in `ProjectUsers` should also be added if it doesn't exist when giving Researcher access to a specific project ([#1464](https://github.com/ScilifelabDataCentre/dds_web/pull/1464))
+- Replace expired invites when there's a new invitation attempt ([#1466](https://github.com/ScilifelabDataCentre/dds_web/pull/1466))
 - Workflow: Update PR template and clarify sections ([#1467](https://github.com/ScilifelabDataCentre/dds_web/pull/1467))

dds_web/api/schemas/user_schemas.py

@@ -4,6 +4,9 @@
 # IMPORTS ################################################################################ IMPORTS #
 ####################################################################################################
 
+# standar library
+import datetime
+
 # Installed
 import flask
 import marshmallow
@@ -72,6 +75,18 @@ def return_invite(self, data, **kwargs):
         # double check if there is no existing user with this email
         userexists = utils.email_in_db(email=data.get("email"))
 
+        # check if the user invite should have expired and be deleted
+        if invite and (
+            (
+                datetime.datetime.utcnow()
+                - datetime.timedelta(hours=flask.current_app.config["INVITATION_EXPIRES_IN_HOURS"])
+            )
+            > invite.created_at
+        ):
+            db.session.delete(invite)
+            db.session.commit()
+            invite = None
+
         if userexists and invite:
             raise ddserr.DatabaseError(message="Email exists for user and invite at the same time")
 

dds_web/api/user.py

@@ -83,8 +83,13 @@ def post(self):
         try:
             existing_user = user_schemas.UserSchema().load({"email": email})
             unanswered_invite = user_schemas.UnansweredInvite().load({"email": email})
-        except sqlalchemy.exc.OperationalError as err:
-            raise ddserr.DatabaseError(message=str(err), alt_message="Unexpected database error.")
+        except (sqlalchemy.exc.SQLAlchemyError, sqlalchemy.exc.OperationalError) as err:
+            db.session.rollback()
+            raise ddserr.DatabaseError(
+                message=str(err),
+                alt_message="Something happened while checking for existig account / active invite.",
+                pass_message=False,
+            )
 
         if existing_user or unanswered_invite:
             if not project:

tests/api/test_user.py

@@ -272,6 +272,131 @@ def test_add_unitadmin_user_with_unitpersonnel_permission_denied(client):
     assert invited_user is None
 
 
+def test_invite_user_expired_not_deleted(client):
+    """If an invite has expired and hasn't been removed from the database, the invite should be replaced"""
+
+    # invite a new user
+    response = client.post(
+        tests.DDSEndpoint.USER_ADD,
+        headers=tests.UserAuth(tests.USER_CREDENTIALS["unitadmin"]).token(client),
+        json=first_new_user,
+    )
+    assert response.status_code == http.HTTPStatus.OK
+
+    # Set the creation date in the DB to -7 days for now
+    invited_user = models.Invite.query.filter_by(email=first_new_user["email"]).one_or_none()
+    invited_user.created_at -= timedelta(hours=168)
+    old_time = invited_user.created_at
+    db.session.commit()
+
+    # Send the invite again and confirm it works
+    response = client.post(
+        tests.DDSEndpoint.USER_ADD,
+        headers=tests.UserAuth(tests.USER_CREDENTIALS["unitadmin"]).token(client),
+        json=first_new_user,
+    )
+    assert response.status_code == http.HTTPStatus.OK
+
+    invited_user = models.Invite.query.filter_by(email=first_new_user["email"]).one_or_none()
+    assert invited_user
+
+    # check that the date has been updated
+    assert invited_user.created_at != old_time
+
+
+def test_invite_user_existing_project_invite_expired(client):
+    """If an invite to a project has expired and hasn't been removed, a new invite should replace the old one"""
+
+    project = models.Project.query.filter_by(public_id="public_project_id").one_or_none()
+
+    # invite a new user
+    response = client.post(
+        tests.DDSEndpoint.USER_ADD,
+        headers=tests.UserAuth(tests.USER_CREDENTIALS["unitadmin"]).token(client),
+        query_string={"project": project.public_id},
+        json=first_new_user,
+    )
+    assert response.status_code == http.HTTPStatus.OK
+
+    invited_user = models.Invite.query.filter_by(email=first_new_user["email"]).one_or_none()
+    assert invited_user
+
+    # check row was added to project invite keys table
+    project_invite_keys = models.ProjectInviteKeys.query.filter_by(
+        invite_id=invited_user.id, project_id=project.id
+    ).one_or_none()
+    assert project_invite_keys
+
+    # Set the creation date in the DB to -7 days for now
+    invited_user.created_at -= timedelta(hours=168)
+    old_time = invited_user.created_at
+    db.session.commit()
+
+    # Send the invite again and confirm it works
+    response = client.post(
+        tests.DDSEndpoint.USER_ADD,
+        headers=tests.UserAuth(tests.USER_CREDENTIALS["unitadmin"]).token(client),
+        query_string={"project": project.public_id},
+        json=first_new_user,
+    )
+
+    assert response.status_code == http.HTTPStatus.OK
+
+    invited_user = models.Invite.query.filter_by(email=first_new_user["email"]).one_or_none()
+    assert invited_user
+
+    # check that the date has been updated
+    assert invited_user.created_at != old_time
+
+    # check that the project invite keys has a new row
+    project_invite_keys_new = models.ProjectInviteKeys.query.filter_by(
+        invite_id=invited_user.id, project_id=project.id
+    ).one_or_none()
+    assert project_invite_keys_new != project_invite_keys
+
+
+def test_invite_user_expired_sqlalchemyerror(client):
+    """Error message should be returned if sqlalchemyerror occurs during deletion of unanswered invite."""
+
+    # Invite a new user
+    response = client.post(
+        tests.DDSEndpoint.USER_ADD,
+        headers=tests.UserAuth(tests.USER_CREDENTIALS["unitadmin"]).token(client),
+        json=first_new_user,
+    )
+    assert response.status_code == http.HTTPStatus.OK
+
+    # Set the creation date in the DB to -7 days for now
+    invited_user = models.Invite.query.filter_by(email=first_new_user["email"]).one_or_none()
+    invited_user.created_at -= timedelta(hours=168)
+    old_time = invited_user.created_at
+    old_id = invited_user.id
+    db.session.commit()
+
+    from tests.api.test_project import mock_sqlalchemyerror
+
+    # Simulate database error while trying to send new invite
+    with unittest.mock.patch("dds_web.db.session.delete", mock_sqlalchemyerror):
+        response = client.post(
+            tests.DDSEndpoint.USER_ADD,
+            headers=tests.UserAuth(tests.USER_CREDENTIALS["unitadmin"]).token(client),
+            json=first_new_user,
+        )
+        assert response.status_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
+        assert (
+            response.json.get("message")
+            == "Something happened while checking for existig account / active invite."
+        )
+
+    # Get invite again
+    invited_user = models.Invite.query.filter_by(email=first_new_user["email"]).one_or_none()
+    assert invited_user
+
+    # The invite should be the same
+    assert invited_user.created_at == old_time
+    assert invited_user.id == old_id
+
+
 # -- Add existing users to projects ################################# Add existing users to projects #
 def test_add_existing_user_without_project(client):
     """Project required if inviting user to project."""