"""
Test Tableau Dashboard
"""
import uuid
from datetime import datetime, timedelta
from types import SimpleNamespace
from unittest import TestCase
from unittest.mock import MagicMock, patch

from metadata.generated.schema.api.data.createChart import CreateChartRequest
from metadata.generated.schema.api.data.createDashboard import CreateDashboardRequest
from metadata.generated.schema.entity.data.dashboard import Dashboard
from metadata.generated.schema.entity.data.dashboardDataModel import DashboardDataModel
from metadata.generated.schema.entity.services.dashboardService import (
    DashboardConnection,
    DashboardService,
    DashboardServiceType,
)
from metadata.generated.schema.metadataIngestion.workflow import (
    OpenMetadataWorkflowConfig,
)
from metadata.generated.schema.type.basic import FullyQualifiedEntityName
from metadata.generated.schema.type.entityReference import EntityReference
from metadata.generated.schema.type.filterPattern import FilterPattern
from metadata.generated.schema.type.usageDetails import UsageDetails, UsageStats
from metadata.generated.schema.type.usageRequest import UsageRequest
from metadata.ingestion.ometa.ometa_api import OpenMetadata
from metadata.ingestion.source.dashboard.dashboard_service import DashboardUsage
from metadata.ingestion.source.dashboard.tableau.metadata import (
    TableauDashboard,
    TableauSource,
)
from metadata.ingestion.source.dashboard.tableau.models import (
    DataSource,
    TableauBaseModel,
    TableauChart,
    TableauOwner,
    UpstreamTable,
)

MOCK_DASHBOARD_SERVICE = DashboardService(
    id="c3eb265f-5445-4ad3-ba5e-797d3a3071bb",
    fullyQualifiedName=FullyQualifiedEntityName("tableau_source_test"),
    name="tableau_source_test",
    connection=DashboardConnection(),
    serviceType=DashboardServiceType.Tableau,
)

mock_tableau_config = {
    "source": {
        "type": "tableau",
        "serviceName": "test2",
        "serviceConnection": {
            "config": {
                "type": "Tableau",
                "authType": {"username": "username", "password": "abcdefg"},
                "hostPort": "http://tableauHost.com",
                "siteName": "tableauSiteName",
            }
        },
        "sourceConfig": {
            "config": {"dashboardFilterPattern": {}, "chartFilterPattern": {}}
        },
    },
    "sink": {"type": "metadata-rest", "config": {}},
    "workflowConfig": {
        "openMetadataServerConfig": {
            "hostPort": "http://localhost:8585/api",
            "authProvider": "openmetadata",
            "securityConfig": {
                "jwtToken": "eyJraWQiOiJHYjM4OWEtOWY3Ni1nZGpzLWE5MmotMDI0MmJrOTQzNTYiLCJ0eXAiOiJKV1QiLCJhbGc"
                "iOiJSUzI1NiJ9.eyJzdWIiOiJhZG1pbiIsImlzQm90IjpmYWxzZSwiaXNzIjoib3Blbi1tZXRhZGF0YS5vcmciLCJpYXQiOjE"
                "6NjM5Mzg0NjIsImVtYWlsIjoiYWRtaW5Ab3Blbm1ldGFkYXRhLm9yZyJ9.tS8um_5DKu7HgzGBzS1VTA5uUjKWOCU0B_j08WXB"
                "iEC0mr0zNREkqVfwFDD-d24HlNEbrqioLsBuFRiwIWKc1m_ZlVQbG7P36RUxhuv2vbSp80FKyNM-Tj93FDzq91jsyNmsQhyNv_fN"
                "r3TXfzzSPjHt8Go0FMMP66weoKMgW2PbXlhVKwEuXUHyakLLzewm9UMeQaEiRzhiTMU3UkLXcKbYEJJvfNFcLwSl9W8JCO_l0Yj3u"
                "d-qt_nQYEZwqW6u5nfdQllN133iikV4fM5QZsMCnm8Rq1mvLR0y9bmJiD7fwM1tmJ791TUWqmKaTnP49U493VanKpUAfzIiOiIbhg"
            },
        }
    },
}

