open-metadata--openmetadata/ingestion/tests/integration/ometa/test_ometa_subscription_api.py

#  Copyright 2025 Collate
#  Licensed under the Collate Community License, Version 1.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#  https://github.com/open-metadata/OpenMetadata/blob/main/ingestion/LICENSE
#  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.

"""
OpenMetadata high-level API EventSubscription test
"""

from copy import deepcopy
from unittest.mock import patch

import pytest
from pydantic import ValidationError

from metadata.generated.schema.api.teams.createUser import CreateUserRequest
from metadata.generated.schema.entity.teams.user import User
from metadata.generated.schema.events.api.createEventSubscription import (
    CreateEventSubscription,
)
from metadata.generated.schema.events.eventSubscription import (
    AlertType,
    Destination,
    EventSubscription,
    SubscriptionCategory,
    SubscriptionType,
)
from metadata.generated.schema.type.basic import EntityName
from metadata.generated.schema.type.entityReference import EntityReference
from metadata.generated.schema.type.entityReferenceList import EntityReferenceList
from metadata.ingestion.ometa.client import REST

from ..integration_base import generate_name  # noqa: TID252

# Mock response with invalid EventSubscription data
BAD_SUBSCRIPTION_RESPONSE = {
    "data": [
        {
            "id": "cb149dd4-f4c2-485e-acd3-74b7dca1015e",
            "name": "valid-subscription",
            "alertType": "Notification",
            "destinations": [
                {
                    "category": "External",
                    "type": "Webhook",
                    "config": {"endpoint": "https://example.com/webhook"},
                }
            ],
        },
        {
            "id": "5d76676c-8e94-4e7e-97b8-294f4c16d0aa",
            "name": "another-valid-subscription",
            "alertType": "Notification",
            "destinations": [
                {
                    "category": "External",
                    "type": "Webhook",
                    "config": {"endpoint": "https://example.com/webhook2"},
                }
            ],
        },
        {
            "id": "f063ff4e-99a3-4d42-8678-c484c2556e8d",
            "name": "invalid-subscription",
            "alertType": "Invalid-Type",  # Invalid enum value
            "destinations": [
                {
                    "category": "External",
                    "type": "Webhook",
                    "config": {"endpoint": "https://example.com/webhook3"},
                }
            ],
        },
    ],
    "paging": {
        "total": 3,
    },
}


@pytest.fixture(scope="module")
def subscription_user(metadata):
    """Create a user for subscription ownership tests."""
    user_name = generate_name()
    user = metadata.create_or_update(
        data=CreateUserRequest(name=user_name, email=f"{user_name.root}@test.com"),
    )

    yield user

    metadata.delete(entity=User, entity_id=user.id, hard_delete=True)


@pytest.fixture(scope="module")
def subscription_owners(subscription_user):
    """Owner reference list for subscription tests."""
    return EntityReferenceList(root=[EntityReference(id=subscription_user.id, type="user")])


@pytest.fixture
def subscription_request():
    """Create a subscription request with a unique name."""
    name = generate_name()
    return CreateEventSubscription(
        name=name,
        description="Test event subscription for integration testing",
        alertType=AlertType.Notification,
        resources=["All"],
        destinations=[
            Destination(
                category=SubscriptionCategory.External,
                type=SubscriptionType.Webhook,
                config={"endpoint": "https://example.com/test-webhook"},
            )
        ],
        enabled=True,
        batchSize=50,
        retries=3,
        pollInterval=30,
    )


@pytest.fixture
def create_subscription(metadata, request):
    """Factory fixture for creating subscriptions with automatic cleanup."""
    subscriptions = []

    def _create_subscription(create_request):
        subscription = metadata.create_or_update(data=create_request)
        subscriptions.append(subscription)
        return subscription

    def teardown():
        for sub in subscriptions:
            metadata.delete(
                entity=EventSubscription,
                entity_id=sub.id,
                hard_delete=True,
            )

    request.addfinalizer(teardown)

    return _create_subscription


