/ingestion/tests/unit/observability/profiler/test_entity_fetcher.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.

"""
Validate entity fetcher filtering strategies
"""

import uuid
from unittest.mock import MagicMock, patch

import pytest

from metadata.generated.schema.entity.data.database import Database
from metadata.generated.schema.entity.data.table import (
    Column,
    DataType,
    Table,
    TableType,
)
from metadata.generated.schema.entity.services.connections.metadata.openMetadataConnection import (
    OpenMetadataConnection,
)
from metadata.generated.schema.metadataIngestion.databaseServiceAutoClassificationPipeline import (
    DatabaseServiceAutoClassificationPipeline,
)
from metadata.generated.schema.metadataIngestion.workflow import (
    OpenMetadataWorkflowConfig,
    Source,
    SourceConfig,
    WorkflowConfig,
)
from metadata.generated.schema.type.entityReference import EntityReference
from metadata.generated.schema.type.tagLabel import TagLabel
from metadata.ingestion.api.status import Status
from metadata.profiler.source.fetcher.fetcher_strategy import DatabaseFetcherStrategy
from metadata.profiler.source.profiler_source_interface import ProfilerSourceInterface

VIEW = Table(
    id=uuid.uuid4(),
    name="view",
    columns=[],
    tableType=TableType.View,
)

TABLE = Table(
    id=uuid.uuid4(),
    name="table",
    columns=[],
    tableType=TableType.Regular,
)


def get_db_fetcher(source_config):
    """Fetch database"""
    workflow_config = OpenMetadataWorkflowConfig(
        source=Source(
            type="mysql",
            serviceName="mysql",
            sourceConfig=SourceConfig(
                config=source_config,
            ),
        ),
        workflowConfig=WorkflowConfig(
            openMetadataServerConfig=OpenMetadataConnection(
                hostPort="localhost:8585/api",
            )
        ),
    )
    return DatabaseFetcherStrategy(
        config=workflow_config,
        metadata=...,
        global_profiler_config=...,
        status=Status(),
    )


def test_include_views():
    """Validate we can include/exclude views"""
    config = DatabaseServiceAutoClassificationPipeline(
        includeViews=False,
    )
    fetcher = get_db_fetcher(config)

    assert fetcher._filter_views(VIEW)
    assert not fetcher._filter_views(TABLE)

    config = DatabaseServiceAutoClassificationPipeline(
        includeViews=True,
    )
    fetcher = get_db_fetcher(config)

    assert not fetcher._filter_views(VIEW)
    assert not fetcher._filter_views(TABLE)


PROFILER_CONFIG = {
    "source": {
        "type": "sqlite",
        "serviceName": "my_service",
        "serviceConnection": {"config": {"type": "SQLite"}},
        "sourceConfig": {"config": {"type": "Profiler"}},
    },
    "processor": {"type": "orm-profiler", "config": {}},
    "sink": {"type": "metadata-rest", "config": {}},
    "workflowConfig": {
        "openMetadataServerConfig": {
            "hostPort": "http://localhost:8585/api",
            "authProvider": "openmetadata",
            "securityConfig": {"jwtToken": "fake.jwt.token"},
        }
    },
}

SERVICE_REF = EntityReference(id=uuid.uuid4(), name="my_service", type="databaseService")

PROD_DB = Database(
    id=uuid.uuid4(),
    name="prod",
    fullyQualifiedName="my_service.prod",
    service=SERVICE_REF,
)

STAGING_DB = Database(
    id=uuid.uuid4(),
    name="staging",
    fullyQualifiedName="my_service.staging",
    service=SERVICE_REF,
)

TEMP_DB = Database(
    id=uuid.uuid4(),
    name="temp_analytics",
    fullyQualifiedName="my_service.temp_analytics",
    service=SERVICE_REF,
)

FINANCE_SCHEMA_REF = EntityReference(
    id=uuid.uuid4(),
    name="finance",
    type="databaseSchema",
    fullyQualifiedName="my_service.prod.finance",
)

HR_SCHEMA_REF = EntityReference(
    id=uuid.uuid4(),
    name="hr",
    type="databaseSchema",
    fullyQualifiedName="my_service.prod.hr",
)

DB_REF = EntityReference(id=uuid.uuid4(), name="prod", type="database")

