Keep an audit track of user renames

h/services/user_rename.py

@@ -1,4 +1,5 @@
-from h import models, tasks
+from h import tasks
+from h.models import Annotation, AuthTicket, User, UserRename
 
 
 class UserRenameError(Exception):
@@ -23,27 +24,36 @@ class UserRenameService:
     UserRenameError if the new username is already taken by another account.
     """
 
-    def __init__(self, session):
-        self.session = session
+    def __init__(self, db):
+        self.db = db
 
     def check(self, user, new_username):
-        existing_user = models.User.get_by_username(
-            self.session, new_username, user.authority
-        )
+        existing_user = User.get_by_username(self.db, new_username, user.authority)
         if existing_user and existing_user != user:
             raise UserRenameError(
                 f'Another user already has the username "{new_username}"'
             )
 
         return True
 
-    def rename(self, user, new_username):
+    def rename(self, user, new_username, requested_by: User, tag: str):
         self.check(user, new_username)
 
         old_userid = user.userid
         user.username = new_username
         new_userid = user.userid
 
+        # Record the rename for record-keeping purposes.
+        self.db.add(
+            UserRename(
+                user_id=user.id,
+                old_userid=old_userid,
+                new_userid=user.userid,
+                requested_by=requested_by.userid,
+                tag=tag,
+            )
+        )
+
         # Remove auth tickets when renaming the user. We cannot just update the
         # denormalized `user_userid` of these because the previous userid values
         # will have been serialized into the session cookies stored in the
@@ -60,9 +70,7 @@ def rename(self, user, new_username):
         )
 
     def _purge_auth_tickets(self, user):
-        self.session.query(models.AuthTicket).filter(
-            models.AuthTicket.user_id == user.id
-        ).delete()
+        self.db.query(AuthTicket).filter(AuthTicket.user_id == user.id).delete()
 
     def _change_annotations(self, old_userid, new_userid):
         annotations = self._fetch_annotations(old_userid)
@@ -72,12 +80,10 @@ def _change_annotations(self, old_userid, new_userid):
 
     def _fetch_annotations(self, userid):
         return (
-            self.session.query(models.Annotation)
-            .filter(models.Annotation.userid == userid)
-            .yield_per(100)
+            self.db.query(Annotation).filter(Annotation.userid == userid).yield_per(100)
         )
 
 
 def service_factory(_context, request):
     """Return a RenameUserService instance for the passed context and request."""
-    return UserRenameService(session=request.db)
+    return UserRenameService(db=request.db)

h/views/admin/users.py

@@ -6,7 +6,7 @@
 from h.accounts.events import ActivationEvent
 from h.i18n import TranslationString as _
 from h.security import Permission
-from h.services.user_rename import UserRenameError
+from h.services.user_rename import UserRenameError, UserRenameService
 
 
 class UserNotFoundError(Exception):
@@ -112,9 +112,9 @@ def users_rename(request):  # pragma: no cover
     old_username = user.username
     new_username = request.params.get("new_username").strip()
 
-    svc = request.find_service(name="user_rename")
+    svc: UserRenameService = request.find_service(name="user_rename")
     try:
-        svc.rename(user, new_username)
+        svc.rename(user, new_username, request.user, tag=request.matched_route.name)
 
     except (UserRenameError, ValueError) as exc:
         request.session.flash(str(exc), "error")

tests/unit/h/services/user_rename_test.py

@@ -44,52 +44,47 @@ def test_check_returns_True_if_new_username_equivalent_to_old(
 
         assert service.check(user, "panda") is True
 
-    def test_rename_checks_first(self, service, check, user):
-        service.rename(user, "panda")
-
-        check.assert_called_once_with(service, user, "panda")
-
-    def test_rename_changes_the_username(self, service, user, db_session):
-        service.rename(user, "panda")
-
-        assert db_session.get(models.User, user.id).username == "panda"
-
-    def test_rename_deletes_auth_tickets(self, service, user, db_session, factories):
+    def test_rename(self, service, check, user, factories, db_session):
         ids = [factories.AuthTicket(user=user).id for _ in range(3)]
+        token = factories.DeveloperToken(user=user)
+        requested_by = factories.User()
+        old_userid = user.userid
 
-        service.rename(user, "panda")
+        service.rename(user, "panda", requested_by, "tag")
 
-        count = (
+        check.assert_called_once_with(service, user, "panda")
+        assert db_session.get(models.User, user.id).username == "panda"
+        assert not (
             db_session.query(models.AuthTicket)
             .filter(models.AuthTicket.id.in_(ids))
             .count()
         )
-        assert not count
-
-    def test_rename_updates_tokens(self, service, user, db_session, factories):
-        token = factories.DeveloperToken(user=user)
-        db_session.add(token)
-
-        service.rename(user, "panda")
-
-        updated_token = (
-            db_session.query(models.Token).filter(models.Token.id == token.id).one()
+        assert (
+            db_session.query(models.Token)
+            .filter(models.Token.id == token.id)
+            .one()
+            .user
+            == user
         )
-        assert updated_token.user == user
+        user_rename = db_session.query(models.UserRename).first()
+        assert user_rename.old_userid == old_userid
+        assert user_rename.new_userid == user.userid
 
     @pytest.mark.usefixtures("annotations")
     def test_rename_changes_the_users_annotations_userid(
-        self, service, user, db_session
+        self, service, user, db_session, factories
     ):
-        service.rename(user, "panda")
+        service.rename(user, "panda", factories.User(), "tag")
 
         userids = [ann.userid for ann in db_session.query(models.Annotation)]
         assert {user.userid} == set(userids)
 
-    def test_rename_reindexes_the_users_annotations(self, service, user, tasks):
+    def test_rename_reindexes_the_users_annotations(
+        self, service, user, tasks, factories
+    ):
         original_userid = user.userid
 
-        service.rename(user, "panda")
+        service.rename(user, "panda", factories.User(), "tag")
 
         tasks.job_queue.add_annotations_from_user.delay.assert_called_once_with(
             "sync_annotation",
@@ -100,7 +95,7 @@ def test_rename_reindexes_the_users_annotations(self, service, user, tasks):
 
     @pytest.fixture
     def service(self, pyramid_request):
-        return UserRenameService(session=pyramid_request.db)
+        return UserRenameService(db=pyramid_request.db)
 
     @pytest.fixture
     def check(self, patch):
@@ -131,7 +126,7 @@ class TestServiceFactory:
     def test_it(self, pyramid_request, UserRenameService):
         svc = service_factory(sentinel.context, pyramid_request)
 
-        UserRenameService.assert_called_once_with(session=pyramid_request.db)
+        UserRenameService.assert_called_once_with(db=pyramid_request.db)
         assert svc == UserRenameService.return_value
 
     @pytest.fixture