class TestOMetaSubscriptionAPI:
    """
    EventSubscription API integration tests.
    Tests CRUD operations, pagination, and configuration options.

    Uses fixtures from conftest:
    - metadata: OpenMetadata client (session scope)
    """

    def test_create(self, metadata, subscription_request, create_subscription):
        """
        We can create an EventSubscription and we receive it back as Entity
        """
        res = create_subscription(subscription_request)
        assert res.name == subscription_request.name
        assert res.alertType == subscription_request.alertType
        assert len(res.destinations) == 1
        assert res.enabled is True
        assert res.batchSize == 50

    def test_get_name(self, metadata, subscription_request, create_subscription):
        """
        We can fetch an EventSubscription by name and get it back as Entity
        """
        created = create_subscription(subscription_request)

        res = metadata.get_by_name(entity=EventSubscription, fqn=subscription_request.name.root)
        assert res.name == created.name

    def test_get_id(self, metadata, subscription_request, create_subscription):
        """
        We can fetch an EventSubscription by ID and get it back as Entity
        """
        created = create_subscription(subscription_request)  # noqa: F841

        res_name = metadata.get_by_name(entity=EventSubscription, fqn=subscription_request.name.root)
        res = metadata.get_by_id(entity=EventSubscription, entity_id=res_name.id)
        assert res_name.id == res.id

    def test_list(self, metadata, subscription_request, create_subscription):
        """
        We can list all our EventSubscriptions
        """
        created = create_subscription(subscription_request)

        res = metadata.list_entities(entity=EventSubscription)

        data = next(
            iter(ent for ent in res.entities if ent.name == created.name),
            None,
        )
        assert data

    def test_list_all_and_paginate(self, metadata, subscription_request, create_subscription):
        """
        Validate generator utility to fetch all event subscriptions
        """
        base_name = subscription_request.name.root
        for i in range(0, 10):  # noqa: PIE808
            fake_create = deepcopy(subscription_request)
            fake_create.name = EntityName(base_name + str(i))
            create_subscription(fake_create)

        all_entities = metadata.list_all_entities(entity=EventSubscription, limit=2)
        assert len(list(all_entities)) >= 10

        entity_list = metadata.list_entities(entity=EventSubscription, limit=2)
        assert len(entity_list.entities) == 2
        after_entity_list = metadata.list_entities(entity=EventSubscription, limit=2, after=entity_list.after)
        assert len(after_entity_list.entities) == 2
        before_entity_list = metadata.list_entities(entity=EventSubscription, limit=2, before=after_entity_list.before)
        assert before_entity_list.entities == entity_list.entities

    def test_delete(self, metadata, subscription_request):
        """
        We can delete an EventSubscription by ID
        """
        subscription = metadata.create_or_update(data=subscription_request)

        res_name = metadata.get_by_name(entity=EventSubscription, fqn=subscription.fullyQualifiedName)
        res_id = metadata.get_by_id(entity=EventSubscription, entity_id=res_name.id)

        metadata.delete(entity=EventSubscription, entity_id=str(res_id.id.root))

        res = metadata.list_entities(entity=EventSubscription)
        assert not next(
            iter(ent for ent in res.entities if ent.fullyQualifiedName == subscription.fullyQualifiedName),
            None,
        )

    def test_update(
        self,
        metadata,
        subscription_request,
        subscription_user,
        subscription_owners,
        create_subscription,
    ):
        """
        Updating it properly changes its properties
        """
        res_create = create_subscription(subscription_request)

        updated = subscription_request.model_dump(exclude_unset=True)
        updated["owners"] = subscription_owners
        updated["description"] = "Updated description"
        updated["batchSize"] = 100
        updated_entity = CreateEventSubscription(**updated)

        res = metadata.create_or_update(data=updated_entity)

        assert res_create.id == res.id
        assert res.owners.root[0].id == subscription_user.id
        assert res.description.root == "Updated description"
        assert res.batchSize == 100

    def test_list_versions(self, metadata, subscription_request, create_subscription):
        """
        test list event subscription entity versions
        """
        subscription = create_subscription(subscription_request)

        res = metadata.get_list_entity_versions(entity=EventSubscription, entity_id=subscription.id.root)
        assert res

    def test_get_entity_version(self, metadata, subscription_request, create_subscription):
        """
        test get event subscription entity version
        """
        subscription = create_subscription(subscription_request)

        res = metadata.get_entity_version(
            entity=EventSubscription,
            entity_id=subscription.id.root,
            version=0.1,
        )

        assert res.version.root == 0.1
        assert res.id == subscription.id

    def test_get_entity_ref(self, metadata, subscription_request, create_subscription):
        """
        test get EventSubscription EntityReference
        """
        subscription = create_subscription(subscription_request)
        entity_ref = metadata.get_entity_reference(entity=EventSubscription, fqn=subscription.fullyQualifiedName)

        assert subscription.id == entity_ref.id

    def test_list_w_skip_on_failure(self, metadata):
        """
        We can list all our EventSubscriptions even when some of them are broken
        """
        with patch.object(REST, "get", return_value=BAD_SUBSCRIPTION_RESPONSE):  # noqa: SIM117
            with pytest.raises(ValidationError):
                metadata.list_entities(entity=EventSubscription)

        with patch.object(REST, "get", return_value=BAD_SUBSCRIPTION_RESPONSE):
            res = metadata.list_entities(entity=EventSubscription, skip_on_failure=True)

        assert len(res.entities) == 2

    def test_list_all_w_skip_on_failure(self, metadata):
        """
        Validate generator utility to fetch all event subscriptions even when some are broken
        """
        with patch.object(REST, "get", return_value=BAD_SUBSCRIPTION_RESPONSE):  # noqa: SIM117
            with pytest.raises(ValidationError):
                res = metadata.list_all_entities(
                    entity=EventSubscription,
                    limit=1,
                )
                list(res)

        with patch.object(REST, "get", return_value=BAD_SUBSCRIPTION_RESPONSE):
            res = metadata.list_all_entities(
                entity=EventSubscription,
                limit=1,
                skip_on_failure=True,
            )

            assert len(list(res)) == 2

    def test_subscription_with_slash_in_name(self, metadata):
        """E.g., `subscription.name/with-slash`"""
        name = EntityName("subscription.name/with-slash")
        create_request = CreateEventSubscription(
            name=name,
            description="Event subscription with slash in name",
            alertType=AlertType.Notification,
            resources=["All"],
            destinations=[
                Destination(
                    category=SubscriptionCategory.External,
                    type=SubscriptionType.Webhook,
                    config={"endpoint": "https://example.com/slash-webhook"},
                )
            ],
        )
        new_subscription: EventSubscription = metadata.create_or_update(data=create_request)

        res: EventSubscription = metadata.get_by_name(entity=EventSubscription, fqn=new_subscription.fullyQualifiedName)

        assert res.name == name

        metadata.delete(
            entity=EventSubscription,
            entity_id=new_subscription.id,
            hard_delete=True,
        )

    def test_different_alert_types(self, metadata):
        """
        Test creating subscriptions with different alert types
        """
        alert_types = [
            AlertType.Notification,
            AlertType.Observability,
        ]

        created_subscriptions = []

        for alert_type in alert_types:
            create_request = CreateEventSubscription(
                name=f"test-{alert_type.value.lower()}-subscription",
                description=f"Test {alert_type.value} subscription",
                alertType=alert_type,
                resources=["table"],
                destinations=[
                    Destination(
                        category=SubscriptionCategory.External,
                        type=SubscriptionType.Webhook,
                        config={"endpoint": f"https://example.com/{alert_type.value.lower()}-webhook"},
                    )
                ],
            )

            subscription = metadata.create_or_update(data=create_request)
            created_subscriptions.append(subscription)

            assert subscription.alertType == alert_type

        for subscription in created_subscriptions:
            metadata.delete(
                entity=EventSubscription,
                entity_id=subscription.id,
                hard_delete=True,
            )

    def test_different_destination_types(self, metadata):
        """
        Test creating subscriptions with different destination types
        """
        destination_configs = [
            {
                "type": SubscriptionType.Webhook,
                "config": {"endpoint": "https://example.com/webhook"},
            },
            {
                "type": SubscriptionType.Email,
                "config": {"receivers": ["test@example.com"]},
            },
        ]

        created_subscriptions = []

        for dest_config in destination_configs:
            create_request = CreateEventSubscription(
                name=f"test-{dest_config['type'].value.lower()}-destination",
                description=f"Test {dest_config['type'].value} destination",
                alertType=AlertType.Notification,
                resources=["All"],
                destinations=[
                    Destination(
                        category=SubscriptionCategory.External,
                        type=dest_config["type"],
                        config=dest_config["config"],
                    )
                ],
            )

            subscription = metadata.create_or_update(data=create_request)
            created_subscriptions.append(subscription)

            assert subscription.destinations[0].type == dest_config["type"]

        for subscription in created_subscriptions:
            metadata.delete(
                entity=EventSubscription,
                entity_id=subscription.id,
                hard_delete=True,
            )

    def test_subscription_configuration_options(self, metadata):
        """
        Test various configuration options for event subscriptions
        """
        create_request = CreateEventSubscription(
            name="test-config-subscription",
            description="Test subscription with custom configuration",
            alertType=AlertType.Observability,
            resources=["table"],
            destinations=[
                Destination(
                    category=SubscriptionCategory.External,
                    type=SubscriptionType.Webhook,
                    config={
                        "endpoint": "https://example.com/config-webhook",
                        "headers": {"Authorization": "Bearer token123"},
                    },
                    timeout=15,
                    readTimeout=20,
                    enabled=True,
                )
            ],
            enabled=True,
            batchSize=25,
            retries=5,
            pollInterval=60,
        )

        subscription = metadata.create_or_update(data=create_request)

        assert subscription.alertType == AlertType.Observability
        assert subscription.batchSize == 25
        assert subscription.retries == 5
        assert subscription.pollInterval == 60
        assert subscription.destinations[0].timeout == 15
        assert subscription.destinations[0].readTimeout == 20

        metadata.delete(
            entity=EventSubscription,
            entity_id=subscription.id,
            hard_delete=True,
        )