Set up UserUniqueLogin/IpAddress relationship (#19145)

* Fix ip_address_id type

* Update usage of UserUniqueLogin.ip_address

* Update tests

* Update translations

Author: di

tests/common/db/accounts.py

@@ -17,7 +17,6 @@
     UserUniqueLogin,
 )
 
-from ...common.constants import REMOTE_ADDR
 from .base import WarehouseFactory
 from .ip_addresses import IpAddressFactory
 
@@ -140,11 +139,4 @@ class Meta:
         model = UserUniqueLogin
 
     user = factory.SubFactory(UserFactory)
-    ip_address = REMOTE_ADDR
-    ip_address_id = factory.LazyAttribute(
-        lambda o: (
-            IpAddressFactory.create(ip_address=o.ip_address).id
-            if o.ip_address is not None
-            else None
-        )
-    )
+    ip_address = factory.SubFactory(IpAddressFactory)

tests/conftest.py

@@ -210,6 +210,7 @@ def pyramid_request(pyramid_services, jinja):
     dummy_request.user = None
     dummy_request.oidc_publisher = None
     dummy_request.metrics = dummy_request.find_service(IMetricsService)
+    dummy_request.ip_address = IpAddressFactory.create()
 
     dummy_request.registry.registerUtility(jinja, IJinja2Environment, name=".jinja2")
 

tests/functional/manage/test_account_publishing.py

@@ -9,6 +9,7 @@
 
 from tests.common.constants import REMOTE_ADDR
 from tests.common.db.accounts import UserFactory, UserUniqueLoginFactory
+from tests.common.db.ip_addresses import IpAddressFactory
 from warehouse.accounts.models import UniqueLoginStatus
 from warehouse.utils.otp import _get_totp
 
@@ -27,8 +28,9 @@ def test_add_pending_github_publisher_succeeds(self, webtest):
             with_terms_of_service_agreement=True,
             clear_pwd="password",
         )
+        ip_address = IpAddressFactory.create(ip_address=REMOTE_ADDR)
         UserUniqueLoginFactory.create(
-            user=user, ip_address=REMOTE_ADDR, status=UniqueLoginStatus.CONFIRMED
+            user=user, ip_address=ip_address, status=UniqueLoginStatus.CONFIRMED
         )
         # Create a response from GitHub API for owner details
         # during form submission validation.
@@ -106,8 +108,9 @@ def test_add_pending_gitlab_publisher_succeeds(self, webtest):
             with_terms_of_service_agreement=True,
             clear_pwd="password",
         )
+        ip_address = IpAddressFactory.create(ip_address=REMOTE_ADDR)
         UserUniqueLoginFactory.create(
-            user=user, ip_address=REMOTE_ADDR, status=UniqueLoginStatus.CONFIRMED
+            user=user, ip_address=ip_address, status=UniqueLoginStatus.CONFIRMED
         )
 
         # Act: Log in

tests/functional/manage/test_organization_publishing.py

@@ -9,6 +9,7 @@
 
 from tests.common.constants import REMOTE_ADDR
 from tests.common.db.accounts import UserFactory, UserUniqueLoginFactory
+from tests.common.db.ip_addresses import IpAddressFactory
 from tests.common.db.organizations import OrganizationFactory, OrganizationRoleFactory
 from warehouse.accounts.models import UniqueLoginStatus
 from warehouse.organizations.models import OrganizationRoleType
@@ -29,8 +30,9 @@ def test_add_pending_github_publisher_to_organization(self, webtest):
             with_terms_of_service_agreement=True,
             clear_pwd="password",
         )
+        ip_address = IpAddressFactory.create(ip_address=REMOTE_ADDR)
         UserUniqueLoginFactory.create(
-            user=user, ip_address=REMOTE_ADDR, status=UniqueLoginStatus.CONFIRMED
+            user=user, ip_address=ip_address, status=UniqueLoginStatus.CONFIRMED
         )
         organization = OrganizationFactory.create(name="test-organization")
         OrganizationRoleFactory.create(
@@ -114,8 +116,9 @@ def test_add_pending_gitlab_publisher_to_organization(self, webtest):
             with_terms_of_service_agreement=True,
             clear_pwd="password",
         )
