mirror of
https://github.com/datahub-project/datahub.git
synced 2025-07-07 09:11:47 +00:00
41 lines
1.4 KiB
Python
41 lines
1.4 KiB
Python
|
from unittest.mock import MagicMock, patch
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from datahub.ingestion.source.sql.mssql.source import SQLServerConfig, SQLServerSource
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def mssql_source():
|
||
|
|
config = SQLServerConfig(
|
||
|
|
host_port="localhost:1433",
|
||
|
|
username="test",
|
||
|
|
password="test",
|
||
|
|
database="test_db",
|
||
|
|
temporary_tables_pattern=["^temp_"],
|
||
|
|
include_descriptions=False, # Disable description loading to avoid DB connections
|
||
|
|
)
|
||
|
|
|
||
|
|
# Mock the parent class's __init__ to avoid DB connections
|
||
|
|
with patch("datahub.ingestion.source.sql.sql_common.SQLAlchemySource.__init__"):
|
||
|
|
source = SQLServerSource(config, MagicMock())
|
||
|
|
source.discovered_datasets = {"test_db.dbo.regular_table"}
|
||
|
|
return source
|
||
|
|
|
||
|
|
|
||
|
|
def test_is_temp_table(mssql_source):
|
||
|
|
# Test tables matching temporary table patterns
|
||
|
|
assert mssql_source.is_temp_table("test_db.dbo.temp_table") is True
|
||
|
|
|
||
|
|
# Test tables starting with # (handled by startswith check in is_temp_table)
|
||
|
|
assert mssql_source.is_temp_table("test_db.dbo.#some_table") is True
|
||
|
|
|
||
|
|
# Test tables that are not in discovered_datasets
|
||
|
|
assert mssql_source.is_temp_table("test_db.dbo.unknown_table") is True
|
||
|
|
|
||
|
|
# Test regular tables that should return False
|
||
|
|
assert mssql_source.is_temp_table("test_db.dbo.regular_table") is False
|
||
|
|
|
||
|
|
# Test invalid table name format
|
||
|
|
assert mssql_source.is_temp_table("invalid_table_name") is False
|