ORDERS_TABLE = Table(
    id=uuid.uuid4(),
    name="orders",
    fullyQualifiedName="my_service.prod.finance.orders",
    columns=[Column(name="id", dataType=DataType.INT)],
    database=DB_REF,
    databaseSchema=FINANCE_SCHEMA_REF,
    tableType=TableType.Regular,
)

CUSTOMERS_TABLE = Table(
    id=uuid.uuid4(),
    name="customers",
    fullyQualifiedName="my_service.prod.finance.customers",
    columns=[Column(name="id", dataType=DataType.INT)],
    database=DB_REF,
    databaseSchema=FINANCE_SCHEMA_REF,
    tableType=TableType.Regular,
)

EMPLOYEES_TABLE = Table(
    id=uuid.uuid4(),
    name="employees",
    fullyQualifiedName="my_service.prod.hr.employees",
    columns=[Column(name="id", dataType=DataType.INT)],
    database=DB_REF,
    databaseSchema=HR_SCHEMA_REF,
    tableType=TableType.Regular,
)

REVENUE_VIEW = Table(
    id=uuid.uuid4(),
    name="revenue_summary",
    fullyQualifiedName="my_service.prod.finance.revenue_summary",
    columns=[Column(name="total", dataType=DataType.DECIMAL)],
    database=DB_REF,
    databaseSchema=FINANCE_SCHEMA_REF,
    tableType=TableType.View,
)

SALARY_TABLE = Table(
    id=uuid.uuid4(),
    name="salaries",
    fullyQualifiedName="my_service.prod.hr.salaries",
    columns=[Column(name="amount", dataType=DataType.DECIMAL)],
    database=DB_REF,
    databaseSchema=HR_SCHEMA_REF,
    tableType=TableType.Regular,
    tags=[
        TagLabel(
            labelType="Manual",
            name="PII",
            tagFQN="PII.Sensitive",
            state="Confirmed",
            source="Classification",
        ),
    ],
)

UNTAGGED_TABLE = Table(
    id=uuid.uuid4(),
    name="departments",
    fullyQualifiedName="my_service.prod.hr.departments",
    columns=[Column(name="name", dataType=DataType.STRING)],
    database=DB_REF,
    databaseSchema=HR_SCHEMA_REF,
    tableType=TableType.Regular,
)

MULTI_TAG_TABLE = Table(
    id=uuid.uuid4(),
    name="transactions",
    fullyQualifiedName="my_service.prod.finance.transactions",
    columns=[Column(name="id", dataType=DataType.INT)],
    database=DB_REF,
    databaseSchema=FINANCE_SCHEMA_REF,
    tableType=TableType.Regular,
    tags=[
        TagLabel(
            labelType="Manual",
            name="PII",
            tagFQN="PII.Sensitive",
            state="Confirmed",
            source="Classification",
        ),
        TagLabel(
            labelType="Manual",
            name="Revenue",
            tagFQN="Finance.Revenue",
            state="Confirmed",
            source="Classification",
        ),
    ],
)

TIER_REVENUE_TABLE = Table(
    id=uuid.uuid4(),
    name="ledger",
    fullyQualifiedName="my_service.prod.finance.ledger",
    columns=[Column(name="id", dataType=DataType.INT)],
    database=DB_REF,
    databaseSchema=FINANCE_SCHEMA_REF,
    tableType=TableType.Regular,
    tags=[
        TagLabel(
            labelType="Manual",
            name="Tier1",
            tagFQN="Tier.Tier1",
            state="Confirmed",
            source="Classification",
        ),
        TagLabel(
            labelType="Manual",
            name="Revenue",
            tagFQN="Finance.Revenue",
            state="Confirmed",
            source="Classification",
        ),
    ],
)


def _make_fetcher(source_config_overrides=None):
    """Build a DatabaseFetcherStrategy with a mocked metadata client"""
    from copy import deepcopy

    cfg = deepcopy(PROFILER_CONFIG)
    if source_config_overrides:
        cfg["source"]["sourceConfig"]["config"].update(source_config_overrides)
    workflow_config = OpenMetadataWorkflowConfig(**cfg)
    mock_metadata = MagicMock()
    return DatabaseFetcherStrategy(
        config=workflow_config,
        metadata=mock_metadata,
        global_profiler_config=None,
        status=Status(),
    )


