Remove usage of username in favor of email

temba/api/support.py

@@ -66,7 +66,7 @@ class APIBasicAuthentication(RequestAttributesMixin, BasicAuthentication):
 
     Clients should authenticate using HTTP Basic Authentication.
 
-    Credentials: username:api_token
+    Credentials: email:api_token
     """
 
     def authenticate_credentials(self, userid, password, request=None):
@@ -75,7 +75,7 @@ def authenticate_credentials(self, userid, password, request=None):
         except APIToken.DoesNotExist:
             raise exceptions.AuthenticationFailed("Invalid token or email")
 
-        if token.user.username != userid:
+        if token.user.email != userid:
             raise exceptions.AuthenticationFailed("Invalid token or email")
 
         if token.user.is_active:

temba/api/v2/tests/test_base.py

@@ -100,7 +100,7 @@ def request_by_session(endpoint, user, post_data=None):
         response = request_by_token(fields_url, token2.key, {"name": "Field 2", "type": "text"})
         self.assertEqual(201, response.status_code)
 
-        response = request_by_basic_auth(fields_url, self.admin.username, token1.key)
+        response = request_by_basic_auth(fields_url, self.admin.email, token1.key)
         self.assertEqual(200, response.status_code)
 
         # can GET using session auth for admins, editors and servicing staff
@@ -129,7 +129,7 @@ def request_by_session(endpoint, user, post_data=None):
         self.assertResponseError(response, None, "Invalid token", status_code=403)
 
         # can't fetch endpoint with invalid token
-        response = request_by_basic_auth(contacts_url, self.admin.username, "1234567890")
+        response = request_by_basic_auth(contacts_url, self.admin.email, "1234567890")
         self.assertResponseError(response, None, "Invalid token or email", status_code=403)
 
         # can't fetch endpoint with invalid username
@@ -141,7 +141,7 @@ def request_by_session(endpoint, user, post_data=None):
         self.assertEqual(200, response.status_code)
         self.assertEqual(str(self.org.id), response["X-Temba-Org"])
 
-        response = request_by_basic_auth(contacts_url, self.editor.username, token2.key)
+        response = request_by_basic_auth(contacts_url, self.editor.email, token2.key)
         self.assertEqual(200, response.status_code)
         self.assertEqual(str(self.org.id), response["X-Temba-Org"])
 
@@ -153,7 +153,7 @@ def request_by_session(endpoint, user, post_data=None):
         self.assertEqual(response.status_code, 429)
 
         # same with basic auth
-        response = request_by_basic_auth(fields_url, self.admin.username, token1.key)
+        response = request_by_basic_auth(fields_url, self.admin.email, token1.key)
         self.assertEqual(response.status_code, 429)
 
         # or if another user in same org makes a request
@@ -168,7 +168,7 @@ def request_by_session(endpoint, user, post_data=None):
         self.org.api_rates = {"v2": "15000/hour"}
         self.org.save(update_fields=("api_rates",))
 
-        response = request_by_basic_auth(fields_url, self.admin.username, token1.key)
+        response = request_by_basic_auth(fields_url, self.admin.email, token1.key)
         self.assertEqual(response.status_code, 200)
 
         cache.set(f"throttle_v2_{self.org.id}", [time.time() for r in range(15000)])
@@ -181,7 +181,7 @@ def request_by_session(endpoint, user, post_data=None):
         self.org.add_user(self.admin, OrgRole.AGENT)
 
         self.assertEqual(request_by_token(campaigns_url, token1.key).status_code, 403)
-        self.assertEqual(request_by_basic_auth(campaigns_url, self.admin.username, token1.key).status_code, 403)
+        self.assertEqual(request_by_basic_auth(campaigns_url, self.admin.email, token1.key).status_code, 403)
 
         # and if user is inactive, disallow the request
         self.org.add_user(self.admin, OrgRole.ADMINISTRATOR)
@@ -191,7 +191,7 @@ def request_by_session(endpoint, user, post_data=None):
         response = request_by_token(contacts_url, token1.key)
         self.assertResponseError(response, None, "Invalid token", status_code=403)
 
-        response = request_by_basic_auth(contacts_url, self.admin.username, token1.key)
+        response = request_by_basic_auth(contacts_url, self.admin.email, token1.key)
         self.assertResponseError(response, None, "Invalid token or email", status_code=403)
 
     @override_settings(SECURE_PROXY_SSL_HEADER=("HTTP_X_FORWARDED_HTTPS", "https"))

temba/channels/tests.py

@@ -6,8 +6,6 @@
 from unittest.mock import patch
 from urllib.parse import quote
 
-from smartmin.tests import SmartminTest
-
 from django.conf import settings
 from django.contrib.auth.models import Group
 from django.core import mail
@@ -1113,7 +1111,7 @@ def test_delete(self):
         self.assertNotIn(self.ex_channel, flow.channel_dependencies.all())
 
 
-class SyncEventTest(SmartminTest):
+class SyncEventTest(TembaTest):
     def setUp(self):
         self.user = self.create_user("tito")
         self.org = Org.objects.create(

temba/flows/tests/test_flow.py

@@ -132,7 +132,7 @@ def test_ensure_current_version(self):
         # check we migrate to current spec version
         self.assertEqual("13.6.1", flow.version_number)
         self.assertEqual(2, flow.revisions.count())
-        self.assertEqual("system", flow.revisions.order_by("id").last().created_by.username)
+        self.assertEqual("system", flow.revisions.order_by("id").last().created_by.email)
 
         # saved on won't have been updated but modified on will
         self.assertEqual(old_saved_on, flow.saved_on)

temba/orgs/models.py

@@ -916,7 +916,7 @@ def create_sample_flows(self, api_url):
         user = self.get_admins().first()
         if user:
             # some some substitutions
-            samples = samples.replace("{{EMAIL}}", user.username).replace("{{API_URL}}", api_url)
+            samples = samples.replace("{{EMAIL}}", user.email).replace("{{API_URL}}", api_url)
 
             try:
                 self.import_app(json.loads(samples), user)

temba/orgs/views/tests/test_loginviews.py

@@ -38,7 +38,7 @@ def test_login(self):
         self.assertFormError(
             response.context["form"],
             None,
-            "Please enter a correct username and password. Note that both fields may be case-sensitive.",
+            "Please enter a correct email address and password. Note that both fields may be case-sensitive.",
         )
 
         # submit incorrect password by case sensitivity
@@ -47,7 +47,7 @@ def test_login(self):
         self.assertFormError(
             response.context["form"],
             None,
-            "Please enter a correct username and password. Note that both fields may be case-sensitive.",
+            "Please enter a correct email address and password. Note that both fields may be case-sensitive.",
         )
 
         # submit correct username and password
@@ -131,7 +131,7 @@ def test_login_lockouts(self):
         self.assertFormError(
             response.context["form"],
             None,
-            "Please enter a correct username and password. Note that both fields may be case-sensitive.",
+            "Please enter a correct email address and password. Note that both fields may be case-sensitive.",
         )
 
         # and successful logins

temba/orgs/views/tests/test_orgcrudl.py

@@ -451,8 +451,8 @@ def test_org_grant(self):
         self.assertEqual(org.date_format, Org.DATE_FORMAT_DAY_FIRST)
 
         # check user exists and is admin
-        self.assertEqual(OrgRole.ADMINISTRATOR, org.get_user_role(User.objects.get(username="[email protected]")))
-        self.assertEqual(OrgRole.ADMINISTRATOR, org.get_user_role(User.objects.get(username="[email protected]")))
+        self.assertEqual(OrgRole.ADMINISTRATOR, org.get_user_role(User.objects.get(email="[email protected]")))
+        self.assertEqual(OrgRole.ADMINISTRATOR, org.get_user_role(User.objects.get(email="[email protected]")))
 
         # try a new org with a user that already exists instead
         del post_data["password"]
@@ -464,8 +464,8 @@ def test_org_grant(self):
         org = Org.objects.get(name="id Software")
         self.assertEqual(org.date_format, Org.DATE_FORMAT_DAY_FIRST)
 
-        self.assertEqual(OrgRole.ADMINISTRATOR, org.get_user_role(User.objects.get(username="[email protected]")))
-        self.assertEqual(OrgRole.ADMINISTRATOR, org.get_user_role(User.objects.get(username="[email protected]")))
+        self.assertEqual(OrgRole.ADMINISTRATOR, org.get_user_role(User.objects.get(email="[email protected]")))
+        self.assertEqual(OrgRole.ADMINISTRATOR, org.get_user_role(User.objects.get(email="[email protected]")))
 
         # try a new org with US timezone
         post_data["name"] = "Bulls"
@@ -828,9 +828,7 @@ def test_signup(self):
         response = self.client.post(edit_url, post_data, HTTP_X_FORMAX=True)
         self.assertEqual(200, response.status_code)
 
-        self.assertTrue(User.objects.get(username="[email protected]"))
         self.assertTrue(User.objects.get(email="[email protected]"))
-        self.assertFalse(User.objects.filter(username="[email protected]"))
         self.assertFalse(User.objects.filter(email="[email protected]"))
 
     def test_create_new(self):

temba/orgs/views/tests/test_usercrudl.py

@@ -323,7 +323,6 @@ def test_edit(self):
         )
 
         self.admin.refresh_from_db()
-        self.assertEqual("[email protected]", self.admin.username)
         self.assertEqual("[email protected]", self.admin.email)
         self.assertEqual("U", self.admin.email_status)  # because email changed
         self.assertNotEqual("old-email-secret", self.admin.email_verification_secret)

temba/orgs/views/views.py

@@ -151,7 +151,7 @@ def post(self, request, *args, **kwargs):
         if not username:
             return self.form_invalid(form)
 
-        user = User.objects.filter(username__iexact=username).first()
+        user = User.objects.filter(email__iexact=username).first()
         valid_password = False
 
         # this could be a valid login by a user
@@ -252,10 +252,10 @@ def form_invalid(self, form):
         lockout_timeout = getattr(settings, "USER_LOCKOUT_TIMEOUT", 10)
         failed_login_limit = getattr(settings, "USER_FAILED_LOGIN_LIMIT", 5)
 
-        FailedLogin.objects.create(username=user.username)
+        FailedLogin.objects.create(username=user.email)
 
         bad_interval = timezone.now() - timedelta(minutes=lockout_timeout)
-        failures = FailedLogin.objects.filter(username__iexact=user.username)
+        failures = FailedLogin.objects.filter(username__iexact=user.email)
 
         # if the failures reset after a period of time, then limit our query to that interval
         if lockout_timeout > 0:
@@ -280,7 +280,7 @@ def form_valid(self, form):
         self.reset_user()
 
         # cleanup any failed logins
-        FailedLogin.objects.filter(username__iexact=user.username).delete()
+        FailedLogin.objects.filter(username__iexact=user.email).delete()
 
         return HttpResponseRedirect(self.get_success_url())
 
@@ -368,7 +368,7 @@ def dispatch(self, request, *args, **kwargs):
         return super().dispatch(request, *args, **kwargs)
 
     def get_username(self, form):
-        return self.request.user.username
+        return self.request.user.email
 
 
 class UserCRUDL(SmartCRUDL):
@@ -678,7 +678,7 @@ def post_save(self, obj):
             obj.recovery_tokens.all().delete()
 
             # delete any failed login records
-            FailedLogin.objects.filter(username__iexact=obj.username).delete()
+            FailedLogin.objects.filter(username__iexact=obj.email).delete()
 
             return obj
 
@@ -758,9 +758,6 @@ def derive_initial(self):
         def pre_save(self, obj):
             obj = super().pre_save(obj)
 
-            # keep our username and email in sync and record if email is changing
-            obj.username = obj.email
-
             # get existing email address to know if it's changing
             obj._prev_email = User.objects.get(id=obj.id).email
 
@@ -910,7 +907,7 @@ def get_context_data(self, **kwargs):
 
             brand = self.request.branding["name"]
             user = self.request.user
-            secret_url = pyotp.TOTP(user.two_factor_secret).provisioning_uri(user.username, issuer_name=brand)
+            secret_url = pyotp.TOTP(user.two_factor_secret).provisioning_uri(user.email, issuer_name=brand)
 
             context["secret_url"] = secret_url
             return context
@@ -1820,7 +1817,7 @@ def pre_process(self, request, *args, **kwargs):
 
             # if user exists and is logged in then they just need to accept
             user = User.get_by_email(self.invitation.email)
-            if user and self.invitation.email.lower() == request.user.username.lower():
+            if user and self.invitation.email.lower() == request.user.email.lower():
                 return HttpResponseRedirect(reverse("orgs.org_join_accept", args=[secret]))
 
             logout(request)
@@ -1860,7 +1857,7 @@ def save(self, obj):
             )
 
             # log the user in
-            user = authenticate(username=user.username, password=self.form.cleaned_data["password"])
+            user = authenticate(username=user.email, password=self.form.cleaned_data["password"])
             login(self.request, user)
 
             self.invitation.accept(user)
@@ -1890,7 +1887,7 @@ def pre_process(self, request, *args, **kwargs):
 
             # if user doesn't already exist or we're logged in as a different user, we shouldn't be here
             user = User.get_by_email(self.invitation.email)
-            if not user or self.invitation.email != request.user.username:
+            if not user or self.invitation.email != request.user.email:
                 return HttpResponseRedirect(reverse("orgs.org_join", args=[self.kwargs["secret"]]))
 
             return None

temba/staff/tests.py

@@ -284,7 +284,6 @@ def test_update(self):
 
         self.editor.refresh_from_db()
         self.assertEqual("[email protected]", self.editor.email)
-        self.assertEqual("[email protected]", self.editor.username)  # should match email
         self.assertEqual("Edward", self.editor.first_name)
         self.assertEqual("", self.editor.last_name)
         self.assertEqual({granters, betas}, set(self.editor.groups.all()))
@@ -305,7 +304,6 @@ def test_update(self):
 
         self.editor.refresh_from_db()
         self.assertEqual("[email protected]", self.editor.email)
-        self.assertEqual("[email protected]", self.editor.username)
         self.assertEqual("Edward", self.editor.first_name)
         self.assertEqual("", self.editor.last_name)
         self.assertEqual({granters}, set(self.editor.groups.all()))

temba/staff/views.py

@@ -302,10 +302,6 @@ class Meta:
         success_message = "User updated successfully."
         title = "Update User"
 
-        def pre_save(self, obj):
-            obj.username = obj.email
-            return obj
-
         def post_save(self, obj):
             """
             Make sure our groups are up-to-date

