update URLs constructed by the backend to support IRM plugin (#5137)

# What this PR does

Introduces a new class,
`apps.grafana_plugin.ui_url_builder.UIURLBuilder`, which is responsible
for... building UI URLs (😄). The class mainly does two things:
- it will decide if the URL should point to `grafana-oncall-app` or
`grafana-irm-app` based on the value of
`organization.is_grafana_irm_enabled` (**NOTE**: this value isn't yet
being set + defaults to `False`; logic for setting this value will be
done in a subsequent PR)
- Adds `enum`s, `OnCallPage` and `IncidentPage` to DRYify hardcoded UI
URLs (in case we decide to change these slightly in the near future)

## Checklist

- [x] Unit, integration, and e2e (if applicable) tests updated
- [x] Documentation added (or `pr:no public docs` PR label added if not
required)
- [x] Added the relevant release notes label (see labels prefixed w/
`release:`). These labels dictate how your PR will
    show up in the autogenerated release notes.

engine/apps/alerts/models/alert_group.py

@@ -4,7 +4,6 @@
 import urllib
 from collections import namedtuple
 from functools import partial
-from urllib.parse import urljoin
 
 from celery import uuid as celery_uuid
 from django.conf import settings
@@ -27,10 +26,10 @@
  send_alert_group_signal_for_delete,
  unsilence_task,
 )
+from apps.grafana_plugin.ui_url_builder import UIURLBuilder
 from apps.metrics_exporter.tasks import update_metrics_for_alert_group
 from apps.slack.slack_formatter import SlackFormatter
 from apps.user_management.models import User
-from common.constants.plugin_ids import PluginID
 from common.public_primary_keys import generate_public_primary_key, increase_public_primary_key_length
 from common.utils import clean_markup, str_or_backup
 
@@ -543,17 +542,19 @@ def permalinks(self) -> Permalinks:
 
  @property
  def web_link(self) -> str:
- return urljoin(self.channel.organization.web_link, f"alert-groups/{self.public_primary_key}")
+ return UIURLBuilder(self.channel.organization).alert_group_detail(self.public_primary_key)
 
  @property
  def declare_incident_link(self) -> str:
- """Generate a link for AlertGroup to declare Grafana Incident by click"""
- incident_link = urljoin(self.channel.organization.grafana_url, f"a/{PluginID.INCIDENT}/incidents/declare/")
+ """
+ Generate a link for AlertGroup to declare Grafana Incident by click
+ """
  caption = urllib.parse.quote_plus("OnCall Alert Group")
  title = urllib.parse.quote_plus(self.web_title_cache) if self.web_title_cache else DEFAULT_BACKUP_TITLE
  title = title[:2000] # set max title length to avoid exceptions with too long declare incident link
  link = urllib.parse.quote_plus(self.web_link)
- return urljoin(incident_link, f"?caption={caption}&url={link}&title={title}")
+
+ return UIURLBuilder(self.channel.organization).declare_incident(f"?caption={caption}&url={link}&title={title}")
 
  @property
  def happened_while_maintenance(self):

engine/apps/alerts/models/alert_receive_channel.py

@@ -1,7 +1,6 @@
 import logging
 import typing
 from functools import cached_property
-from urllib.parse import urljoin
 
 import emoji
 from celery import uuid as celery_uuid
@@ -21,6 +20,7 @@
 from apps.alerts.tasks import disable_maintenance, disconnect_integration_from_alerting_contact_points
 from apps.base.messaging import get_messaging_backend_from_id
 from apps.base.utils import live_settings
+from apps.grafana_plugin.ui_url_builder import UIURLBuilder
 from apps.integrations.legacy_prefix import remove_legacy_prefix
 from apps.integrations.metadata import heartbeat
 from apps.integrations.tasks import create_alert, create_alertmanager_alerts
@@ -422,8 +422,8 @@ def emojized_verbal_name(self):
 
  @property
  def new_incidents_web_link(self):
