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_exists(tmpdir):
"""
Test that exists() returns true if path does exist
"""
config = DEFAULT_CONFIG
config.cache_path = tmpdir
assert cache.exists(config)
def test_clean_when_not_exists(tmpdir):
"""
Test that clean() will continue if the folder does not exist
"""
config = DEFAULT_CONFIG
cache_path = pathlib.Path(tmpdir) / ".wily"
config.cache_path = str(cache_path)
assert not cache.exists(config)
assert cache.clean(config) is None
def ensure_exists(self):
"""Ensure that cache directory exists."""
if not cache.exists(self.config):
logger.debug("Wily cache not found, creating.")
cache.create(self.config)
logger.debug("Created wily cache")
else:
logger.debug(f"Cache {self.config.cache_path} exists")
def save(self):
"""Save the index data back to the wily cache."""
data = [i.asdict() for i in self._revisions.values()]
logger.debug("Saving data")
cache.store_archiver_index(self.config, self.archiver, data)
def __init__(self, config, archiver=None):
"""
Instantiate a new process state.
:param config: The wily configuration.
:type config: :class:`WilyConfig`
:param archiver: The archiver (optional).
:type archiver: :class:`wily.archivers.Archiver`
"""
if archiver:
self.archivers = [archiver.name]
else:
self.archivers = cache.list_archivers(config)
logger.debug(f"Initialised state indexes for archivers {self.archivers}")
self.config = config
self.index = {}
for archiver in self.archivers:
self.index[archiver] = Index(self.config, resolve_archiver(archiver))
self.default_archiver = self.archivers[0]
def ensure_exists(self):
"""Ensure that cache directory exists."""
if not cache.exists(self.config):
logger.debug("Wily cache not found, creating.")
cache.create(self.config)
logger.debug("Created wily cache")
else:
logger.debug(f"Cache {self.config.cache_path} exists")
def __init__(self, config, archiver):
"""
Instantiate a new index.
:param config: The wily config.
:type config: :class:`wily.config.WilyConfig`
:param archiver: The archiver.
:type archiver: :class:`wily.archivers.Archiver`
"""
self.config = config
self.archiver = archiver
self.data = (
cache.get_archiver_index(config, archiver.name)
if cache.has_archiver_index(config, archiver.name)
else []
)
self._revisions = OrderedDict(
{d["key"]: IndexedRevision.fromdict(d) for d in self.data}
)
def __init__(self, config, archiver):
"""
Instantiate a new index.
:param config: The wily config.
:type config: :class:`wily.config.WilyConfig`
:param archiver: The archiver.
:type archiver: :class:`wily.archivers.Archiver`
"""
self.config = config
self.archiver = archiver
self.data = (
cache.get_archiver_index(config, archiver.name)
if cache.has_archiver_index(config, archiver.name)
else []
)
self._revisions = OrderedDict(
{d["key"]: IndexedRevision.fromdict(d) for d in self.data}
)
:type config: :class:`wily.config.WilyConfig`
:param archiver: The archiver.
:type archiver: :class:`wily.archivers.Archiver`
:param operator: The operator to find
:type operator: ``str``
:param path: The path to find
:type path: ``str``
:param key: The metric key
:type key: ``str``
"""
if not self._data:
self._data = cache.get(
config=config, archiver=archiver, revision=self.revision.key
)["operator_data"]
logger.debug(f"Fetching metric {path} - {key} for operator {operator}")
return get_metric(self._data, operator, path, key)