How to use get_hook method in Lemoncheesecake

Best Python code snippet using lemoncheesecake

test_odbc.py

Source:test_odbc.py

...22import pyodbc23from airflow.models import Connection24from airflow.providers.odbc.hooks.odbc import OdbcHook25class TestOdbcHook:26    def get_hook(self=None, hook_params=None, conn_params=None):27        hook_params = hook_params or {}28        conn_params = conn_params or {}29        connection = Connection(30            **{31                **dict(login='login', password='password', host='host', schema='schema', port=1234),32                **conn_params,33            }34        )35        hook = OdbcHook(**hook_params)36        hook.get_connection = mock.Mock()37        hook.get_connection.return_value = connection38        return hook39    def test_driver_in_extra(self):40        conn_params = dict(extra=json.dumps(dict(Driver='Fake Driver', Fake_Param='Fake Param')))41        hook = self.get_hook(conn_params=conn_params)42        expected = (43            'DRIVER={Fake Driver};'44            'SERVER=host;'45            'DATABASE=schema;'46            'UID=login;'47            'PWD=password;'48            'PORT=1234;'49            'Fake_Param=Fake Param;'50        )51        assert hook.odbc_connection_string == expected52    def test_driver_in_both(self):53        conn_params = dict(extra=json.dumps(dict(Driver='Fake Driver', Fake_Param='Fake Param')))54        hook_params = dict(driver='ParamDriver')55        hook = self.get_hook(hook_params=hook_params, conn_params=conn_params)56        expected = (57            'DRIVER={ParamDriver};'58            'SERVER=host;'59            'DATABASE=schema;'60            'UID=login;'61            'PWD=password;'62            'PORT=1234;'63            'Fake_Param=Fake Param;'64        )65        assert hook.odbc_connection_string == expected66    def test_dsn_in_extra(self):67        conn_params = dict(extra=json.dumps(dict(DSN='MyDSN', Fake_Param='Fake Param')))68        hook = self.get_hook(conn_params=conn_params)69        expected = (70            'DSN=MyDSN;SERVER=host;DATABASE=schema;UID=login;PWD=password;PORT=1234;Fake_Param=Fake Param;'71        )72        assert hook.odbc_connection_string == expected73    def test_dsn_in_both(self):74        conn_params = dict(extra=json.dumps(dict(DSN='MyDSN', Fake_Param='Fake Param')))75        hook_params = dict(driver='ParamDriver', dsn='ParamDSN')76        hook = self.get_hook(hook_params=hook_params, conn_params=conn_params)77        expected = (78            'DRIVER={ParamDriver};'79            'DSN=ParamDSN;'80            'SERVER=host;'81            'DATABASE=schema;'82            'UID=login;'83            'PWD=password;'84            'PORT=1234;'85            'Fake_Param=Fake Param;'86        )87        assert hook.odbc_connection_string == expected88    def test_get_uri(self):89        conn_params = dict(extra=json.dumps(dict(DSN='MyDSN', Fake_Param='Fake Param')))90        hook_params = dict(dsn='ParamDSN')91        hook = self.get_hook(hook_params=hook_params, conn_params=conn_params)92        uri_param = quote_plus(93            'DSN=ParamDSN;SERVER=host;DATABASE=schema;UID=login;PWD=password;PORT=1234;Fake_Param=Fake Param;'94        )95        expected = 'mssql+pyodbc:///?odbc_connect=' + uri_param96        assert hook.get_uri() == expected97    def test_connect_kwargs_from_hook(self):98        hook = self.get_hook(99            hook_params=dict(100                connect_kwargs={101                    'attrs_before': {102                        1: 2,103                        pyodbc.SQL_TXN_ISOLATION: pyodbc.SQL_TXN_READ_UNCOMMITTED,104                    },105                    'readonly': True,106                    'autocommit': False,107                }108            ),109        )110        assert hook.connect_kwargs == {111            'attrs_before': {1: 2, pyodbc.SQL_TXN_ISOLATION: pyodbc.SQL_TXN_READ_UNCOMMITTED},112            'readonly': True,113            'autocommit': False,114        }115    def test_connect_kwargs_from_conn(self):116        extra = json.dumps(117            dict(118                connect_kwargs={119                    'attrs_before': {120                        1: 2,121                        pyodbc.SQL_TXN_ISOLATION: pyodbc.SQL_TXN_READ_UNCOMMITTED,122                    },123                    'readonly': True,124                    'autocommit': True,125                }126            )127        )128        hook = self.get_hook(conn_params=dict(extra=extra))129        assert hook.connect_kwargs == {130            'attrs_before': {1: 2, pyodbc.SQL_TXN_ISOLATION: pyodbc.SQL_TXN_READ_UNCOMMITTED},131            'readonly': True,132            'autocommit': True,133        }134    def test_connect_kwargs_from_conn_and_hook(self):135        """136        When connect_kwargs in both hook and conn, should be merged properly.137        Hook beats conn.138        """139        conn_extra = json.dumps(dict(connect_kwargs={'attrs_before': {1: 2, 3: 4}, 'readonly': False}))140        hook_params = dict(141            connect_kwargs={'attrs_before': {3: 5, pyodbc.SQL_TXN_ISOLATION: 0}, 'readonly': True}142        )143        hook = self.get_hook(conn_params=dict(extra=conn_extra), hook_params=hook_params)144        assert hook.connect_kwargs == {145            'attrs_before': {1: 2, 3: 5, pyodbc.SQL_TXN_ISOLATION: 0},146            'readonly': True,147        }148    def test_connect_kwargs_bool_from_uri(self):149        """150        Bools will be parsed from uri as strings151        """152        conn_extra = json.dumps(dict(connect_kwargs={'ansi': True}))153        hook = self.get_hook(conn_params=dict(extra=conn_extra))154        assert hook.connect_kwargs == {155            'ansi': True,156        }157    def test_driver(self):158        hook = self.get_hook(hook_params=dict(driver='Blah driver'))159        assert hook.driver == 'Blah driver'160        hook = self.get_hook(hook_params=dict(driver='{Blah driver}'))161        assert hook.driver == 'Blah driver'162        hook = self.get_hook(conn_params=dict(extra='{"driver": "Blah driver"}'))163        assert hook.driver == 'Blah driver'164        hook = self.get_hook(conn_params=dict(extra='{"driver": "{Blah driver}"}'))165        assert hook.driver == 'Blah driver'166    def test_database(self):167        hook = self.get_hook(hook_params=dict(database='abc'))168        assert hook.database == 'abc'169        hook = self.get_hook()170        assert hook.database == 'schema'171    def test_sqlalchemy_scheme_default(self):172        hook = self.get_hook()173        uri = hook.get_uri()174        assert urlparse(uri).scheme == 'mssql+pyodbc'175    def test_sqlalchemy_scheme_param(self):176        hook = self.get_hook(hook_params=dict(sqlalchemy_scheme='my-scheme'))177        uri = hook.get_uri()178        assert urlparse(uri).scheme == 'my-scheme'179    def test_sqlalchemy_scheme_extra(self):180        hook = self.get_hook(conn_params=dict(extra=json.dumps(dict(sqlalchemy_scheme='my-scheme'))))181        uri = hook.get_uri()...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.