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_lib_repr():
mongo = MagicMock()
with patch('arctic.arctic.ArcticLibraryBinding._auth'):
ml = ArcticLibraryBinding(mongo, 'asdf')
assert str(ml) == repr(ml)
def test_check_quota_info():
self = create_autospec(ArcticLibraryBinding, database_name='arctic_db',
library='lib')
self.arctic = create_autospec(Arctic)
self.get_library_metadata.return_value = 1024 * 1024 * 1024
self.quota_countdown = 0
self.arctic.__getitem__.return_value = Mock(stats=Mock(return_value={'totals':
{'size': 1 * 1024 * 1024,
'count': 100,
}
}))
with patch('arctic.arctic.logger.info') as info:
ArcticLibraryBinding.check_quota(self)
self.arctic.__getitem__.assert_called_once_with(self.get_name.return_value)
info.assert_called_once_with('Mongo Quota: arctic_db.lib 0.001 / 1 GB used')
assert self.quota_countdown == 51153
def test_enable_sharding():
arctic_lib = create_autospec(ArcticLibraryBinding)
arctic_lib.arctic = create_autospec(Arctic)
with patch('arctic.store.bson_store.enable_sharding', autospec=True) as enable_sharding:
arctic_lib.get_top_level_collection.return_value.database.create_collection.__name__ = 'some_name'
arctic_lib.get_top_level_collection.return_value.database.collection_names.__name__ = 'some_name'
bsons = BSONStore(arctic_lib)
bsons.enable_sharding()
# Check we always set the sharding to be hashed.
assert enable_sharding.call_args_list == [call(arctic_lib.arctic, arctic_lib.get_name(), hashed=True, key='_id')]
def get_quota(self):
"""
Get the current quota on this user library.
"""
return self.get_library_metadata(ArcticLibraryBinding.QUOTA)
def get_library_type(self):
return self.get_library_metadata(ArcticLibraryBinding.TYPE_FIELD)
def delete_library(self, library):
"""
Delete an Arctic Library, and all associated collections in the MongoDB.
Parameters
----------
library : `str`
The name of the library. e.g. 'library' or 'user.library'
"""
lib = ArcticLibraryBinding(self, library)
colname = lib.get_top_level_collection().name
if not [c for c in lib._db.list_collection_names(False) if re.match(r"^{}([\.].*)?$".format(colname), c)]:
logger.info('Nothing to delete. Arctic library %s does not exist.' % colname)
logger.info('Dropping collection: %s' % colname)
lib._db.drop_collection(colname)
for coll in lib._db.list_collection_names():
if coll.startswith(colname + '.'):
logger.info('Dropping collection: %s' % coll)
lib._db.drop_collection(coll)
if library in self._library_cache:
del self._library_cache[library]
del self._library_cache[lib.get_name()]
self._cache.delete_item_from_key('list_libraries', self._sanitize_lib_name(library))
def set_quota(self, library, quota):
"""
Set a quota (in bytes) on this user library. The quota is 'best effort',
and should be set conservatively.
Parameters
----------
library : `str`
The name of the library. e.g. 'library' or 'user.library'
quota : `int`
Advisory quota for the library - in bytes
"""
ArcticLibraryBinding(self, library).set_quota(quota)
def get_library_type(self, lib):
"""
Returns the type of the library
Parameters
----------
lib: str
the library
"""
return ArcticLibraryBinding(self, lib).get_library_type()
def get_library(self, library):
"""
Return the library instance. Can generally use slicing to return the library:
arctic_store[library]
Parameters
----------
library : `str`
The name of the library. e.g. 'library' or 'user.library'
"""
if library in self._library_cache:
return self._library_cache[library]
try:
error = None
lib = ArcticLibraryBinding(self, library)
lib_type = lib.get_library_type()
except (OperationFailure, AutoReconnect) as e:
error = e
if error:
raise LibraryNotFoundException("Library %s was not correctly initialized in %s.\nReason: %r)" %
(library, self, error))
elif not lib_type:
raise LibraryNotFoundException("Library %s was not correctly initialized in %s." %
(library, self))
elif lib_type not in LIBRARY_TYPES:
raise LibraryNotFoundException("Couldn't load LibraryType '%s' for '%s' (has the class been registered?)" %
(lib_type, library))
instance = LIBRARY_TYPES[lib_type](lib)
self._library_cache[library] = instance
# The library official name may be different from 'library': e.g. 'library' vs 'user.library'