chore: refactoring and upgrade telegram/urllib3 deps (#5611)

Author: matiasb

engine/apps/alerts/tests/test_related_incident.py

@@ -1,7 +1,7 @@
 from unittest.mock import patch
 
-import httpretty
 import pytest
+import responses
 
 from apps.alerts.models import AlertGroupLogRecord, EscalationPolicy, RelatedIncident
 from apps.alerts.tasks.declare_incident import (
@@ -72,7 +72,7 @@ def test_declare_incident_default_route(setup_alert_group_and_escalation_step):
 
 
 @pytest.mark.django_db
-@httpretty.activate(verbose=True, allow_net_connect=False)
+@responses.activate
 def test_declare_incident_ok(setup_alert_group_and_escalation_step):
     alert_group, declare_incident_step, _ = setup_alert_group_and_escalation_step(already_declared_incident=False)
 
@@ -105,7 +105,7 @@ def test_declare_incident_ok(setup_alert_group_and_escalation_step):
 
 
 @pytest.mark.django_db
-@httpretty.activate(verbose=True, allow_net_connect=False)
+@responses.activate
 def test_declare_incident_set_severity(setup_alert_group_and_escalation_step):
     alert_group, declare_incident_step, _ = setup_alert_group_and_escalation_step(already_declared_incident=False)
     severity = "critical"
@@ -120,7 +120,7 @@ def test_declare_incident_set_severity(setup_alert_group_and_escalation_step):
 
 
 @pytest.mark.django_db
-@httpretty.activate(verbose=True, allow_net_connect=False)
+@responses.activate
 def test_declare_incident_set_severity_from_label(setup_alert_group_and_escalation_step):
     alert_group, declare_incident_step, _ = setup_alert_group_and_escalation_step(already_declared_incident=False)
     expected_severity = "minor"
@@ -143,7 +143,7 @@ def test_declare_incident_set_severity_from_label(setup_alert_group_and_escalati
 
 
 @pytest.mark.django_db
-@httpretty.activate(verbose=True, allow_net_connect=False)
+@responses.activate
 def test_declare_incident_invalid_severity_fallback(setup_alert_group_and_escalation_step):
     alert_group, declare_incident_step, _ = setup_alert_group_and_escalation_step(already_declared_incident=False)
     severity = "INVALID"
@@ -166,7 +166,7 @@ def test_declare_incident_invalid_severity_fallback(setup_alert_group_and_escala
 
 
 @pytest.mark.django_db
-@httpretty.activate(verbose=True, allow_net_connect=False)
+@responses.activate
 def test_declare_incident_attach_alert_group(setup_alert_group_and_escalation_step):
     alert_group, declare_incident_step, existing_open_incident = setup_alert_group_and_escalation_step(
         already_declared_incident=True
@@ -191,7 +191,7 @@ def test_declare_incident_attach_alert_group(setup_alert_group_and_escalation_st
 
 
 @pytest.mark.django_db
-@httpretty.activate(verbose=True, allow_net_connect=False)
+@responses.activate
 def test_declare_incident_resolved_update(setup_alert_group_and_escalation_step):
     alert_group, declare_incident_step, existing_open_incident = setup_alert_group_and_escalation_step(
         already_declared_incident=True
@@ -226,7 +226,7 @@ def test_declare_incident_resolved_update(setup_alert_group_and_escalation_step)
 
 
 @pytest.mark.django_db
-@httpretty.activate(verbose=True, allow_net_connect=False)
+@responses.activate
 def test_declare_incident_attach_alert_group_skip_incident_update(
     setup_alert_group_and_escalation_step, make_alert_group
 ):
@@ -261,7 +261,7 @@ def test_declare_incident_attach_alert_group_skip_incident_update(
 
 
 @pytest.mark.django_db
-@httpretty.activate(verbose=True, allow_net_connect=False)
+@responses.activate
 def test_get_existing_incident_error(setup_alert_group_and_escalation_step):
     alert_group, declare_incident_step, existing_open_incident = setup_alert_group_and_escalation_step(
         already_declared_incident=True
@@ -295,7 +295,7 @@ def test_get_existing_incident_error(setup_alert_group_and_escalation_step):
 
 
 @pytest.mark.django_db
-@httpretty.activate(verbose=True, allow_net_connect=False)
+@responses.activate
 def test_attach_alert_group_error(setup_alert_group_and_escalation_step):
     alert_group, declare_incident_step, existing_open_incident = setup_alert_group_and_escalation_step(
         already_declared_incident=True
@@ -322,7 +322,7 @@ def test_attach_alert_group_error(setup_alert_group_and_escalation_step):
 
 
 @pytest.mark.django_db
-@httpretty.activate(verbose=True, allow_net_connect=False)
+@responses.activate
 def test_create_incident_error(setup_alert_group_and_escalation_step):
     alert_group, declare_incident_step, _ = setup_alert_group_and_escalation_step(already_declared_incident=False)
 

engine/apps/api/tests/test_organization.py

@@ -199,7 +199,8 @@ def test_organization_get_telegram_verification_code_permissions(
     client = APIClient()
 
     url = reverse("api-internal:api-get-telegram-verification-code")
-    response = client.get(url, format="json", **make_user_auth_headers(tester, token))
+    with patch("apps.telegram.client.TelegramClient.get_bot_username", return_value="test_bot_username"):
+        response = client.get(url, format="json", **make_user_auth_headers(tester, token))
 
     assert response.status_code == expected_status
 

engine/apps/api/tests/test_user.py

@@ -808,7 +808,8 @@ def test_user_get_own_telegram_verification_code_with_telegram_connected(
     client = APIClient()
     make_telegram_user_connector(tester)
     url = reverse("api-internal:user-get-telegram-verification-code", kwargs={"pk": tester.public_primary_key})
-    response = client.get(url, format="json", **make_user_auth_headers(tester, token))
+    with patch("apps.telegram.client.TelegramClient.get_bot_username", return_value="test_bot_username"):
+        response = client.get(url, format="json", **make_user_auth_headers(tester, token))
     assert response.status_code == expected_status
 
 
@@ -867,7 +868,8 @@ def test_user_get_own_telegram_verification_code(
     client = APIClient()
     url = reverse("api-internal:user-get-telegram-verification-code", kwargs={"pk": tester.public_primary_key})
 
-    response = client.get(url, format="json", **make_user_auth_headers(tester, token))
+    with patch("apps.telegram.client.TelegramClient.get_bot_username", return_value="test_bot_username"):
+        response = client.get(url, format="json", **make_user_auth_headers(tester, token))
     assert response.status_code == expected_status
 
 
@@ -894,7 +896,8 @@ def test_user_get_another_telegram_verification_code(
     client = APIClient()
     url = reverse("api-internal:user-get-telegram-verification-code", kwargs={"pk": other_user.public_primary_key})
 
-    response = client.get(url, format="json", **make_user_auth_headers(tester, token))
+    with patch("apps.telegram.client.TelegramClient.get_bot_username", return_value="test_bot_username"):
+        response = client.get(url, format="json", **make_user_auth_headers(tester, token))
     assert response.status_code == expected_status
 
 
@@ -1042,7 +1045,8 @@ def test_admin_can_get_own_telegram_verification_code(
     client = APIClient()
     url = reverse("api-internal:user-get-telegram-verification-code", kwargs={"pk": user.public_primary_key})
 
-    response = client.get(url, format="json", **make_user_auth_headers(user, token))
+    with patch("apps.telegram.client.TelegramClient.get_bot_username", return_value="test_bot_username"):
+        response = client.get(url, format="json", **make_user_auth_headers(user, token))
     assert response.status_code == status.HTTP_200_OK
 
 
@@ -1058,7 +1062,8 @@ def test_admin_can_get_another_user_telegram_verification_code(
     client = APIClient()
     url = reverse("api-internal:user-get-telegram-verification-code", kwargs={"pk": first_user.public_primary_key})
 
-    response = client.get(url, format="json", **make_user_auth_headers(second_user, token))
+    with patch("apps.telegram.client.TelegramClient.get_bot_username", return_value="test_bot_username"):
+        response = client.get(url, format="json", **make_user_auth_headers(second_user, token))
     assert response.status_code == status.HTTP_200_OK
 
 
@@ -1280,7 +1285,8 @@ def test_user_can_get_own_telegram_verification_code(
     client = APIClient()
     url = reverse("api-internal:user-get-telegram-verification-code", kwargs={"pk": user.public_primary_key})
 
-    response = client.get(f"{url}", format="json", **make_user_auth_headers(user, token))
+    with patch("apps.telegram.client.TelegramClient.get_bot_username", return_value="test_bot_username"):
+        response = client.get(f"{url}", format="json", **make_user_auth_headers(user, token))
     assert response.status_code == status.HTTP_200_OK
 
 
@@ -1298,7 +1304,8 @@ def test_user_cant_get_another_user_telegram_verification_code(
     client = APIClient()
     url = reverse("api-internal:user-get-telegram-verification-code", kwargs={"pk": first_user.public_primary_key})
 
-    response = client.get(f"{url}", format="json", **make_user_auth_headers(second_user, token))
+    with patch("apps.telegram.client.TelegramClient.get_bot_username", return_value="test_bot_username"):
+        response = client.get(f"{url}", format="json", **make_user_auth_headers(second_user, token))
     assert response.status_code == status.HTTP_403_FORBIDDEN
 
 
@@ -1584,7 +1591,8 @@ def test_viewer_cant_get_own_telegram_verification_code(
     client = APIClient()
     url = reverse("api-internal:user-get-telegram-verification-code", kwargs={"pk": user.public_primary_key})
 
-    response = client.get(f"{url}", format="json", **make_user_auth_headers(user, token))
+    with patch("apps.telegram.client.TelegramClient.get_bot_username", return_value="test_bot_username"):
+        response = client.get(f"{url}", format="json", **make_user_auth_headers(user, token))
     assert response.status_code == status.HTTP_403_FORBIDDEN
 
 
@@ -1600,7 +1608,8 @@ def test_viewer_cant_get_another_user_telegram_verification_code(
     client = APIClient()
     url = reverse("api-internal:user-get-telegram-verification-code", kwargs={"pk": first_user.public_primary_key})
 
-    response = client.get(url, format="json", **make_user_auth_headers(second_user, token))
+    with patch("apps.telegram.client.TelegramClient.get_bot_username", return_value="test_bot_username"):
+        response = client.get(url, format="json", **make_user_auth_headers(second_user, token))
     assert response.status_code == status.HTTP_403_FORBIDDEN
 
 

engine/apps/api/views/organization.py

@@ -82,7 +82,7 @@ def get(self, request):
             existing_verification_code.delete()
         new_code = TelegramChannelVerificationCode.objects.create(organization=organization, author=user)
         telegram_client = TelegramClient()
-        bot_username = telegram_client.api_client.username
+        bot_username = telegram_client.get_bot_username()
         bot_link = f"https://t.me/{bot_username}"
         return Response(
             {"telegram_code": str(new_code.uuid_with_org_uuid), "bot_link": bot_link}, status=status.HTTP_200_OK

engine/apps/api/views/user.py

@@ -667,7 +667,7 @@ def get_telegram_verification_code(self, request, pk) -> Response:
         new_code.save()
 
         telegram_client = TelegramClient()
-        bot_username = telegram_client.api_client.username
+        bot_username = telegram_client.get_bot_username()
         bot_link = f"https://t.me/{bot_username}"
 
         return Response(

engine/apps/auth_token/tests/helpers.py

@@ -1,18 +1,12 @@
-import json
-
-import httpretty
+import responses
 
 
 def setup_service_account_api_mocks(grafana_url, perms=None, user_data=None, perms_status=200, user_status=200):
-    # requires enabling httpretty
+    # requires enabling responses
     if perms is None:
         perms = {}
-    mock_response = httpretty.Response(status=perms_status, body=json.dumps(perms))
-    perms_url = f"{grafana_url}/api/access-control/user/permissions"
-    httpretty.register_uri(httpretty.GET, perms_url, responses=[mock_response])
+    responses.add(responses.GET, f"{grafana_url}/api/access-control/user/permissions", json=perms, status=perms_status)
 
     if user_data is None:
         user_data = {"login": "some-login", "uid": "service-account:42"}
-    mock_response = httpretty.Response(status=user_status, body=json.dumps(user_data))
-    user_url = f"{grafana_url}/api/user"
-    httpretty.register_uri(httpretty.GET, user_url, responses=[mock_response])
+    responses.add(responses.GET, f"{grafana_url}/api/user", json=user_data, status=user_status)

engine/apps/auth_token/tests/test_grafana_auth.py

@@ -1,8 +1,8 @@
 import typing
 from unittest.mock import patch
 
-import httpretty
 import pytest
+import responses
 from rest_framework import exceptions
 from rest_framework.test import APIRequestFactory
 
@@ -83,7 +83,7 @@ def check_common_inputs() -> tuple[dict[str, typing.Any], str]:
 
 
 @pytest.mark.django_db
-@httpretty.activate(verbose=True, allow_net_connect=False)
+@responses.activate
 def test_grafana_authentication_missing_org():
     token = f"{ServiceAccountToken.GRAFANA_SA_PREFIX}xyz"
     headers = {
@@ -97,7 +97,7 @@ def test_grafana_authentication_missing_org():
 
 
 @pytest.mark.django_db
-@httpretty.activate(verbose=True, allow_net_connect=False)
+@responses.activate
 def test_grafana_authentication_no_org_grafana_url():
     grafana_url = "http://grafana.test"
     token = f"{ServiceAccountToken.GRAFANA_SA_PREFIX}xyz"
@@ -108,7 +108,7 @@ def test_grafana_authentication_no_org_grafana_url():
     request = APIRequestFactory().get("/", **headers)
 
     request_sync_url = f"{grafana_url}/api/plugins/{PluginID.ONCALL}/resources/plugin/sync?wait=true&force=true"
-    httpretty.register_uri(httpretty.POST, request_sync_url, status=404)
+    responses.add(responses.POST, request_sync_url, status=404)
 
     with pytest.raises(exceptions.Throttled) as exc:
         GrafanaServiceAccountAuthentication().authenticate(request)
@@ -117,7 +117,7 @@ def test_grafana_authentication_no_org_grafana_url():
 
 @pytest.mark.parametrize("grafana_url", ["null;", "foo", ""])
 @pytest.mark.django_db
-@httpretty.activate(verbose=True, allow_net_connect=False)
+@responses.activate
 def test_grafana_authentication_invalid_grafana_url(grafana_url):
     token = f"{ServiceAccountToken.GRAFANA_SA_PREFIX}xyz"
     headers = {
@@ -133,7 +133,7 @@ def test_grafana_authentication_invalid_grafana_url(grafana_url):
 
 
 @pytest.mark.django_db
-@httpretty.activate(verbose=True, allow_net_connect=False)
+@responses.activate
 def test_grafana_authentication_permissions_call_fails(make_organization):
     organization = make_organization(grafana_url="http://grafana.test")
 
@@ -152,7 +152,7 @@ def test_grafana_authentication_permissions_call_fails(make_organization):
         GrafanaServiceAccountAuthentication().authenticate(request)
     assert exc.value.detail == "Invalid token."
 
-    last_request = httpretty.last_request()
+    last_request = responses.calls[-1].request
     assert last_request.method == "GET"
     expected_url = f"{organization.grafana_url}/api/access-control/user/permissions"
     assert last_request.url == expected_url
@@ -162,7 +162,7 @@ def test_grafana_authentication_permissions_call_fails(make_organization):
 
 @pytest.mark.django_db
 @pytest.mark.parametrize("grafana_url", ["http://grafana.test", "http://grafana.test/"])
-@httpretty.activate(verbose=True, allow_net_connect=False)
+@responses.activate
 def test_grafana_authentication_existing_token(
     make_organization, make_service_account_for_organization, make_token_for_service_account, grafana_url
 ):
@@ -190,7 +190,7 @@ def test_grafana_authentication_existing_token(
     assert user.role == LegacyAccessControlRole.NONE
     assert auth_token == token
 
-    last_request = httpretty.last_request()
+    last_request = responses.calls[-1].request
     assert last_request.method == "GET"
     expected_url = f"{organization.grafana_url}/api/access-control/user/permissions"
     assert last_request.url == expected_url
@@ -199,7 +199,7 @@ def test_grafana_authentication_existing_token(
 
 
 @pytest.mark.django_db
-@httpretty.activate(verbose=True, allow_net_connect=False)
+@responses.activate
 def test_grafana_authentication_token_created(make_organization):
     organization = make_organization(grafana_url="http://grafana.test")
     token_string = "glsa_the-token"
@@ -228,18 +228,18 @@ def test_grafana_authentication_token_created(make_organization):
     assert user.permissions == [{"action": p} for p in permissions]
     assert auth_token.service_account == user.service_account
 
-    perms_request, user_request = httpretty.latest_requests()
+    perms_request, user_request = responses.calls
     for req in (perms_request, user_request):
-        assert req.method == "GET"
-        assert req.headers["Authorization"] == f"Bearer {token_string}"
+        assert req.request.method == "GET"
+        assert req.request.headers["Authorization"] == f"Bearer {token_string}"
     perms_url = f"{organization.grafana_url}/api/access-control/user/permissions"
-    assert perms_request.url == perms_url
+    assert perms_request.request.url == perms_url
     user_url = f"{organization.grafana_url}/api/user"
-    assert user_request.url == user_url
+    assert user_request.request.url == user_url
 
 
 @pytest.mark.django_db
-@httpretty.activate(verbose=True, allow_net_connect=False)
+@responses.activate
 def test_grafana_authentication_token_created_older_grafana(make_organization):
     organization = make_organization(grafana_url="http://grafana.test")
     token_string = "glsa_the-token"
@@ -267,7 +267,7 @@ def test_grafana_authentication_token_created_older_grafana(make_organization):
 
 
 @pytest.mark.django_db
-@httpretty.activate(verbose=True, allow_net_connect=False)
+@responses.activate
 def test_grafana_authentication_token_reuse_service_account(make_organization, make_service_account_for_organization):
     organization = make_organization(grafana_url="http://grafana.test")
     service_account = make_service_account_for_organization(organization)
@@ -295,7 +295,7 @@ def test_grafana_authentication_token_reuse_service_account(make_organization, m
 
 
 @pytest.mark.django_db
-@httpretty.activate(verbose=True, allow_net_connect=False)
+@responses.activate
 def test_grafana_authentication_token_setup_org_if_missing(make_organization):
     grafana_url = "http://grafana.test"
     token_string = "glsa_the-token"
@@ -311,7 +311,7 @@ def test_grafana_authentication_token_setup_org_if_missing(make_organization):
     setup_service_account_api_mocks(grafana_url, permissions)
 
     request_sync_url = f"{grafana_url}/api/plugins/{PluginID.ONCALL}/resources/plugin/sync?wait=true&force=true"
-    httpretty.register_uri(httpretty.POST, request_sync_url)
+    responses.add(responses.POST, request_sync_url)
 
     assert Organization.objects.filter(grafana_url=grafana_url).count() == 0