Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_from_conn_not_registered():
"""
Tests helpful error message on attempt to choose unregistered conn type.
"""
conn = Mock()
conn.__class__ = "Not a real class"
with pytest.raises(ETLHelperHelperError,
match=r'Unsupported connection type.*'):
DB_HELPER_FACTORY.from_conn(conn)
def test_from_db_params_not_registered():
"""
Tests helpful error message on attempt to choose unregistered db_params
type.
"""
db_params = MagicMock(DbParams)
db_params.dbtype = 'Not a real type'
with pytest.raises(ETLHelperHelperError,
match=r'Unsupported DbParams.dbtype.*'):
DB_HELPER_FACTORY.from_db_params(db_params)
def from_dbtype(self, dbtype):
"""
Return initialised db helper based on type
"""
try:
helper = self.helpers[dbtype]()
except KeyError:
msg = f"Unsupported DbParams.dbtype: {dbtype}"
raise ETLHelperHelperError(msg)
return helper
def validate_params(self):
"""
Validate database parameters.
Should validate that a dbtype is a valid one and that the appropriate
params have been passed for a particular db_type.
:raises ETLHelperParamsError: Error if params are invalid
"""
# Get a set of the attributes to compare against required attributes.
given = set(self.keys())
try:
required_params = DB_HELPER_FACTORY.from_dbtype(self.dbtype).required_params
except ETLHelperHelperError:
msg = f'{self.dbtype} not in valid types ({DB_HELPER_FACTORY.helpers.keys()})'
raise ETLHelperDbParamsError(msg)
unset_params = (given ^ required_params) & required_params
if unset_params:
msg = f'{unset_params} not set. Required parameters are {required_params}'
raise ETLHelperDbParamsError(msg)
def from_conn(self, conn):
"""
Return initialised db_helper based on connection.
"""
if not hasattr(conn, 'cursor'):
msg = f"Expected connection-like object, got {type(conn)}"
raise ETLHelperHelperError(msg)
try:
conn_type = str(conn.__class__)
dbtype = self._conn_types[conn_type]
except KeyError:
msg = f"Unsupported connection type: {conn_type}"
raise ETLHelperHelperError(msg)
return self.from_dbtype(dbtype)