|
| 1 | +import pytest |
| 2 | + |
| 3 | +import dbt_common.exceptions |
| 4 | +from dbt.adapters.redshift.impl import RedshiftAdapter |
| 5 | + |
| 6 | + |
| 7 | +@pytest.fixture |
| 8 | +def adapter(mocker): |
| 9 | + mock_config = mocker.MagicMock() |
| 10 | + mock_config.credentials.database = "dev" |
| 11 | + mock_config.credentials.ra3_node = False |
| 12 | + mock_mp_context = mocker.MagicMock() |
| 13 | + adapter = RedshiftAdapter(mock_config, mock_mp_context) |
| 14 | + adapter.use_show_apis = lambda: False |
| 15 | + return adapter |
| 16 | + |
| 17 | + |
| 18 | +class TestVerifyDatabase: |
| 19 | + """Tests for verify_database cross-database reference gating.""" |
| 20 | + |
| 21 | + def test_same_database_always_passes(self, adapter): |
| 22 | + assert adapter.verify_database("dev") == "" |
| 23 | + |
| 24 | + def test_same_database_case_insensitive(self, adapter): |
| 25 | + assert adapter.verify_database("DEV") == "" |
| 26 | + |
| 27 | + def test_quoted_database_strips_quotes(self, adapter): |
| 28 | + assert adapter.verify_database('"dev"') == "" |
| 29 | + |
| 30 | + def test_cross_db_blocked_by_default(self, adapter): |
| 31 | + with pytest.raises(dbt_common.exceptions.NotImplementedError, match="Cross-db"): |
| 32 | + adapter.verify_database("other_db") |
| 33 | + |
| 34 | + def test_cross_db_allowed_with_ra3_node(self, adapter): |
| 35 | + adapter.config.credentials.ra3_node = True |
| 36 | + assert adapter.verify_database("other_db") == "" |
| 37 | + |
| 38 | + def test_cross_db_allowed_with_use_show_apis(self, adapter): |
| 39 | + adapter.use_show_apis = lambda: True |
| 40 | + assert adapter.verify_database("other_db") == "" |
| 41 | + |
| 42 | + def test_cross_db_allowed_with_both_flags(self, adapter): |
| 43 | + adapter.config.credentials.ra3_node = True |
| 44 | + adapter.use_show_apis = lambda: True |
| 45 | + assert adapter.verify_database("other_db") == "" |
| 46 | + |
| 47 | + def test_cross_db_blocked_without_either_flag(self, adapter): |
| 48 | + adapter.config.credentials.ra3_node = False |
| 49 | + adapter.use_show_apis = lambda: False |
| 50 | + with pytest.raises(dbt_common.exceptions.NotImplementedError, match="Cross-db"): |
| 51 | + adapter.verify_database("other_db") |
0 commit comments