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_creation_invalid(driver_path, provider):
from terracotta import drivers
from terracotta.exceptions import InvalidDatabaseError
db = drivers.get_driver(driver_path, provider=provider)
keys = ('invalid keyname',)
with pytest.raises(ValueError, InvalidDatabaseError):
db.create(keys)
db = drivers.get_driver(driver_path, provider=provider)
keys = ('some', 'keynames')
db.create(keys)
db.insert(['some', 'value'], str(raster_file))
# works
with db.connect():
pass
with monkeypatch.context() as m:
fake_version = '0.0.0'
m.setattr(f'{db.__module__}.__version__', fake_version)
db._version_checked = False
with pytest.raises(exceptions.InvalidDatabaseError) as exc:
with db.connect():
pass
assert fake_version in str(exc.value)
def _connection_callback(self) -> None:
"""Called after opening a new connection"""
# check for version compatibility
def versiontuple(version_string: str) -> Sequence[str]:
return version_string.split('.')
db_version = self.db_version
current_version = __version__
if versiontuple(db_version)[:2] != versiontuple(current_version)[:2]:
raise exceptions.InvalidDatabaseError(
f'Version conflict: database was created in v{db_version}, '
f'but this is v{current_version}'
def _connection_callback(self) -> None:
if not self._version_checked:
# check for version compatibility
def versiontuple(version_string: str) -> Sequence[str]:
return version_string.split('.')
db_version = self.db_version
current_version = __version__
if versiontuple(db_version)[:2] != versiontuple(current_version)[:2]:
raise exceptions.InvalidDatabaseError(
f'Version conflict: database was created in v{db_version}, '
f'but this is v{current_version}'
)
self._version_checked = True
def convert_exceptions(msg: str) -> Iterator:
"""Convert internal mysql exceptions to our InvalidDatabaseError"""
from pymysql import OperationalError, InternalError, ProgrammingError
try:
yield
except (OperationalError, InternalError, ProgrammingError) as exc:
raise exceptions.InvalidDatabaseError(msg) from exc