+        ip_address = IpAddressFactory.create(ip_address=REMOTE_ADDR)
         UserUniqueLoginFactory.create(
-            user=user, ip_address=REMOTE_ADDR, status=UniqueLoginStatus.CONFIRMED
+            user=user, ip_address=ip_address, status=UniqueLoginStatus.CONFIRMED
         )
         organization = OrganizationFactory.create(name="test-organization")
         OrganizationRoleFactory.create(
@@ -188,8 +191,9 @@ def test_add_pending_google_publisher_to_organization(self, webtest):
             with_terms_of_service_agreement=True,
             clear_pwd="password",
         )
+        ip_address = IpAddressFactory.create(ip_address=REMOTE_ADDR)
         UserUniqueLoginFactory.create(
-            user=user, ip_address=REMOTE_ADDR, status=UniqueLoginStatus.CONFIRMED
+            user=user, ip_address=ip_address, status=UniqueLoginStatus.CONFIRMED
         )
         organization = OrganizationFactory.create(name="test-organization")
         OrganizationRoleFactory.create(
@@ -260,8 +264,9 @@ def test_add_pending_activestate_publisher_to_organization(self, webtest):
             with_terms_of_service_agreement=True,
             clear_pwd="password",
         )
+        ip_address = IpAddressFactory.create(ip_address=REMOTE_ADDR)
         UserUniqueLoginFactory.create(
-            user=user, ip_address=REMOTE_ADDR, status=UniqueLoginStatus.CONFIRMED
+            user=user, ip_address=ip_address, status=UniqueLoginStatus.CONFIRMED
         )
         organization = OrganizationFactory.create(name="test-organization")
         OrganizationRoleFactory.create(

tests/functional/manage/test_project_publishing.py

@@ -9,6 +9,7 @@
 
 from tests.common.constants import REMOTE_ADDR
 from tests.common.db.accounts import UserFactory, UserUniqueLoginFactory
+from tests.common.db.ip_addresses import IpAddressFactory
 from tests.common.db.packaging import ProjectFactory, RoleFactory
 from warehouse.accounts.models import UniqueLoginStatus
 from warehouse.utils.otp import _get_totp
@@ -30,8 +31,9 @@ def test_add_github_publisher_to_existing_project(self, webtest):
         )
         project = ProjectFactory.create(name="existing-project")
         RoleFactory.create(user=user, project=project, role_name="Owner")
+        ip_address = IpAddressFactory.create(ip_address=REMOTE_ADDR)
         UserUniqueLoginFactory.create(
-            user=user, ip_address=REMOTE_ADDR, status=UniqueLoginStatus.CONFIRMED
+            user=user, ip_address=ip_address, status=UniqueLoginStatus.CONFIRMED
         )
 
         # Mock GitHub API for owner validation
@@ -110,8 +112,9 @@ def test_add_gitlab_publisher_to_existing_project(self, webtest):
         )
         project = ProjectFactory.create(name="gitlab-project")
         RoleFactory.create(user=user, project=project, role_name="Owner")
+        ip_address = IpAddressFactory.create(ip_address=REMOTE_ADDR)
         UserUniqueLoginFactory.create(
-            user=user, ip_address=REMOTE_ADDR, status=UniqueLoginStatus.CONFIRMED
+            user=user, ip_address=ip_address, status=UniqueLoginStatus.CONFIRMED
         )
 
         # Act: Log in

tests/functional/manage/test_views.py

@@ -20,6 +20,7 @@
 from warehouse.utils.otp import _get_totp
 
 from ...common.db.accounts import EmailFactory, UserFactory, UserUniqueLoginFactory
+from ...common.db.ip_addresses import IpAddressFactory
 
 
 class TestManageAccount:
@@ -54,8 +55,9 @@ def test_changing_password_succeeds(self, webtest, socket_enabled):
             with_terms_of_service_agreement=True,
             clear_pwd="password",
         )
+        ip_address = IpAddressFactory.create(ip_address=REMOTE_ADDR)
         UserUniqueLoginFactory.create(
-            user=user, ip_address=REMOTE_ADDR, status=UniqueLoginStatus.CONFIRMED
+            user=user, ip_address=ip_address, status=UniqueLoginStatus.CONFIRMED
         )
 
         # visit login page

tests/unit/accounts/test_services.py

@@ -2018,17 +2018,13 @@ def test_factory(self):
 
 
 class TestDeviceIsKnown:
-    def test_device_is_known(self, user_service):
+    def test_device_is_known(self, user_service, db_request):
         user = UserFactory.create()
         UserUniqueLoginFactory.create(
-            user=user, ip_address=REMOTE_ADDR, status="confirmed"
+            user=user, ip_address=db_request.ip_address, status="confirmed"
         )
