pypi · di · Apr 5, 2022 · Feb 14, 2022 · Feb 14, 2022 · Feb 14, 2022
@@ -49,3 +49,4 @@ GITHUB_TOKEN_SCANNING_META_API_URL="http://notgithub:8000/meta/public_keys/token
 TWOFACTORREQUIREMENT_ENABLED=true
 TWOFACTORMANDATE_AVAILABLE=true
 TWOFACTORMANDATE_ENABLED=true
+OIDC_ENABLED=true
@@ -3575,3 +3575,96 @@ def test_recovery_code_emails(
                 },
             )
         ]
+
+
+class TestOIDCProviderEmails:
+    @pytest.mark.parametrize(
+        "fn, template_name",
+        [
+            (email.send_oidc_provider_added_email, "oidc-provider-added"),
+            (email.send_oidc_provider_removed_email, "oidc-provider-removed"),
+        ],
+    )
+    def test_oidc_provider_emails(
+        self, pyramid_request, pyramid_config, monkeypatch, fn, template_name
+    ):
+        stub_user = pretend.stub(
+            id="id",
+            username="username",
+            name="",
+            email="email@example.com",
+            primary_email=pretend.stub(email="email@example.com", verified=True),
+        )
+        subject_renderer = pyramid_config.testing_add_renderer(
+            f"email/{ template_name }/subject.txt"
+        )
+        subject_renderer.string_response = "Email Subject"
+        body_renderer = pyramid_config.testing_add_renderer(
+            f"email/{ template_name }/body.txt"
+        )
+        body_renderer.string_response = "Email Body"
+        html_renderer = pyramid_config.testing_add_renderer(
+            f"email/{ template_name }/body.html"
+        )
+        html_renderer.string_response = "Email HTML Body"
+
+        send_email = pretend.stub(
+            delay=pretend.call_recorder(lambda *args, **kwargs: None)
+        )
+        pyramid_request.task = pretend.call_recorder(lambda *args, **kwargs: send_email)
+        monkeypatch.setattr(email, "send_email", send_email)
+
+        pyramid_request.db = pretend.stub(
+            query=lambda a: pretend.stub(
+                filter=lambda *a: pretend.stub(
+                    one=lambda: pretend.stub(user_id=stub_user.id)
+                )
+            ),
+        )
+        pyramid_request.user = stub_user
+        pyramid_request.registry.settings = {"mail.sender": "noreply@example.com"}
+
+        project_name = "test_project"
+        fakeprovider = pretend.stub(provider_name="fakeprovider")
+        # NOTE: Can't set __str__ using pretend.stub()
+        monkeypatch.setattr(
+            fakeprovider.__class__, "__str__", lambda s: "fakespecifier"
+        )
+
+        result = fn(
+            pyramid_request, stub_user, project_name=project_name, provider=fakeprovider
+        )
+
+        assert result == {
+            "username": stub_user.username,
+            "project_name": project_name,
+            "provider_name": "fakeprovider",
+            "provider_spec": "fakespecifier",
+        }
+        subject_renderer.assert_()
+        body_renderer.assert_(username=stub_user.username, project_name=project_name)
+        html_renderer.assert_(username=stub_user.username, project_name=project_name)
+        assert pyramid_request.task.calls == [pretend.call(send_email)]
+        assert send_email.delay.calls == [
+            pretend.call(
+                f"{stub_user.username} <{stub_user.email}>",
+                {
+                    "subject": "Email Subject",
+                    "body_text": "Email Body",
+                    "body_html": (
+                        "<html>\n<head></head>\n"
+                        "<body><p>Email HTML Body</p></body>\n</html>\n"
+                    ),
+                },
+                {
+                    "tag": "account:email:sent",
+                    "user_id": stub_user.id,
+                    "additional": {
+                        "from_": "noreply@example.com",
+                        "to": stub_user.email,
+                        "subject": "Email Subject",
+                        "redact_ip": False,
+                    },
+                },
+            )
+        ]
@@ -94,13 +94,37 @@ def view(context, request):
         assert request.session.needs_reauthentication.calls == needs_reauth_calls
 
 
-def test_includeme():
+def test_includeme(monkeypatch):
+    settings = {
+        "warehouse.manage.oidc.user_registration_ratelimit_string": "10 per day",
+        "warehouse.manage.oidc.ip_registration_ratelimit_string": "100 per day",
+    }
+
     config = pretend.stub(
         add_view_deriver=pretend.call_recorder(lambda f, over, under: None),
+        register_service_factory=pretend.call_recorder(lambda s, i, **kw: None),
+        registry=pretend.stub(
+            settings=pretend.stub(get=pretend.call_recorder(lambda k: settings.get(k)))
+        ),
     )
 
+    rate_limit_class = pretend.call_recorder(lambda s: s)
+    rate_limit_iface = pretend.stub()
+    monkeypatch.setattr(manage, "RateLimit", rate_limit_class)
+    monkeypatch.setattr(manage, "IRateLimiter", rate_limit_iface)
+
     manage.includeme(config)
 
     assert config.add_view_deriver.calls == [
         pretend.call(manage.reauth_view, over="rendered_view", under="decorated_view")
     ]
+    assert config.register_service_factory.calls == [
+        pretend.call(
+            "10 per day", rate_limit_iface, name="user_oidc.provider.register"
+        ),
+        pretend.call("100 per day", rate_limit_iface, name="ip_oidc.provider.register"),
+    ]
+    assert config.registry.settings.get.calls == [
+        pretend.call("warehouse.manage.oidc.user_registration_ratelimit_string"),
+        pretend.call("warehouse.manage.oidc.ip_registration_ratelimit_string"),
+    ]