class TestGetDatabaseEntities:
    """Validate _get_database_entities generator and param passing"""

    def test_include_pattern_sends_regex_and_yields_matching_databases(self):
        """With includes=["prod"], the API receives databaseRegex="prod"
        with regexMode=include. The server returns only matching databases."""
        fetcher = _make_fetcher({"databaseFilterPattern": {"includes": ["prod"]}})
        fetcher.metadata.list_all_entities.return_value = iter([PROD_DB])

        result = list(fetcher._get_database_entities())

        fetcher.metadata.list_all_entities.assert_called_once_with(
            entity=Database,
            params={
                "service": "my_service",
                "databaseRegex": "prod",
                "regexMode": "include",
            },
        )
        assert result == [PROD_DB]

    def test_exclude_pattern_sends_exclude_mode(self):
        """With excludes=["temp.*"], the API receives regexMode=exclude.
        The server filters out temp_analytics and returns only prod+staging."""
        fetcher = _make_fetcher({"databaseFilterPattern": {"excludes": ["temp.*"]}})
        fetcher.metadata.list_all_entities.return_value = iter([PROD_DB, STAGING_DB])

        result = list(fetcher._get_database_entities())

        call_params = fetcher.metadata.list_all_entities.call_args[1]["params"]
        assert call_params["databaseRegex"] == "temp.*"
        assert call_params["regexMode"] == "exclude"
        assert result == [PROD_DB, STAGING_DB]
        assert TEMP_DB not in result

    def test_multiple_includes_combined_with_or(self):
        """includes=["prod", "staging"] combines into "(prod)|(staging)".
        The server returns both matching databases but not temp_analytics."""
        fetcher = _make_fetcher({"databaseFilterPattern": {"includes": ["prod", "staging"]}})
        fetcher.metadata.list_all_entities.return_value = iter([PROD_DB, STAGING_DB])

        result = list(fetcher._get_database_entities())

        call_params = fetcher.metadata.list_all_entities.call_args[1]["params"]
        assert call_params["databaseRegex"] == "(prod)|(staging)"
        assert result == [PROD_DB, STAGING_DB]
        assert TEMP_DB not in result

    def test_fqn_filtering_adds_regex_filter_by_fqn(self):
        """When useFqnForFiltering=True, regexFilterByFqn must be passed
        so the server matches against fullyQualifiedName, not just name.
        Only prod matches "my_service\\.prod"."""
        fetcher = _make_fetcher(
            {
                "useFqnForFiltering": True,
                "databaseFilterPattern": {"includes": ["my_service\\.prod"]},
            }
        )
        fetcher.metadata.list_all_entities.return_value = iter([PROD_DB])

        result = list(fetcher._get_database_entities())

        call_params = fetcher.metadata.list_all_entities.call_args[1]["params"]
        assert call_params["regexFilterByFqn"] == "true"
        assert call_params["databaseRegex"] == "my_service\\.prod"
        assert result == [PROD_DB]

    def test_no_filter_pattern_fetches_all_databases(self):
        """Without a filter pattern, no regex params should be sent
        and the API returns all databases for the service."""
        fetcher = _make_fetcher()
        fetcher.metadata.list_all_entities.return_value = iter([PROD_DB, STAGING_DB, TEMP_DB])

        result = list(fetcher._get_database_entities())

        call_params = fetcher.metadata.list_all_entities.call_args[1]["params"]
        assert call_params == {"service": "my_service"}
        assert len(result) == 3

    def test_raises_when_server_returns_no_databases(self):
        """If the server returns 0 results (e.g., overly restrictive regex),
        a ValueError should be raised with the filter pattern details."""
        fetcher = _make_fetcher({"databaseFilterPattern": {"includes": ["nonexistent"]}})
        fetcher.metadata.list_all_entities.return_value = iter([])

        with pytest.raises(ValueError, match="databaseFilterPattern returned 0 result"):
            list(fetcher._get_database_entities())