-        request = pretend.stub(
-            db=user_service.db,
-            remote_addr=REMOTE_ADDR,
-            find_service=lambda *a, **kw: pretend.stub(),
-        )
-        assert user_service.device_is_known(user.id, request)
+        db_request.find_service = lambda *a, **kw: pretend.stub()
+        assert user_service.device_is_known(user.id, db_request)
 
     def test_device_is_not_known(self, user_service, monkeypatch):
         user = UserFactory.create(with_verified_primary_email=True)
@@ -2054,7 +2050,7 @@ def test_device_is_not_known(self, user_service, monkeypatch):
             user_service.db.query(services.UserUniqueLogin)
             .filter(
                 services.UserUniqueLogin.user_id == user.id,
-                services.UserUniqueLogin.ip_address == REMOTE_ADDR,
+                services.UserUniqueLogin.ip_address.has(ip_address=REMOTE_ADDR),
             )
             .one()
         )
@@ -2070,55 +2066,40 @@ def test_device_is_not_known(self, user_service, monkeypatch):
             )
         ]
 
-    def test_device_is_pending_not_expired(self, user_service, monkeypatch):
+    def test_device_is_pending_not_expired(self, user_service, monkeypatch, db_request):
         user = UserFactory.create(with_verified_primary_email=True)
         UserUniqueLoginFactory.create(
-            user=user, ip_address=REMOTE_ADDR, status="pending"
+            user=user, ip_address=db_request.ip_address, status="pending"
         )
         send_email = pretend.call_recorder(lambda *a, **kw: None)
         monkeypatch.setattr(services, "send_unrecognized_login_email", send_email)
         token_service = pretend.stub(dumps=lambda d: "fake_token", max_age=60)
-        user_service.request = pretend.stub(
-            db=user_service.db,
-            remote_addr=REMOTE_ADDR,
-            headers={
-                "User-Agent": (
-                    "Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:15.0) "
-                    "Gecko/20100101 Firefox/15.0.1"
-                )
-            },
-            find_service=lambda *a, **kw: token_service,
-        )
+        user_service.request = db_request
+        db_request.find_service = lambda *a, **kw: token_service
 
         assert not user_service.device_is_known(user.id, user_service.request)
         assert send_email.calls == []
 
-    def test_device_is_pending_and_expired(self, user_service, monkeypatch):
+    def test_device_is_pending_and_expired(self, user_service, monkeypatch, db_request):
         user = UserFactory.create(with_verified_primary_email=True)
-        ip_address = IpAddressFactory.create(ip_address=REMOTE_ADDR)
         UserUniqueLoginFactory.create(
             user=user,
             status="pending",
-            ip_address=str(ip_address.ip_address),
-            ip_address_id=ip_address.id,
+            ip_address=db_request.ip_address,
             created=datetime.datetime(1970, 1, 1),
             expires=datetime.datetime(1970, 1, 1),
         )
         send_email = pretend.call_recorder(lambda *a, **kw: None)
         monkeypatch.setattr(services, "send_unrecognized_login_email", send_email)
         token_service = pretend.stub(dumps=lambda d: "fake_token", max_age=60)