temba/tests/base.py

@@ -143,8 +143,8 @@ def tearDown(self):
 
     def login(self, user, *, update_last_auth_on: bool = True, choose_org=None):
         self.assertTrue(
-            self.client.login(username=user.username, password=self.default_password),
-            f"couldn't login as {user.username}:{self.default_password}",
+            self.client.login(username=user.email, password=self.default_password),
+            f"couldn't login as {user.email}:{self.default_password}",
         )
 
         # infer our org if we weren't handed one
@@ -194,8 +194,7 @@ def get_flow(self, filename, substitutions=None, name=None):
         return flow
 
     def create_user(self, email, group_names=(), **kwargs):
-        user = User.objects.create_user(username=email, email=email, **kwargs)
-        user.set_password(self.default_password)
+        user = User.objects.create_user(email=email, password=self.default_password, **kwargs)
         user.save()
 
         for group in group_names:

temba/tests/crudl.py

@@ -14,7 +14,7 @@ def requestView(self, url, user, *, post_data=None, checks=(), choose_org=None,
         """
 
         method = "POST" if post_data is not None else "GET"
-        user_name = user.username if user else "anonymous"
+        user_name = user.email if user else "anonymous"
         msg_prefix = f"{method} {url} as {user_name}"
         pre_msg_prefix = f"before {msg_prefix}"
 

temba/users/apps.py

@@ -19,4 +19,4 @@ def on_post_migrate(sender, **kwargs):
     try:
         User.get_system_user()
     except User.DoesNotExist:
-        User._create_system_user()
+        User.objects.create_system_user()

temba/users/migrations/0011_alter_user_email_alter_user_username.py

@@ -0,0 +1,23 @@
+# Generated by Django 5.1.4 on 2025-02-21 01:09
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("users", "0010_alter_user_managers_alter_user_date_joined_and_more"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="user",
+            name="email",
+            field=models.EmailField(max_length=254, unique=True, verbose_name="email address"),
+        ),
+        migrations.AlterField(
+            model_name="user",
+            name="username",
+            field=models.CharField(max_length=150, null=True, verbose_name="username"),
+        ),
+    ]