class TestGetTableEntities:
    """Validate _get_table_entities generator, param passing, and client-side filters"""

    def test_schema_and_table_includes_sent_as_regex(self):
        """With schemaFilterPattern=finance and tableFilterPattern=orders,
        the server returns only finance.orders — not hr tables or customers."""
        fetcher = _make_fetcher(
            {
                "schemaFilterPattern": {"includes": ["finance"]},
                "tableFilterPattern": {"includes": ["orders"]},
                "includeViews": True,
            }
        )
        fetcher.metadata.list_all_entities.return_value = iter([ORDERS_TABLE])

        result = list(fetcher._get_table_entities(PROD_DB))

        call_params = fetcher.metadata.list_all_entities.call_args[1]["params"]
        assert call_params["databaseSchemaRegex"] == "finance"
        assert call_params["tableRegex"] == "orders"
        assert call_params["regexMode"] == "include"
        assert call_params["database"] == "my_service.prod"
        assert result == [ORDERS_TABLE]

    def test_views_filtered_client_side_from_mixed_results(self):
        """The API returns a mix of regular tables and views.
        With includeViews=False, views must be stripped client-side."""
        fetcher = _make_fetcher({"includeViews": False})
        fetcher.metadata.list_all_entities.return_value = iter([ORDERS_TABLE, REVENUE_VIEW, CUSTOMERS_TABLE])

        result = list(fetcher._get_table_entities(PROD_DB))

        assert REVENUE_VIEW not in result
        assert result == [ORDERS_TABLE, CUSTOMERS_TABLE]

    def test_views_included_when_configured(self):
        """With includeViews=True, views should pass through."""
        fetcher = _make_fetcher({"includeViews": True})
        fetcher.metadata.list_all_entities.return_value = iter([ORDERS_TABLE, REVENUE_VIEW])

        result = list(fetcher._get_table_entities(PROD_DB))

        assert result == [ORDERS_TABLE, REVENUE_VIEW]

    def test_classification_exclude_filters_tagged_tables(self):
        """With classificationFilterPattern excludes=["PII"],
        tables tagged with PII should be removed client-side
        while untagged tables pass through."""
        fetcher = _make_fetcher(
            {
                "classificationFilterPattern": {"excludes": ["PII.*"]},
            }
        )
        fetcher.metadata.list_all_entities.return_value = iter([EMPLOYEES_TABLE, SALARY_TABLE, UNTAGGED_TABLE])

        result = list(fetcher._get_table_entities(PROD_DB))

        assert SALARY_TABLE not in result
        assert result == [EMPLOYEES_TABLE, UNTAGGED_TABLE]

    def test_classification_include_filters_untagged_tables(self):
        """With classificationFilterPattern includes=["PII.*"],
        only tables with matching tags should be returned.
        Untagged tables and tables without matching tags are excluded."""
        fetcher = _make_fetcher(
            {
                "classificationFilterPattern": {"includes": ["PII.*"]},
            }
        )
        fetcher.metadata.list_all_entities.return_value = iter([EMPLOYEES_TABLE, SALARY_TABLE, UNTAGGED_TABLE])

        result = list(fetcher._get_table_entities(PROD_DB))

        assert result == [SALARY_TABLE]

    def test_classification_include_keeps_table_with_extra_tags(self):
        """A table carrying a matching include tag is kept even when it also
        has other, non-matching tags. A single matching tag is sufficient."""
        fetcher = _make_fetcher(
            {
                "classificationFilterPattern": {"includes": ["PII.*"]},
            }
        )
        fetcher.metadata.list_all_entities.return_value = iter([MULTI_TAG_TABLE, TIER_REVENUE_TABLE])

        result = list(fetcher._get_table_entities(PROD_DB))

        assert result == [MULTI_TAG_TABLE]

    def test_classification_exclude_takes_precedence_over_include(self):
        """When a table matches both an include and an exclude pattern,
        exclude wins and the table is filtered out."""
        fetcher = _make_fetcher(
            {
                "classificationFilterPattern": {
                    "includes": ["Tier.*"],
                    "excludes": ["Revenue.*"],
                },
            }
        )
        fetcher.metadata.list_all_entities.return_value = iter([TIER_REVENUE_TABLE])

        result = list(fetcher._get_table_entities(PROD_DB))

        assert result == []

    def test_classification_exclude_keeps_multi_tag_table_without_excluded_tag(self):
        """A multi-tag table is kept when none of its tags match the exclude
        pattern, regardless of how many other tags it carries."""
        fetcher = _make_fetcher(
            {
                "classificationFilterPattern": {"excludes": ["PII.*"]},
            }
        )
        fetcher.metadata.list_all_entities.return_value = iter([TIER_REVENUE_TABLE, MULTI_TAG_TABLE])

        result = list(fetcher._get_table_entities(PROD_DB))

        assert result == [TIER_REVENUE_TABLE]

    def test_views_and_classifications_combined(self):
        """Both view filtering and classification filtering
        should apply simultaneously."""
        fetcher = _make_fetcher(
            {
                "includeViews": False,
                "classificationFilterPattern": {"excludes": ["PII.*"]},
            }
        )
        fetcher.metadata.list_all_entities.return_value = iter(
            [ORDERS_TABLE, REVENUE_VIEW, SALARY_TABLE, UNTAGGED_TABLE]
        )

        result = list(fetcher._get_table_entities(PROD_DB))

        assert REVENUE_VIEW not in result
        assert SALARY_TABLE not in result
        assert result == [ORDERS_TABLE, UNTAGGED_TABLE]

    def test_default_config_excludes_views_only(self):
        """Default config has includeViews=False and no classification filter.
        Only views should be excluded."""
        fetcher = _make_fetcher()
        fetcher.metadata.list_all_entities.return_value = iter(
            [ORDERS_TABLE, REVENUE_VIEW, SALARY_TABLE, UNTAGGED_TABLE]
        )

        result = list(fetcher._get_table_entities(PROD_DB))

        assert REVENUE_VIEW not in result
        assert result == [ORDERS_TABLE, SALARY_TABLE, UNTAGGED_TABLE]

    def test_fqn_filtering_with_schema_exclude(self):
        """useFqnForFiltering + schema exclude sends regexFilterByFqn.
        Server excludes hr schema tables, returns only finance tables."""
        fetcher = _make_fetcher(
            {
                "useFqnForFiltering": True,
                "schemaFilterPattern": {"excludes": ["my_service\\.prod\\.hr"]},
                "includeViews": True,
            }
        )
        finance_tables = [ORDERS_TABLE, CUSTOMERS_TABLE, REVENUE_VIEW]
        fetcher.metadata.list_all_entities.return_value = iter(finance_tables)

        result = list(fetcher._get_table_entities(PROD_DB))

        call_params = fetcher.metadata.list_all_entities.call_args[1]["params"]
        assert call_params["regexFilterByFqn"] == "true"
        assert call_params["databaseSchemaRegex"] == "my_service\\.prod\\.hr"
        assert call_params["regexMode"] == "exclude"
        assert result == finance_tables
        assert EMPLOYEES_TABLE not in result
        assert SALARY_TABLE not in result

    def test_no_regex_params_when_no_schema_or_table_filter(self):
        """Without schema or table filters, no regex params should be sent.
        The API returns all tables for the database."""
        fetcher = _make_fetcher({"includeViews": True})
        all_tables = [
            ORDERS_TABLE,
            CUSTOMERS_TABLE,
            EMPLOYEES_TABLE,
            REVENUE_VIEW,
            SALARY_TABLE,
            UNTAGGED_TABLE,
        ]
        fetcher.metadata.list_all_entities.return_value = iter(all_tables)

        result = list(fetcher._get_table_entities(PROD_DB))

        call_params = fetcher.metadata.list_all_entities.call_args[1]["params"]
        assert "databaseSchemaRegex" not in call_params
        assert "tableRegex" not in call_params
        assert "regexMode" not in call_params
        assert len(result) == len(all_tables)

    def test_conflicting_modes_include_schema_exclude_table(self):
        """When schemaFilterPattern uses includes and tableFilterPattern
        uses excludes, only the schema include is pushed to the backend.
        The table exclude is applied client-side."""
        fetcher = _make_fetcher(
            {
                "schemaFilterPattern": {"includes": ["finance"]},
                "tableFilterPattern": {"excludes": ["customers"]},
                "includeViews": True,
            }
        )
        fetcher.metadata.list_all_entities.return_value = iter([ORDERS_TABLE, CUSTOMERS_TABLE, REVENUE_VIEW])

        result = list(fetcher._get_table_entities(PROD_DB))

        call_params = fetcher.metadata.list_all_entities.call_args[1]["params"]
        assert call_params["databaseSchemaRegex"] == "finance"
        assert call_params["regexMode"] == "include"
        assert "tableRegex" not in call_params
        assert CUSTOMERS_TABLE not in result
        assert result == [ORDERS_TABLE, REVENUE_VIEW]

    def test_conflicting_modes_exclude_schema_include_table(self):
        """When schemaFilterPattern uses excludes and tableFilterPattern
        uses includes, only the table include is pushed to the backend.
        The schema exclude is applied client-side."""
        fetcher = _make_fetcher(
            {
                "schemaFilterPattern": {"excludes": ["hr"]},
                "tableFilterPattern": {"includes": ["orders"]},
                "includeViews": True,
            }
        )
        fetcher.metadata.list_all_entities.return_value = iter([ORDERS_TABLE, EMPLOYEES_TABLE])

        result = list(fetcher._get_table_entities(PROD_DB))

        call_params = fetcher.metadata.list_all_entities.call_args[1]["params"]
        assert call_params["tableRegex"] == "orders"
        assert call_params["regexMode"] == "include"
        assert "databaseSchemaRegex" not in call_params
        assert EMPLOYEES_TABLE not in result
        assert result == [ORDERS_TABLE]


