sadpandajoe commented on code in PR #45508: URL: https://github.com/apache/superset/pull/45508#discussion_r4374074467
########## tests/unit_tests/integrations/teams/delivery_test.py: ########## @@ -0,0 +1,965 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Teams notification formatting, resolution, retry, and SDK delivery tests.""" + +from __future__ import annotations + +import asyncio +import importlib +import importlib.util +import threading +import time +from collections.abc import AsyncIterator, Iterator +from contextlib import asynccontextmanager +from datetime import datetime, timedelta, timezone +from email.utils import format_datetime +from types import SimpleNamespace +from typing import Any +from unittest.mock import MagicMock + +import httpx +import pytest +import requests +from billiard.exceptions import SoftTimeLimitExceeded +from flask import current_app, Flask +from sqlalchemy.engine import Engine +from sqlalchemy.orm import Session + +from superset.integrations.teams import validation as teams_validation +from superset.integrations.teams.config import ( + parse_teams_config, + TeamsConfig, + TeamsDestination, +) +from superset.integrations.teams.models import TeamsInstallation +from superset.integrations.teams.store import TeamsInstallationStore +from superset.reports.models import ReportRecipients, ReportRecipientType +from superset.reports.notifications import create_notification, teams as notifications +from superset.reports.notifications.base import NotificationContent +from superset.reports.notifications.exceptions import ( + NotificationAuthorizationException, + NotificationError, + NotificationParamException, + NotificationTransientError, +) +from superset.utils import json + +if importlib.util.find_spec("microsoft_teams") is None: + pytest.skip("Optional Teams SDK is not installed", allow_module_level=True) + +adapter = importlib.import_module("superset.integrations.teams.adapter") +TENANT = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa" +APPLICATION = "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb" +SERVICE_URL = "https://smba.trafficmanager.net/amer/"; + + [email protected] +def configured(teams_config: dict[str, Any]) -> TeamsConfig: + """Parse the same deployment configuration used by real notifications.""" + config = parse_teams_config(teams_config) + assert config is not None + return config + + [email protected] +def content() -> NotificationContent: + """Include difficult text to verify normal structured JSON serialization.""" + return NotificationContent( + name='Alert "Revenue"\nCafé 東京 🚨', + description='Description "quoted"\n第二行', + header_data={ + "notification_type": "Alert", + "notification_format": "NONE", + "editors": [], + "notification_source": None, + "chart_id": None, + "dashboard_id": None, + "slack_channels": None, + "execution_id": None, + }, + url="https://superset.example/chart/42/";, + ) + + [email protected] +def notification_context( + teams_config: dict[str, Any], + teams_engine: Engine, + monkeypatch: pytest.MonkeyPatch, +) -> Iterator[SimpleNamespace]: + """Use the real installation table and resolve through the notification class.""" + app = Flask(__name__) + app.config.update( + ALERT_REPORTS_TEAMS=teams_config, FEATURE_FLAGS={"ALERT_REPORT_TEAMS": True} + ) + engine = teams_engine + with Session(engine) as session: + TeamsInstallationStore(session).install( + TENANT, APPLICATION, "team-1", SERVICE_URL, datetime.now(timezone.utc) + ) + session.commit() + monkeypatch.setattr(notifications, "db", SimpleNamespace(engine=engine)) + monkeypatch.setattr( + teams_validation.feature_flag_manager, + "is_feature_enabled", + lambda name: current_app.config["FEATURE_FLAGS"].get(name, False), + ) + send = MagicMock() + monkeypatch.setattr(notifications, "_send_to_teams", send) + with app.app_context(): + yield SimpleNamespace(app=app, engine=engine, send=send) + + +def recipient(target: str = "ops") -> ReportRecipients: + """Recipients store only a configured destination key.""" + return ReportRecipients( + type=ReportRecipientType.TEAMS, + recipient_config_json=json.dumps({"target": target}), + ) + + +def test_factory_resolves_active_installation_and_formats_card( + notification_context: SimpleNamespace, + content: NotificationContent, +) -> None: + notification = create_notification(recipient(), content) + assert isinstance(notification, notifications.TeamsNotification) + notification.send() + config, destination, service_url, card = notification_context.send.call_args.args + assert config.credential == "deployment-secret" + assert destination.key == "ops" + assert destination.channel_id == "channel-1" + assert service_url == SERVICE_URL + assert card["body"][0]["text"] == content.name + assert card["body"][1]["text"] == content.description + assert card["actions"][0]["url"] == content.url + assert "deployment-secret" not in json.dumps(card) + + [email protected]( + "mode", ["inactive", "missing", "other-tenant", "other-application"] +) +def test_installation_must_be_active_for_exact_identity( + notification_context: SimpleNamespace, + content: NotificationContent, + mode: str, +) -> None: + with Session(notification_context.engine) as session: + store = TeamsInstallationStore(session) + if mode == "inactive": + store.remove( + TENANT, APPLICATION, "team-1", SERVICE_URL, datetime.now(timezone.utc) + ) + else: + session.query(TeamsInstallation).delete() + other = "cccccccc-cccc-4ccc-8ccc-cccccccccccc" + if mode != "missing": + store.install( + other if mode == "other-tenant" else TENANT, + other if mode == "other-application" else APPLICATION, + "team-1", + SERVICE_URL, + datetime.now(timezone.utc), + ) + session.commit() + with pytest.raises(NotificationAuthorizationException): + notifications.TeamsNotification(recipient(), content).send() + notification_context.send.assert_not_called() + + +class FakeTransport: + """Deterministic replaceable transport for bounded retry policy tests.""" + + def __init__(self, errors: list[Exception | None]) -> None: + """Consume one outcome for each root post attempt.""" + self.errors = iter(errors) + self.posts: list[tuple[TeamsDestination, dict[str, Any]]] = [] + self.refreshes = 0 + self.closed = False + self.loop: asyncio.AbstractEventLoop | None = None + + async def post_root( + self, destination: TeamsDestination, card: dict[str, Any] + ) -> None: + """Record every fresh root-post attempt.""" + self.posts.append((destination, card)) + error = next(self.errors) + if error: + raise error + + async def refresh_auth(self) -> None: + """Record a cache invalidation without contacting Entra.""" + self.refreshes += 1 + + @asynccontextmanager + async def factory( + self, config: TeamsConfig, service_url: str, deadline: float + ) -> AsyncIterator[FakeTransport]: + """Track operation cleanup on both successful and failed delivery.""" + self.loop = asyncio.get_running_loop() + try: + yield self + finally: + self.closed = True + + [email protected] +def retry_clock(monkeypatch: pytest.MonkeyPatch) -> list[float]: + """Advance retry time without actually sleeping.""" + elapsed = [0.0] + sleeps: list[float] = [] + monkeypatch.setattr(adapter, "monotonic", lambda: elapsed[0]) + + async def sleep(delay: float) -> None: + """Advance the same clock used by the retry budget.""" + elapsed[0] += delay + sleeps.append(delay) + + monkeypatch.setattr(adapter.asyncio, "sleep", sleep) + return sleeps + + [email protected]( + "status,retryable", + [ + (None, False), + (None, True), + (408, False), + (429, False), + (500, False), + (502, False), + (503, False), + (504, False), + ], +) +def test_transport_failures_respect_retry_safety_and_attempt_bound( + configured: TeamsConfig, + retry_clock: list[float], + status: int | None, + retryable: bool, +) -> None: + transport = FakeTransport( + [adapter.TeamsTransportError(status, retryable=retryable)] * 4 + ) + with pytest.raises(adapter.TeamsDeliveryError) as exc: + adapter.deliver_teams_alert( + configured, + configured.destinations[0], + SERVICE_URL, + {}, + transport_factory=transport.factory, + ) + ambiguous = status is None and not retryable + assert exc.value.category == ("unknown" if ambiguous else "transient") + assert len(transport.posts) == (1 if ambiguous else 4) + assert retry_clock == ([] if ambiguous else [1.0, 2.0, 4.0]) + assert transport.closed + assert transport.loop is not None + assert transport.loop.is_closed() + + [email protected]( + "headers,expected", + [ + ({"Retry-After": "3"}, 3.0), + ({"x-ms-retry-after-ms": "1500"}, 1.5), + ({"Retry-After": "2", "x-ms-retry-after-ms": "3000"}, 3.0), + ({"Retry-After": "invalid"}, 1.0), + ], +) +def test_microsoft_retry_delays_are_respected( + configured: TeamsConfig, + retry_clock: list[float], + headers: dict[str, str], + expected: float, +) -> None: + transport = FakeTransport( + [adapter.TeamsTransportError(429, httpx.Headers(headers)), None] + ) + adapter.deliver_teams_alert( + configured, + configured.destinations[0], + SERVICE_URL, + {}, + transport_factory=transport.factory, + ) + assert retry_clock == [expected] + assert len(transport.posts) == 2 + assert transport.closed + + +def test_retry_after_http_date() -> None: + retry_at = datetime.now(timezone.utc) + timedelta(seconds=5) + error = adapter.TeamsTransportError( + 429, httpx.Headers({"Retry-After": format_datetime(retry_at, usegmt=True)}) + ) + assert 3.0 <= adapter._retry_delay(error, 0) <= 5.0 + + +def test_excessive_server_delay_is_not_shortened( + configured: TeamsConfig, retry_clock: list[float] +) -> None: + transport = FakeTransport( + [adapter.TeamsTransportError(429, httpx.Headers({"Retry-After": "120"}))] + ) + with pytest.raises(adapter.TeamsDeliveryError) as exc: + adapter.deliver_teams_alert( + configured, + configured.destinations[0], + SERVICE_URL, + {}, + transport_factory=transport.factory, + ) + assert exc.value.category == "transient" + assert len(transport.posts) == 1 + assert retry_clock == [] + assert transport.closed + + [email protected]( + "outcomes,category,posts,refreshes", + [ + ([401, None], None, 2, 1), + ([401, 401], "authorization", 2, 1), + ([403], "authorization", 1, 0), + ([400], "parameter", 1, 0), + ([404], "parameter", 1, 0), + ([302], "parameter", 1, 0), + ([401, 503, 503, None], None, 4, 1), Review Comment: If the fourth send is the first to return `401`, the `attempt == _DELIVERY_ATTEMPTS - 1` guard in `_deliver_with_retries` is what makes delivery fail as `authorization` with no token refresh. Every case here hits the `401` on attempt one or two, so removing that guard would still pass: the fourth-attempt `401` would refresh the token and then fall out of the loop as `transient`. Could you add a `[503, 503, 503, 401]` case that asserts four posts, zero refreshes, and category `authorization`? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