-        user_service.request = pretend.stub(
-            db=user_service.db,
-            remote_addr=REMOTE_ADDR,
-            headers={
-                "User-Agent": (
-                    "Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:15.0) "
-                    "Gecko/20100101 Firefox/15.0.1"
-                )
-            },
-            find_service=lambda *a, **kw: token_service,
-            ip_address=ip_address,
-        )
+        user_service.request = db_request
+        db_request.find_service = lambda *a, **kw: token_service
+        db_request.headers = {
+            "User-Agent": (
+                "Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:15.0) "
+                "Gecko/20100101 Firefox/15.0.1"
+            )
+        }
 
         assert not user_service.device_is_known(user.id, user_service.request)
         assert send_email.calls == [

tests/unit/accounts/test_views.py

@@ -213,53 +213,47 @@ def test_no_query_string_raises_400(self):
         with pytest.raises(HTTPBadRequest):
             views.accounts_search(pyramid_request)
 
-    def test_returns_users_with_prefix(self, db_session, user_service):
+    def test_returns_users_with_prefix(self, db_request, user_service):
         foo = UserFactory.create(username="foo")
         bas = [
             UserFactory.create(username="bar"),
             UserFactory.create(username="baz"),
         ]
 
-        request = pretend.stub(
-            user=pretend.stub(),
-            find_service=lambda svc, **kw: {
-                IUserService: user_service,
-                IRateLimiter: pretend.stub(
-                    test=pretend.call_recorder(lambda ip_address: True),
-                    hit=pretend.call_recorder(lambda ip_address: None),
-                ),
-            }[svc],
-            ip_address=IpAddressFactory.build(),
-        )
+        db_request.user = pretend.stub()
+        db_request.find_service = lambda svc, **kw: {
+            IUserService: user_service,
+            IRateLimiter: pretend.stub(
+                test=pretend.call_recorder(lambda ip_address: True),
+                hit=pretend.call_recorder(lambda ip_address: None),
+            ),
+        }[svc]
 
-        request.params = MultiDict({"username": "f"})
-        result = views.accounts_search(request)
+        db_request.params = MultiDict({"username": "f"})
+        result = views.accounts_search(db_request)
         assert result == {"users": [foo]}
 
-        request.params = MultiDict({"username": "ba"})
-        result = views.accounts_search(request)
+        db_request.params = MultiDict({"username": "ba"})
+        result = views.accounts_search(db_request)
         assert result == {"users": bas}
 
-        request.params = MultiDict({"username": "zzz"})
+        db_request.params = MultiDict({"username": "zzz"})
         with pytest.raises(HTTPNotFound):
-            views.accounts_search(request)
+            views.accounts_search(db_request)
 
-    def test_when_rate_limited(self, db_session):
+    def test_when_rate_limited(self, db_request):
         search_limiter = pretend.stub(
             test=pretend.call_recorder(lambda ip_address: False),
         )
-        request = pretend.stub(
-            user=pretend.stub(),
-            find_service=lambda svc, **kw: {
-                IRateLimiter: search_limiter,
-            }[svc],
-            ip_address=IpAddressFactory.build(),
-        )
+        db_request.user = pretend.stub()
+        db_request.find_service = lambda svc, **kw: {
+            IRateLimiter: search_limiter,
+        }[svc]
 
-        request.params = MultiDict({"username": "foo"})
-        result = views.accounts_search(request)
+        db_request.params = MultiDict({"username": "foo"})
+        result = views.accounts_search(db_request)
 
-        assert search_limiter.test.calls == [pretend.call(request.ip_address)]
+        assert search_limiter.test.calls == [pretend.call(db_request.ip_address)]
         assert result == {"users": []}
 
 
@@ -641,8 +635,7 @@ def test_login_with_remembered_device_confirms_unique_login(
 
         UserUniqueLoginFactory.create(
             user=user,
-            ip_address=str(db_request.ip_address.ip_address),
-            ip_address_id=db_request.ip_address.id,
+            ip_address=db_request.ip_address,
             status=UniqueLoginStatus.PENDING,
         )
 
@@ -700,8 +693,7 @@ def test_login_updates_last_used(self, monkeypatch, db_request, pyramid_services
         past_timestamp = datetime.datetime(1970, 1, 1)
         UserUniqueLoginFactory.create(
             user=user,
-            ip_address=str(db_request.ip_address.ip_address),
-            ip_address_id=db_request.ip_address.id,
+            ip_address=db_request.ip_address,
             status=UniqueLoginStatus.CONFIRMED,
             last_used=past_timestamp,
         )
@@ -5475,7 +5467,8 @@ def test_unique_login_not_found(self, db_request):
 
     def test_ip_address_mismatch(self, db_request):
         user = UserFactory.create(last_login=datetime.datetime.now(datetime.UTC))
-        unique_login = UserUniqueLoginFactory.create(user=user, ip_address="1.1.1.1")
+        ip_address = IpAddressFactory.create(ip_address="1.1.1.1")
+        unique_login = UserUniqueLoginFactory.create(user=user, ip_address=ip_address)
         db_request.user = None
         db_request.params = {"token": "foo"}
         token_data = {
@@ -5510,8 +5503,7 @@ def test_success(self, monkeypatch, db_request):
         user = UserFactory.create(last_login=datetime.datetime.now(datetime.UTC))
         unique_login = UserUniqueLoginFactory.create(
             user=user,
-            ip_address=str(db_request.ip_address.ip_address),
-            ip_address_id=db_request.ip_address.id,
+            ip_address=db_request.ip_address,
         )
         db_request.user = None
         db_request.params = {"token": "foo"}