class TestFetch:
    """Validate end-to-end fetch() pipeline across multiple databases"""

    @patch("metadata.profiler.source.fetcher.fetcher_strategy.profiler_source_factory")
    def test_fetch_iterates_databases_and_tables_with_correct_params(self, mock_factory):
        """fetch() should iterate over each database from _get_database_entities,
        then for each database call _get_table_entities with the right params.
        Verifies the full param chain from config to API calls."""
        mock_factory.create.return_value = MagicMock(spec=ProfilerSourceInterface)
        fetcher = _make_fetcher(
            {
                "databaseFilterPattern": {"includes": ["prod", "staging"]},
                "schemaFilterPattern": {"includes": ["finance"]},
                "includeViews": True,
            }
        )
        fetcher.metadata.list_all_entities.side_effect = [
            iter([PROD_DB, STAGING_DB]),
            iter([ORDERS_TABLE, CUSTOMERS_TABLE]),
            iter([]),
        ]

        results = list(fetcher.fetch())

        entities = [r.right.entity for r in results if r.right]
        assert entities == [ORDERS_TABLE, CUSTOMERS_TABLE]

        (
            db_call,
            table_call_1,
            table_call_2,
        ) = fetcher.metadata.list_all_entities.call_args_list

        db_params = db_call[1]["params"]
        assert db_params["databaseRegex"] == "(prod)|(staging)"
        assert db_params["regexMode"] == "include"

        table_params_1 = table_call_1[1]["params"]
        assert table_params_1["database"] == "my_service.prod"
        assert table_params_1["databaseSchemaRegex"] == "finance"

        table_params_2 = table_call_2[1]["params"]
        assert table_params_2["database"] == "my_service.staging"
        assert table_params_2["databaseSchemaRegex"] == "finance"

    @patch("metadata.profiler.source.fetcher.fetcher_strategy.profiler_source_factory")
    def test_fetch_applies_client_side_filters_on_server_results(self, mock_factory):
        """The server returns tables including views and PII-tagged tables.
        fetch() must strip views and PII tables client-side while
        keeping everything else."""
        mock_factory.create.return_value = MagicMock(spec=ProfilerSourceInterface)
        fetcher = _make_fetcher(
            {
                "includeViews": False,
                "classificationFilterPattern": {"excludes": ["PII.*"]},
            }
        )
        fetcher.metadata.list_all_entities.side_effect = [
            iter([PROD_DB]),
            iter(
                [
                    ORDERS_TABLE,
                    REVENUE_VIEW,
                    SALARY_TABLE,
                    CUSTOMERS_TABLE,
                    UNTAGGED_TABLE,
                ]
            ),
        ]

        results = list(fetcher.fetch())

        entities = [r.right.entity for r in results if r.right]
        assert REVENUE_VIEW not in entities
        assert SALARY_TABLE not in entities
        assert entities == [ORDERS_TABLE, CUSTOMERS_TABLE, UNTAGGED_TABLE]

    @patch("metadata.profiler.source.fetcher.fetcher_strategy.profiler_source_factory")
    def test_fetch_handles_profiler_source_error_per_database(self, mock_factory):
        """If profiler_source_factory.create fails for one database,
        it should yield an error for that database but continue
        processing the next one."""
        call_count = 0

        def create_side_effect(*args, **kwargs):
            nonlocal call_count
            call_count += 1
            if call_count == 1:
                raise RuntimeError("connection failed for prod")
            return MagicMock(spec=ProfilerSourceInterface)

        mock_factory.create.side_effect = create_side_effect
        fetcher = _make_fetcher()
        fetcher.metadata.list_all_entities.side_effect = [
            iter([PROD_DB, STAGING_DB]),
            iter([ORDERS_TABLE]),
        ]

        results = list(fetcher.fetch())

        assert len(results) == 2
        assert results[0].left is not None
        assert "connection failed for prod" in results[0].left.error
        assert results[1].right is not None
        assert results[1].right.entity == ORDERS_TABLE