MOCK_DASHBOARD = TableauDashboard(
    id="42a5b706-739d-4d62-94a2-faedf33950a5",
    name="Regional",
    webpageUrl="http://tableauHost.com/#/site/hidarsite/workbooks/897790",
    description="tableau dashboard description",
    user_views=10,
    tags=[],
    owner=TableauOwner(
        id="1234", name="Dashboard Owner", email="samplemail@sample.com"
    ),
    charts=[
        TableauChart(
            id="b05695a2-d1ea-428e-96b2-858809809da4",
            name="Obesity",
            workbook=TableauBaseModel(id="42a5b706-739d-4d62-94a2-faedf33950a5"),
            sheetType="dashboard",
            viewUrlName="Obesity",
            contentUrl="Regional/sheets/Obesity",
            tags=[],
        ),
        TableauChart(
            id="106ff64d-537b-4534-8140-5d08c586e077",
            name="College",
            workbook=TableauBaseModel(id="42a5b706-739d-4d62-94a2-faedf33950a5"),
            sheetType="view",
            viewUrlName="College",
            contentUrl="Regional/sheets/College",
            tags=[],
        ),
        TableauChart(
            id="c1493abc-9057-4bdf-9061-c6d2908e4eaa",
            name="Global Temperatures",
            workbook=TableauBaseModel(id="42a5b706-739d-4d62-94a2-faedf33950a5"),
            sheetType="dashboard",
            viewUrlName="GlobalTemperatures",
            contentUrl="Regional/sheets/GlobalTemperatures",
            tags=[],
        ),
    ],
)

EXPECTED_DASHBOARD = [
    CreateDashboardRequest(
        name="42a5b706-739d-4d62-94a2-faedf33950a5",
        displayName="Regional",
        description="tableau dashboard description",
        sourceUrl="http://tableauHost.com/#/site/hidarsite/workbooks/897790/views",
        charts=[],
        tags=[],
        owners=None,
        service=FullyQualifiedEntityName("tableau_source_test"),
        extension=None,
    )
]

EXPECTED_CHARTS = [
    CreateChartRequest(
        name="b05695a2-d1ea-428e-96b2-858809809da4",
        displayName="Obesity",
        description=None,
        chartType="Other",
        sourceUrl="http://tableauHost.com/#/site/tableauSiteUrl/views/Regional/Obesity",
        tags=None,
        owners=None,
        service=FullyQualifiedEntityName("tableau_source_test"),
    ),
    CreateChartRequest(
        name="106ff64d-537b-4534-8140-5d08c586e077",
        displayName="College",
        description=None,
        chartType="Other",
        sourceUrl="http://tableauHost.com/#/site/tableauSiteUrl/views/Regional/College",
        tags=None,
        owners=None,
        service=FullyQualifiedEntityName("tableau_source_test"),
    ),
    CreateChartRequest(
        name="c1493abc-9057-4bdf-9061-c6d2908e4eaa",
        displayName="Global Temperatures",
        description=None,
        chartType="Other",
        sourceUrl="http://tableauHost.com/#/site/tableauSiteUrl/views/Regional/GlobalTemperatures",
        tags=None,
        owners=None,
        service=FullyQualifiedEntityName("tableau_source_test"),
    ),
]


