How to use the etlhelper.db_helpers.OracleDbHelper function in etlhelper

To help you get started, we’ve selected a few etlhelper examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github BritishGeologicalSurvey / etlhelper / test / unit / test_helper_factory.py View on Github external
                         [(OracleDbHelper, cx_Oracle.Connection),
                          (PostgresDbHelper, psycopg2.extensions.connection),
                          (MSSQLDbHelper, pyodbc.Connection)])
def test_from_conn(expected_helper, db_class):
    """
    Tests correct helper produced given a conn object
    """
    conn = Mock()
    # conn.__class__ = cx_Oracle.Connection
    conn.__class__ = db_class
    helper = DB_HELPER_FACTORY.from_conn(conn)
    assert isinstance(helper, expected_helper)
github BritishGeologicalSurvey / etlhelper / test / unit / test_db_helpers.py View on Github external
def test_oracle_sql_exceptions():
    helper = OracleDbHelper()
    assert helper.sql_exceptions == (cx_Oracle.DatabaseError)
github BritishGeologicalSurvey / etlhelper / test / unit / test_db_helpers.py View on Github external
def test_oracle_connect(monkeypatch):
    # Arrange
    # TODO: Fix DbParams class to take driver as init input.
    db_params = DbParams(dbtype='ORACLE',
                         host='server', port='1521', dbname='testdb',
                         user='testuser')
    monkeypatch.setenv('DB_PASSWORD', 'mypassword')
    expected_conn_str = 'testuser/mypassword@server:1521/testdb'

    mock_connect = Mock()
    monkeypatch.setattr(cx_Oracle, 'connect', mock_connect)

    # Act
    helper = OracleDbHelper()
    helper.connect(db_params, 'DB_PASSWORD')

    # Assert
    mock_connect.assert_called_with(expected_conn_str)
github BritishGeologicalSurvey / etlhelper / test / unit / test_db_helpers.py View on Github external
def test_oracle_connect_exceptions():
    helper = OracleDbHelper()
    assert helper.connect_exceptions == (cx_Oracle.DatabaseError)
github BritishGeologicalSurvey / etlhelper / test / unit / test_db_helpers.py View on Github external
def test_oracle_sqlalchemy_conn_string(monkeypatch):
    db_params = DbParams(dbtype='ORACLE',
                         host='server', port='1521', dbname='testdb',
                         user='testuser')
    monkeypatch.setenv('DB_PASSWORD', 'mypassword')
    helper = OracleDbHelper()
    conn_str = helper.get_sqlalchemy_connection_string(db_params, 'DB_PASSWORD')
    expected_conn_str = ('oracle://testuser:mypassword@server:1521/testdb')

    assert conn_str == expected_conn_str