- return urljoin(
- self.organization.web_link, f"?page=incidents&integration={self.public_primary_key}&status=0&p=1"
+ return UIURLBuilder(self.organization).alert_groups(
+ f"?integration={self.public_primary_key}&status={AlertGroup.NEW}",
  )
 
  @property
@@ -531,8 +531,8 @@ def created_name(self):
  return f"{self.get_integration_display()} {self.smile_code}"
 
  @property
- def web_link(self):
- return urljoin(self.organization.web_link, f"integrations/{self.public_primary_key}")
+ def web_link(self) -> str:
+ return UIURLBuilder(self.organization).integration_detail(self.public_primary_key)
 
  @property
  def integration_url(self) -> str | None:

engine/apps/api/serializers/user.py

@@ -181,7 +181,7 @@ def get_cloud_connection_status(self, obj: User) -> CloudSyncStatus | None:
  connector = self.context.get("connector", None)
  identities = self.context.get("cloud_identities", {})
  identity = identities.get(obj.email, None)
- status, _ = cloud_user_identity_status(connector, identity)
+ status, _ = cloud_user_identity_status(obj.organization, connector, identity)
  return status
  return None
 

engine/apps/api/tests/test_auth.py

@@ -10,15 +10,18 @@
 
 from apps.auth_token.constants import SLACK_AUTH_TOKEN_NAME
 from apps.social_auth.backends import SLACK_INSTALLATION_BACKEND
+from common.constants.plugin_ids import PluginID
 from common.constants.slack_auth import SLACK_OAUTH_ACCESS_RESPONSE
 
+GRAFANA_URL = "http://example.com"
+
 
 @pytest.mark.django_db
 @pytest.mark.parametrize(
  "backend_name,expected_url",
  (
- ("slack-login", "/a/grafana-oncall-app/users/me"),
- (SLACK_INSTALLATION_BACKEND, "/a/grafana-oncall-app/chat-ops"),
+ ("slack-login", f"{GRAFANA_URL}/a/{PluginID.ONCALL}/users/me"),
+ (SLACK_INSTALLATION_BACKEND, f"{GRAFANA_URL}/a/{PluginID.ONCALL}/chat-ops"),
  ),
 )
 def test_complete_slack_auth_redirect_ok(
@@ -28,7 +31,7 @@ def test_complete_slack_auth_redirect_ok(
  backend_name,
  expected_url,
 ):
- organization = make_organization()
+ organization = make_organization(grafana_url=GRAFANA_URL)
  admin = make_user_for_organization(organization)
  _, slack_token = make_slack_token_for_user(admin)
 
@@ -181,7 +184,7 @@ def test_google_complete_auth_redirect_ok(
  make_user_for_organization,
  make_google_oauth2_token_for_user,
 ):
- organization = make_organization()
+ organization = make_organization(grafana_url=GRAFANA_URL)
  admin = make_user_for_organization(organization)
  _, google_oauth2_token = make_google_oauth2_token_for_user(admin)
 
@@ -194,7 +197,7 @@ def test_google_complete_auth_redirect_ok(
  response = client.get(url)
 
  assert response.status_code == status.HTTP_302_FOUND
- assert response.url == "/a/grafana-oncall-app/users/me"
+ assert response.url == f"{GRAFANA_URL}/a/{PluginID.ONCALL}/users/me"
 
 
 @pytest.mark.django_db

engine/apps/api/views/auth.py

@@ -1,5 +1,4 @@
 import logging
-from urllib.parse import urljoin
 
 from django.conf import settings
 from django.contrib.auth import REDIRECT_FIELD_NAME
@@ -19,6 +18,7 @@
  get_installation_link_from_chatops_proxy,
  get_slack_oauth_response_from_chatops_proxy,
 )
+from apps.grafana_plugin.ui_url_builder import UIURLBuilder
 from apps.slack.installation import install_slack_integration
 from apps.social_auth.backends import SLACK_INSTALLATION_BACKEND, LoginSlackOAuth2V2
 
@@ -73,13 +73,6 @@ def overridden_login_social_auth(request: Request, backend: str) -> Response:
 @psa("social:complete")
 def overridden_complete_social_auth(request: Request, backend: str, *args, **kwargs) -> Response:
  """Authentication complete view"""
- if isinstance(request.backend, (LoginSlackOAuth2V2, GoogleOAuth2)):
- # if this was a user login/linking account, redirect to profile
- redirect_to = "/a/grafana-oncall-app/users/me"
- else:
- # InstallSlackOAuth2V2 backend
- redirect_to = "/a/grafana-oncall-app/chat-ops"
-
  kwargs.update(
  user=request.user,
  redirect_name=REDIRECT_FIELD_NAME,
@@ -99,8 +92,16 @@ def overridden_complete_social_auth(request: Request, backend: str, *args, **kwa
  return_to = request.backend.strategy.session.get(REDIRECT_FIELD_NAME)
 
  if return_to is None:
- # We build the frontend url using org url since multiple stacks could be connected to one backend.
- return_to = urljoin(request.user.organization.grafana_url, redirect_to)
+ url_builder = UIURLBuilder(request.user.organization)
+
+ # if this was a user login/linking account, redirect to profile (ie. users/me)
+ # otherwise it pertains to the InstallSlackOAuth2V2 backend, and we should redirect to the chat-ops page
+ return_to = (
+ url_builder.user_profile()
+ if isinstance(request.backend, (LoginSlackOAuth2V2, GoogleOAuth2))
+ else url_builder.chatops()
+ )
+
  return HttpResponseRedirect(return_to)
 
 

engine/apps/chatops_proxy/register_oncall_tenant.py

@@ -1,10 +1,15 @@
 # register_oncall_tenant moved to separate file from engine/apps/chatops_proxy/utils.py to avoid circular imports.
+import typing
+
 from django.conf import settings
 
 from apps.chatops_proxy.client import APP_TYPE_ONCALL, ChatopsProxyAPIClient
 
+if typing.TYPE_CHECKING:
+ from apps.user_management.models import Organization
+
 
-def register_oncall_tenant(org):
+def register_oncall_tenant(org: "Organization") -> None:
  """
  register_oncall_tenant registers oncall organization as a tenant in chatops-proxy.
  """

engine/apps/chatops_proxy/utils.py

@@ -3,10 +3,11 @@
 """
 import logging
 import typing
-from urllib.parse import urljoin
 
 from django.conf import settings
 
+from apps.grafana_plugin.ui_url_builder import UIURLBuilder
+
 from .client import APP_TYPE_ONCALL, PROVIDER_TYPE_SLACK, ChatopsProxyAPIClient, ChatopsProxyAPIException
 from .register_oncall_tenant import register_oncall_tenant
 from .tasks import (
@@ -16,10 +17,13 @@
  unregister_oncall_tenant_async,
 )
 
+if typing.TYPE_CHECKING:
+ from apps.user_management.models import Organization, User
+
 logger = logging.getLogger(__name__)
 
 
-def get_installation_link_from_chatops_proxy(user) -> typing.Optional[str]:
+def get_installation_link_from_chatops_proxy(user: "User") -> typing.Optional[str]:
  """
  get_installation_link_from_chatops_proxy fetches slack installation link from chatops proxy.
  If there is no existing slack installation - if returns link, If slack already installed, it returns None.
@@ -30,7 +34,7 @@ def get_installation_link_from_chatops_proxy(user) -> typing.Optional[str]:
  link, _ = client.get_slack_oauth_link(
  org.stack_id,
  user.user_id,
- urljoin(org.web_link, "settings?tab=ChatOps&chatOpsTab=Slack"),
+ UIURLBuilder(org).settings("?tab=ChatOps&chatOpsTab=Slack"),
  APP_TYPE_ONCALL,
  )
  return link
@@ -44,13 +48,13 @@ def get_installation_link_from_chatops_proxy(user) -> typing.Optional[str]:
  raise api_exc
 
 
-def get_slack_oauth_response_from_chatops_proxy(stack_id) -> dict:
+def get_slack_oauth_response_from_chatops_proxy(stack_id: int) -> dict:
  client = ChatopsProxyAPIClient(settings.ONCALL_GATEWAY_URL, settings.ONCALL_GATEWAY_API_TOKEN)
  slack_installation, _ = client.get_oauth_installation(stack_id, PROVIDER_TYPE_SLACK)
  return slack_installation.oauth_response
 
 
-def register_oncall_tenant_with_async_fallback(org):
+def register_oncall_tenant_with_async_fallback(org: "Organization") -> None:
  """
  register_oncall_tenant tries to register oncall tenant synchronously and fall back to task in case of any exceptions
  to make sure that tenant is registered.

engine/apps/grafana_plugin/tests/test_ui_url_builder.py

@@ -0,0 +1,105 @@
+import pytest
+
+from apps.grafana_plugin.ui_url_builder import UIURLBuilder
+from common.constants.plugin_ids import PluginID
+
+GRAFANA_URL = "http://example.com"
+ALERT_GROUP_ID = "1234"
+INTEGRATION_ID = "5678"
+SCHEDULE_ID = "lasdfasdf"
+PATH_EXTRA = "/extra?foo=bar"
+
+
+@pytest.fixture
+def org_setup(make_organization):
+ def _org_setup(is_grafana_irm_enabled=False):
+ return make_organization(grafana_url=GRAFANA_URL, is_grafana_irm_enabled=is_grafana_irm_enabled)
+
+ return _org_setup
+
+
+@pytest.mark.parametrize(
+ "func,call_kwargs,expected_url",
+ [
+ # oncall pages
+ (
+ "home",
+ {"path_extra": PATH_EXTRA},
+ f"{GRAFANA_URL}/a/{PluginID.ONCALL}{PATH_EXTRA}",
+ ),
+ (
+ "alert_groups",
+ {"path_extra": PATH_EXTRA},
+ f"{GRAFANA_URL}/a/{PluginID.ONCALL}/alert-groups{PATH_EXTRA}",
+ ),
+ (
+ "alert_group_detail",
+ {"id": ALERT_GROUP_ID, "path_extra": PATH_EXTRA},
+ f"{GRAFANA_URL}/a/{PluginID.ONCALL}/alert-groups/{ALERT_GROUP_ID}{PATH_EXTRA}",
+ ),
+ (
+ "integration_detail",
+ {"id": INTEGRATION_ID, "path_extra": PATH_EXTRA},
+ f"{GRAFANA_URL}/a/{PluginID.ONCALL}/integrations/{INTEGRATION_ID}{PATH_EXTRA}",
+ ),
+ (
+ "schedules",
+ {"path_extra": PATH_EXTRA},
+ f"{GRAFANA_URL}/a/{PluginID.ONCALL}/schedules{PATH_EXTRA}",
+ ),
+ (
+ "schedule_detail",
+ {"id": SCHEDULE_ID, "path_extra": PATH_EXTRA},
+ f"{GRAFANA_URL}/a/{PluginID.ONCALL}/schedules/{SCHEDULE_ID}{PATH_EXTRA}",
+ ),
+ (
+ "users",
+ {"path_extra": PATH_EXTRA},
+ f"{GRAFANA_URL}/a/{PluginID.ONCALL}/users{PATH_EXTRA}",
+ ),
+ (
+ "user_profile",
+ {"path_extra": PATH_EXTRA},
+ f"{GRAFANA_URL}/a/{PluginID.ONCALL}/users/me{PATH_EXTRA}",
+ ),
+ (
+ "chatops",
+ {"path_extra": PATH_EXTRA},
+ f"{GRAFANA_URL}/a/{PluginID.ONCALL}/chat-ops{PATH_EXTRA}",
+ ),
+ (
+ "settings",
+ {"path_extra": PATH_EXTRA},
+ f"{GRAFANA_URL}/a/{PluginID.ONCALL}/settings{PATH_EXTRA}",
+ ),
+ # incident pages
+ (
+ "declare_incident",
+ {"path_extra": "?caption=abcd&url=asdf&title=test1234"},
+ f"{GRAFANA_URL}/a/{PluginID.INCIDENT}/incidents/declare?caption=abcd&url=asdf&title=test1234",
+ ),
+ ],
+)
+@pytest.mark.django_db
+def test_build_page_urls(org_setup, func, call_kwargs, expected_url):
+ builder = UIURLBuilder(org_setup())
+ assert getattr(builder, func)(**call_kwargs) == expected_url
+
+
+@pytest.mark.django_db
+def test_build_url_overriden_base_url(org_setup):
+ overriden_base_url = "http://overriden.com"
+ builder = UIURLBuilder(org_setup(), base_url=overriden_base_url)
+ assert builder.chatops() == f"{overriden_base_url}/a/{PluginID.ONCALL}/chat-ops"
+
+
+@pytest.mark.parametrize(
+ "is_grafana_irm_enabled,expected_url",
+ [
+ (True, f"{GRAFANA_URL}/a/{PluginID.IRM}/alert-groups/{ALERT_GROUP_ID}"),
+ (False, f"{GRAFANA_URL}/a/{PluginID.ONCALL}/alert-groups/{ALERT_GROUP_ID}"),
+ ],
+)
+@pytest.mark.django_db
+def test_build_url_works_for_irm_and_oncall_plugins(org_setup, is_grafana_irm_enabled, expected_url):
+ assert UIURLBuilder(org_setup(is_grafana_irm_enabled)).alert_group_detail(ALERT_GROUP_ID) == expected_url