class TableauUnitTest(TestCase):
    """
    Implements the necessary methods to extract
    Domo Dashboard Unit Test
    """

    @patch(
        "metadata.ingestion.source.dashboard.dashboard_service.DashboardServiceSource.test_connection"
    )
    @patch("metadata.ingestion.source.dashboard.tableau.connection.get_connection")
    def __init__(self, methodName, get_connection, test_connection) -> None:
        super().__init__(methodName)
        get_connection.return_value = False
        test_connection.return_value = False
        self.config = OpenMetadataWorkflowConfig.model_validate(mock_tableau_config)
        self.tableau = TableauSource.create(
            mock_tableau_config["source"],
            OpenMetadata(self.config.workflowConfig.openMetadataServerConfig),
        )
        self.tableau.client = SimpleNamespace()
        self.tableau.context.get().__dict__[
            "dashboard_service"
        ] = MOCK_DASHBOARD_SERVICE.fullyQualifiedName.root

    def test_dashboard_name(self):
        assert self.tableau.get_dashboard_name(MOCK_DASHBOARD) == MOCK_DASHBOARD.name

    def test_yield_chart(self):
        """
        Function for testing charts
        """
        chart_list = []
        results = self.tableau.yield_dashboard_chart(MOCK_DASHBOARD)
        for result in results:
            if isinstance(result, CreateChartRequest):
                chart_list.append(result)

        for _, (exptected, original) in enumerate(zip(EXPECTED_CHARTS, chart_list)):
            self.assertEqual(exptected, original)

    def test_yield_dashboard_usage(self):
        """
        Validate the logic for existing or new usage
        """
        self.tableau.context.get().__dict__["dashboard"] = "dashboard_name"

        # Start checking dashboard without usage
        # and a view count
        return_value = Dashboard(
            id=uuid.uuid4(),
            name="dashboard_name",
            fullyQualifiedName="dashboard_service.dashboard_name",
            service=EntityReference(id=uuid.uuid4(), type="dashboardService"),
        )
        with patch.object(OpenMetadata, "get_by_name", return_value=return_value):
            got_usage = next(self.tableau.yield_dashboard_usage(MOCK_DASHBOARD))
            self.assertEqual(
                got_usage.right,
                DashboardUsage(
                    dashboard=return_value,
                    usage=UsageRequest(date=self.tableau.today, count=10),
                ),
            )

        # Now check what happens if we already have some summary data for today
        return_value = Dashboard(
            id=uuid.uuid4(),
            name="dashboard_name",
            fullyQualifiedName="dashboard_service.dashboard_name",
            service=EntityReference(id=uuid.uuid4(), type="dashboardService"),
            usageSummary=UsageDetails(
                dailyStats=UsageStats(count=10), date=self.tableau.today
            ),
        )
        with patch.object(OpenMetadata, "get_by_name", return_value=return_value):
            # Nothing is returned
            self.assertEqual(
                len(list(self.tableau.yield_dashboard_usage(MOCK_DASHBOARD))), 0
            )

        # But if we have usage for today but the count is 0, we'll return the details
        return_value = Dashboard(
            id=uuid.uuid4(),
            name="dashboard_name",
            fullyQualifiedName="dashboard_service.dashboard_name",
            service=EntityReference(id=uuid.uuid4(), type="dashboardService"),
            usageSummary=UsageDetails(
                dailyStats=UsageStats(count=0), date=self.tableau.today
            ),
        )
        with patch.object(OpenMetadata, "get_by_name", return_value=return_value):
            self.assertEqual(
                next(self.tableau.yield_dashboard_usage(MOCK_DASHBOARD)).right,
                DashboardUsage(
                    dashboard=return_value,
                    usage=UsageRequest(date=self.tableau.today, count=10),
                ),
            )

        # But if we have usage for another day, then we do the difference
        return_value = Dashboard(
            id=uuid.uuid4(),
            name="dashboard_name",
            fullyQualifiedName="dashboard_service.dashboard_name",
            service=EntityReference(id=uuid.uuid4(), type="dashboardService"),
            usageSummary=UsageDetails(
                dailyStats=UsageStats(count=5),
                date=datetime.strftime(datetime.now() - timedelta(1), "%Y-%m-%d"),
            ),
        )
        with patch.object(OpenMetadata, "get_by_name", return_value=return_value):
            self.assertEqual(
                next(self.tableau.yield_dashboard_usage(MOCK_DASHBOARD)).right,
                DashboardUsage(
                    dashboard=return_value,
                    usage=UsageRequest(date=self.tableau.today, count=5),
                ),
            )

        # If the past usage is higher than what we have today, something weird is going on
        # we don't return usage but don't explode
        return_value = Dashboard(
            id=uuid.uuid4(),
            name="dashboard_name",
            fullyQualifiedName="dashboard_service.dashboard_name",
            service=EntityReference(id=uuid.uuid4(), type="dashboardService"),
            usageSummary=UsageDetails(
                dailyStats=UsageStats(count=1000),
                date=datetime.strftime(datetime.now() - timedelta(1), "%Y-%m-%d"),
            ),
        )
        with patch.object(OpenMetadata, "get_by_name", return_value=return_value):
            self.assertEqual(
                len(list(self.tableau.yield_dashboard_usage(MOCK_DASHBOARD))), 1
            )

            self.assertIsNotNone(
                list(self.tableau.yield_dashboard_usage(MOCK_DASHBOARD))[0].left
            )

    def test_check_basemodel_returns_id_as_string(self):
        """
        Test that the basemodel returns the id as a string
        """
        base_model = TableauBaseModel(id=uuid.uuid4())
        self.assertEqual(base_model.id, str(base_model.id))

        base_model = TableauBaseModel(id="1234")
        self.assertEqual(base_model.id, "1234")

    def test_get_dashboard_project_filter(self):
        """
        Test get_dashboard filters dashboards based on projectFilterPattern
        """

        mock_dashboard_details_list = [
            TableauDashboard(
                id="dashboard1",
                name="dashboard1",
                project=TableauBaseModel(id="p1", name="FilteredProject"),
                charts=[],
                dataModels=[],
                tags=[],
            ),
            TableauDashboard(
                id="dashboard2",
                name="dashboard2",
                project=TableauBaseModel(id="p2", name="OtherProject"),
                charts=[],
                dataModels=[],
                tags=[],
            ),
            TableauDashboard(
                id="dashboard3",
                name="dashboard3",
                project=TableauBaseModel(id="p3", name="excludedDashboard"),
                charts=[],
                dataModels=[],
                tags=[],
            ),
            TableauDashboard(
                id="dashboard4",
                name="dashboard4",
                project=TableauBaseModel(id="p4", name="excludedDashboard"),
                charts=[],
                dataModels=[],
                tags=[],
            ),
        ]

        project_names_return_map = {
            "dashboard1": "FilteredProject.OtherProject",
            "dashboard2": "FilteredProject.OtherProject.ChildProject",
            "dashboard3": "AnFilteredProject.OtherProject.ChildProject",
            "dashboard4": "AnFilteredProject.OtherProject1.ChildProject2.ExcludedProject2",
        }

        self.tableau.source_config.projectFilterPattern = FilterPattern(
            includes=["^FilteredProject.OtherProject$"]
        )

        with patch.object(
            self.tableau,
            "get_dashboards_list",
            return_value=mock_dashboard_details_list,
        ):

            with patch.object(
                self.tableau,
                "get_project_names",
                side_effect=lambda dashboard_details: project_names_return_map[
                    dashboard_details.name
                ],
            ), patch.object(
                self.tableau,
                "get_dashboards_list",
                return_value=mock_dashboard_details_list,
            ), patch.object(
                self.tableau,
                "get_dashboard_details",
                side_effect=lambda x: x,
            ):
                dashboards = list(self.tableau.get_dashboard())
                self.assertEqual(len(dashboards), 1)
                self.assertEqual(dashboards[0].name, "dashboard1")

        # Test with other project names
        self.tableau.source_config.projectFilterPattern = FilterPattern(
            includes=[
                "^FilteredProject.OtherProject.*",
                "^AnFilteredProject.OtherProject.ChildProject$",
            ]
        )

        with patch.object(
            self.tableau,
            "get_dashboards_list",
            return_value=mock_dashboard_details_list,
        ):

            with patch.object(
                self.tableau,
                "get_project_names",
                side_effect=lambda dashboard_details: project_names_return_map[
                    dashboard_details.name
                ],
            ), patch.object(
                self.tableau,
                "get_dashboards_list",
                return_value=mock_dashboard_details_list,
            ), patch.object(
                self.tableau,
                "get_dashboard_details",
                side_effect=lambda x: x,
            ):
                dashboards = list(self.tableau.get_dashboard())
                self.assertEqual(len(dashboards), 3)
                self.assertEqual(dashboards[0].name, "dashboard1")
                self.assertEqual(dashboards[1].name, "dashboard2")
                self.assertEqual(dashboards[2].name, "dashboard3")

        # Test with includes and excludes

        self.tableau.source_config.projectFilterPattern = FilterPattern(
            includes=["^AnFilteredProject.OtherProject1.*"],
            excludes=[".*ExcludedProject2.*"],
        )

        with patch.object(
            self.tableau,
            "get_dashboards_list",
            return_value=mock_dashboard_details_list,
        ):

            with patch.object(
                self.tableau,
                "get_project_names",
                side_effect=lambda dashboard_details: project_names_return_map[
                    dashboard_details.name
                ],
            ), patch.object(
                self.tableau,
                "get_dashboards_list",
                return_value=mock_dashboard_details_list,
            ), patch.object(
                self.tableau,
                "get_dashboard_details",
                side_effect=lambda x: x,
            ):
                dashboards = list(self.tableau.get_dashboard())
                self.assertEqual(len(dashboards), 0)

    def test_generate_dashboard_url(self):
        """
        Test that the dashboard url is generated correctly with proxyURL
        """
        self.tableau.config.serviceConnection.root.config.proxyURL = (
            "http://mockTableauServer.com"
        )
        result = list(self.tableau.yield_dashboard(MOCK_DASHBOARD))
        self.assertEqual(
            result[0].right.sourceUrl.root,
            "http://mockTableauServer.com/#/site/hidarsite/workbooks/897790/views",
        )

    def _setup_ssl_config(self, verify_ssl_value="no-ssl", ssl_config=None):
        """
        Helper method to set up SSL configuration for testing
        """
        from types import SimpleNamespace

        from pydantic import SecretStr

        # Set up verifySSL
        self.tableau.config.serviceConnection.root.config.verifySSL = SimpleNamespace()
        self.tableau.config.serviceConnection.root.config.verifySSL.value = (
            verify_ssl_value
        )

        # Set up sslConfig if provided
        if ssl_config:
            self.tableau.config.serviceConnection.root.config.sslConfig = (
                SimpleNamespace()
            )
            self.tableau.config.serviceConnection.root.config.sslConfig.root = (
                SimpleNamespace()
            )

            if "caCertificate" in ssl_config:
                self.tableau.config.serviceConnection.root.config.sslConfig.root.caCertificate = SecretStr(
                    ssl_config["caCertificate"]
                )
            else:
                self.tableau.config.serviceConnection.root.config.sslConfig.root.caCertificate = (
                    None
                )

            if "sslCertificate" in ssl_config:
                self.tableau.config.serviceConnection.root.config.sslConfig.root.sslCertificate = SecretStr(
                    ssl_config["sslCertificate"]
                )
            else:
                self.tableau.config.serviceConnection.root.config.sslConfig.root.sslCertificate = (
                    None
                )

            if "sslKey" in ssl_config:
                self.tableau.config.serviceConnection.root.config.sslConfig.root.sslKey = SecretStr(
                    ssl_config["sslKey"]
                )
            else:
                self.tableau.config.serviceConnection.root.config.sslConfig.root.sslKey = (
                    None
                )
        else:
            self.tableau.config.serviceConnection.root.config.sslConfig = None

    def test_tableau_ssl_auth(self):
        """
        Test that Tableau SSL authentication works correctly
        """
        # Set up SSL configuration with all certificates
        self._setup_ssl_config(
            verify_ssl_value="validate",
            ssl_config={
                "caCertificate": "/path/to/ca.pem",
                "sslCertificate": "/path/to/cert.pem",
                "sslKey": "/path/to/key.pem",
            },
        )

        # Test that SSL configuration was set correctly
        self.assertEqual(
            self.tableau.config.serviceConnection.root.config.sslConfig.root.sslCertificate.get_secret_value(),
            "/path/to/cert.pem",
        )
        self.assertEqual(
            self.tableau.config.serviceConnection.root.config.sslConfig.root.sslKey.get_secret_value(),
            "/path/to/key.pem",
        )
        self.assertEqual(
            self.tableau.config.serviceConnection.root.config.sslConfig.root.caCertificate.get_secret_value(),
            "/path/to/ca.pem",
        )

        # Test SSL connection establishment
        with patch.object(
            self.tableau, "get_dashboards_list", return_value=[]
        ) as mock_get_dashboards:
            list(self.tableau.get_dashboard())
            mock_get_dashboards.assert_called_once()

    def test_tableau_ssl_auth_without_cert(self):
        """
        Test that Tableau SSL authentication works without client certificates
        """
        # Set up SSL configuration with only CA certificate
        self._setup_ssl_config(
            verify_ssl_value="validate", ssl_config={"caCertificate": "/path/to/ca.pem"}
        )

        # Verify SSL configuration was set correctly
        self.assertEqual(
            self.tableau.config.serviceConnection.root.config.sslConfig.root.caCertificate.get_secret_value(),
            "/path/to/ca.pem",
        )
        self.assertIsNone(
            self.tableau.config.serviceConnection.root.config.sslConfig.root.sslCertificate
        )
        self.assertIsNone(
            self.tableau.config.serviceConnection.root.config.sslConfig.root.sslKey
        )

        # Test SSL connection establishment
        with patch.object(
            self.tableau, "get_dashboards_list", return_value=[]
        ) as mock_get_dashboards:
            list(self.tableau.get_dashboard())
            mock_get_dashboards.assert_called_once()

    def test_tableau_ssl_auth_disabled(self):
        """
        Test that Tableau works correctly when SSL is disabled
        """
        # Set up SSL configuration with SSL disabled
        self._setup_ssl_config(verify_ssl_value="ignore")

        # Verify SSL verification is disabled
        self.assertEqual(
            self.tableau.config.serviceConnection.root.config.verifySSL.value, "ignore"
        )

        # Test SSL connection establishment
        with patch.object(
            self.tableau, "get_dashboards_list", return_value=[]
        ) as mock_get_dashboards:
            list(self.tableau.get_dashboard())
            mock_get_dashboards.assert_called_once()

    def test_get_datamodel_table_lineage_with_empty_from_entities(self):
        """
        Test that _get_datamodel_table_lineage handles empty from_entities gracefully
        """
        # Mock data for the test
        mock_datamodel = DataSource(
            id="datasource1",
            name="Test Datasource",
            upstreamDatasources=[
                DataSource(
                    id="upstream_datasource1",
                    name="Upstream Datasource",
                    upstreamTables=[
                        UpstreamTable(
                            id="table1",
                            luid="table1_luid",
                            name="test_table",
                            referencedByQueries=[
                                {
                                    "id": "query1",
                                    "name": "test_query",
                                    "query": "SELECT * FROM test_table",
                                }
                            ],
                        )
                    ],
                )
            ],
        )

        mock_data_model_entity = DashboardDataModel(
            id=uuid.uuid4(),
            name="Test Data Model",
            service=EntityReference(id=uuid.uuid4(), type="dashboardService"),
            dataModelType="TableauDataModel",
            columns=[],
        )

        mock_upstream_data_model_entity = DashboardDataModel(
            id=uuid.uuid4(),
            name="Upstream Data Model",
            service=EntityReference(id=uuid.uuid4(), type="dashboardService"),
            dataModelType="TableauDataModel",
            columns=[],
        )

        # Mock the client to return custom SQL queries
        self.tableau.client.get_custom_sql_table_queries = MagicMock(
            return_value=["SELECT * FROM test_table"]
        )

        # Mock the _get_datamodel method
        with patch.object(
            self.tableau, "_get_datamodel", return_value=mock_upstream_data_model_entity
        ):
            # Mock the metadata search to return empty results (simulating no table entities found)
            with patch.object(
                self.tableau.metadata, "search_in_any_service", return_value=[]
            ):
                # Mock the _get_add_lineage_request method to avoid actual lineage creation
                with patch.object(
                    self.tableau, "_get_add_lineage_request"
                ) as mock_lineage_request:
                    # Call the method under test
                    lineage_results = list(
                        self.tableau._get_datamodel_table_lineage(
                            datamodel=mock_datamodel,
                            data_model_entity=mock_data_model_entity,
                            db_service_prefix=None,
                        )
                    )

                    # Verify that the method completes without throwing an error
                    # Even though from_entities is empty, the method should handle it gracefully
                    self.assertIsInstance(lineage_results, list)

                    # Verify that the lineage request was called for the datamodel lineage
                    # (but not for table lineage since from_entities was empty)
                    mock_lineage_request.assert_called()

                    # Verify that the method didn't throw any exceptions
                    # The test passes if we reach this point without exceptions

    def test_get_datamodel_table_lineage_with_none_from_entities(self):
        """
        Test that _get_datamodel_table_lineage handles None from_entities gracefully
        """
        # Mock data for the test
        mock_datamodel = DataSource(
            id="datasource2",
            name="Test Datasource 2",
            upstreamDatasources=[
                DataSource(
                    id="upstream_datasource2",
                    name="Upstream Datasource 2",
                    upstreamTables=[
                        UpstreamTable(
                            id="table2",
                            luid="table2_luid",
                            name="test_table_2",
                            referencedByQueries=[
                                {
                                    "id": "query2",
                                    "name": "test_query_2",
                                    "query": "SELECT * FROM test_table_2",
                                }
                            ],
                        )
                    ],
                )
            ],
        )

        mock_data_model_entity = DashboardDataModel(
            id=uuid.uuid4(),
            name="Test Data Model 2",
            service=EntityReference(id=uuid.uuid4(), type="dashboardService"),
            dataModelType="TableauDataModel",
            columns=[],
        )

        mock_upstream_data_model_entity = DashboardDataModel(
            id=uuid.uuid4(),
            name="Upstream Data Model 2",
            service=EntityReference(id=uuid.uuid4(), type="dashboardService"),
            dataModelType="TableauDataModel",
            columns=[],
        )

        # Mock the client to return custom SQL queries
        self.tableau.client.get_custom_sql_table_queries = MagicMock(
            return_value=["SELECT * FROM test_table_2"]
        )

        # Mock the _get_datamodel method
        with patch.object(
            self.tableau, "_get_datamodel", return_value=mock_upstream_data_model_entity
        ):
            # Mock the metadata search to return None (simulating search failure)
            with patch.object(
                self.tableau.metadata, "search_in_any_service", return_value=None
            ):
                # Mock the _get_add_lineage_request method to avoid actual lineage creation
                with patch.object(
                    self.tableau, "_get_add_lineage_request"
                ) as mock_lineage_request:
                    # Call the method under test
                    lineage_results = list(
                        self.tableau._get_datamodel_table_lineage(
                            datamodel=mock_datamodel,
                            data_model_entity=mock_data_model_entity,
                            db_service_prefix=None,
                        )
                    )

                    # Verify that the method completes without throwing an error
                    # Even though from_entities is None, the method should handle it gracefully
                    self.assertIsInstance(lineage_results, list)

                    # Verify that the lineage request was called for the datamodel lineage
                    # (but not for table lineage since from_entities was None)
                    mock_lineage_request.assert_called()

                    # Verify that the method didn't throw any exceptions
                    # The test passes if we reach this point without exceptions