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_index(config):
""" Test the state index """
state = wily.state.State(config)
assert state.index
assert state.index["git"] is not None
last_revision = state.index["git"].last_revision
assert last_revision.revision.message == "remove line"
for revision in state.index["git"].revisions:
assert state.index["git"][revision.revision.key]
assert revision.revision in state.index["git"]
assert revision.revision.key in state.index["git"]
def test_state_defaults(config):
""" Test the state defaults """
state = wily.state.State(config)
assert state.index
assert "git" in state.index
assert state.default_archiver == "git"
assert state.config is config
:param revision_index: Version of git repository to revert to.
:type revision_index: ``str``
:param limit: Limit the number of items in the table
:type limit: ``int``
:return: Sorted table of all files in path, sorted in order of metric.
"""
logger.debug("Running rank command")
data = []
operator, metric = resolve_metric_as_tuple(metric)
operator = operator.name
state = State(config)
if not revision_index:
target_revision = state.index[state.default_archiver].last_revision
else:
rev = resolve_archiver(state.default_archiver).cls(config).find(revision_index)
logger.debug(f"Resolved {revision_index} to {rev.key} ({rev.message})")
try:
target_revision = state.index[state.default_archiver][rev.key]
except KeyError:
logger.error(
f"Revision {revision_index} is not in the cache, make sure you have run wily build."
)
exit(1)
logger.info(
f"-----------Rank for {metric.description} for {format_revision(target_revision.revision.key)} by {target_revision.revision.author_name} on {format_date(target_revision.revision.date)}.------------"
logger.debug(f"Using {archiver.name} archiver module")
_archiver = archiver.cls(config)
revisions = _archiver.revisions(config.path, config.max_revisions)
except InvalidGitRepositoryError:
# TODO: This logic shouldn't really be here (SoC)
logger.info(f"Defaulting back to the filesystem archiver, not a valid git repo")
_archiver = FilesystemArchiver(config)
revisions = _archiver.revisions(config.path, config.max_revisions)
except Exception as e:
if hasattr(e, "message"):
logger.error(f"Failed to setup archiver: '{e.message}'")
else:
logger.error(f"Failed to setup archiver: '{type(e)} - {e}'")
exit(1)
state = State(config, archiver=_archiver)
# Check for existence of cache, else provision
state.ensure_exists()
index = state.get_index(_archiver)
# remove existing revisions from the list
revisions = [revision for revision in revisions if revision not in index][::-1]
logger.info(
f"Found {len(revisions)} revisions from '{_archiver.name}' archiver in '{config.path}'."
)
_op_desc = ",".join([operator.name for operator in operators])
logger.info(f"Running operators - {_op_desc}")
bar = Bar("Processing", max=len(revisions) * len(operators))
:param metrics: The metrics to measure.
:type metrics: ``list`` of ``str``
:param changes_only: Only include changes files in output.
:type changes_only: ``bool``
:param detail: Show details (function-level)
:type detail: ``bool``
:param revision: Compare with specific revision
:type revision: ``str``
"""
config.targets = files
files = list(files)
state = State(config)
# Resolve target paths when the cli has specified --path
if config.path != DEFAULT_PATH:
targets = [str(Path(config.path) / Path(file)) for file in files]
else:
targets = files
# Expand directories to paths
files = [
os.path.relpath(fn, config.path)
for fn in radon.cli.harvest.iter_filenames(targets)
]
logger.debug(f"Targeting - {files}")
if not revision:
target_revision = state.index[state.default_archiver].last_revision
:param config: The configuration.
:type config: :class:`wily.config.WilyConfig`
:param path: The path to the files.
:type path: ``list``
:param metrics: The Y and Z-axis metrics to report on.
:type metrics: ``tuple``
:param output: Save report to specified path instead of opening browser.
:type output: ``str``
"""
logger.debug("Running report command")
data = []
state = State(config)
abs_path = config.path / pathlib.Path(path)
if x_axis is None:
x_axis = "history"
else:
x_operator, x_key = metric_parts(x_axis)
if abs_path.is_dir():
paths = [
p.relative_to(config.path) for p in pathlib.Path(abs_path).glob("**/*.py")
]
else:
paths = [path]
operator, key = metric_parts(metrics[0])
if len(metrics) == 1: # only y-axis
def index(config, include_message=False):
"""
Show information about the cache and runtime.
:param config: The wily configuration
:type config: :namedtuple:`wily.config.WilyConfig`
:param include_message: Include revision messages
:type include_message: ``bool``
"""
state = State(config=config)
logger.debug("Running show command")
logger.info("--------Configuration---------")
logger.info(f"Path: {config.path}")
logger.info(f"Archiver: {config.archiver}")
logger.info(f"Operators: {config.operators}")
logger.info("")
logger.info("-----------History------------")
data = []
for archiver in state.archivers:
for rev in state.index[archiver].revisions[::-1]:
if include_message:
data.append(
(
format_revision(rev.revision.key),
rev.revision.author_name,
def index(config, include_message=False):
"""
Show information about the cache and runtime.
:param config: The wily configuration
:type config: :namedtuple:`wily.config.WilyConfig`
:param include_message: Include revision messages
:type include_message: ``bool``
"""
state = State(config=config)
logger.debug("Running show command")
logger.info("--------Configuration---------")
logger.info(f"Path: {config.path}")
logger.info(f"Archiver: {config.archiver}")
logger.info(f"Operators: {config.operators}")
logger.info("")
logger.info("-----------History------------")
data = []
for archiver in state.archivers:
for rev in state.index[archiver].revisions:
if include_message:
data.append(
(
format_revision(rev.revision.key),
rev.revision.author_name,
good_color = 31
bad_color = 32
elif metric.measure == MetricType.Informational:
good_color = 33
bad_color = 33
metric_meta = {
"key": key,
"operator": operator,
"good_color": good_color,
"bad_color": bad_color,
"title": metric.description,
"type": metric.type,
}
metric_metas.append(metric_meta)
state = State(config)
for archiver in state.archivers:
history = state.index[archiver].revisions[:n][::-1]
last = {}
for rev in history:
vals = []
for meta in metric_metas:
try:
logger.debug(
f"Fetching metric {meta['key']} for {meta['operator']} in {path}"
)
val = rev.get(config, archiver, meta["operator"], path, meta["key"])
last_val = last.get(meta["key"], None)
# Measure the difference between this value and the last
if meta["type"] in (int, float):
if last_val: