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_store_relative_paths(tmpdir):
"""
Test that the store command works when absolute paths are used for the targets..
"""
config = DEFAULT_CONFIG
cache_path = pathlib.Path(tmpdir) / ".wily"
target_path = str(pathlib.Path(tmpdir) / "foo" / "bar.py")
cache_path.mkdir()
config.cache_path = cache_path
config.path = tmpdir
_TEST_STATS = {"operator_data": {"test": {target_path: {"metric1": 1}}}}
_TEST_REVISION = Revision(
key="12345",
author_name="Anthony Shaw",
author_email="anthony@test.com",
date="17/01/1990",
message="my changes",
files=[target_path],
)
fn = cache.store(config, ARCHIVER_GIT, _TEST_REVISION, _TEST_STATS)
with open(fn) as cache_item:
result = json.load(cache_item)
assert isinstance(result, dict)
if sys.platform == "win32":
assert "foo\\bar.py" in result["operator_data"]["test"].keys()
else:
assert "foo/bar.py" in result["operator_data"]["test"].keys()
def revisions(self, path, max_revisions):
return [
Revision(
key="12345",
author_name="Local User", # Don't want to leak local data
author_email="-", # as above
date=12_345_679,
message="None",
files=[],
)
def test_store_twice(tmpdir):
""" Test that you can't write the same revision twice """
config = DEFAULT_CONFIG
cache_path = pathlib.Path(tmpdir) / ".wily"
cache_path.mkdir()
config.cache_path = cache_path
target_path = str(pathlib.Path(tmpdir) / "foo" / "bar.py")
_TEST_STATS = {"operator_data": {"test": {target_path: {"metric1": 1}}}}
_TEST_REVISION = Revision(
key="12345",
author_name="Anthony Shaw",
author_email="anthony@test.com",
date="17/01/1990",
message="my changes",
files=[target_path],
)
fn = cache.store(config, ARCHIVER_GIT, _TEST_REVISION, _TEST_STATS)
with pytest.raises(RuntimeError):
cache.store(config, ARCHIVER_GIT, _TEST_REVISION, _TEST_STATS)
def test_store_basic(tmpdir):
config = DEFAULT_CONFIG
cache_path = pathlib.Path(tmpdir) / ".wily"
cache_path.mkdir()
config.cache_path = cache_path
target_path = str(pathlib.Path(tmpdir) / "foo" / "bar.py")
_TEST_STATS = {"operator_data": {"test": {target_path: {"metric1": 1}}}}
_TEST_REVISION = Revision(
key="12345",
author_name="Anthony Shaw",
author_email="anthony@test.com",
date="17/01/1990",
message="my changes",
files=[target_path],
)
fn = cache.store(config, ARCHIVER_GIT, _TEST_REVISION, _TEST_STATS)
with open(fn) as cache_item:
result = json.load(cache_item)
assert isinstance(result, dict)
assert result == _TEST_STATS
def find(self, search):
"""
Search a string and return a single revision.
:param search: The search term.
:type search: ``str``
:return: An instance of revision.
:rtype: Instance of :class:`Revision`
"""
commit = self.repo.commit(search)
return Revision(
key=commit.name_rev.split(" ")[0],
author_name=commit.author.name,
author_email=commit.author.email,
date=commit.committed_date,
message=commit.message,
files=list(commit.stats.files.keys()),
)
:type path: ``str``
:param max_revisions: the maximum number of revisions.
:type max_revisions: ``int``
:return: A list of revisions.
:rtype: ``list`` of :class:`Revision`
"""
if self.repo.is_dirty():
raise DirtyGitRepositoryError(self.repo.untracked_files)
revisions = []
for commit in self.repo.iter_commits(
self.current_branch, max_count=max_revisions
):
rev = Revision(
key=commit.name_rev.split(" ")[0],
author_name=commit.author.name,
author_email=commit.author.email,
date=commit.committed_date,
message=commit.message,
files=list(commit.stats.files.keys()),
)
revisions.append(rev)
return revisions
"""
Get the list of revisions.
:param path: the path to target.
:type path: ``str``
:param max_revisions: the maximum number of revisions.
:type max_revisions: ``int``
:return: A list of revisions.
:rtype: ``list`` of :class:`Revision`
"""
mtime = os.path.getmtime(path)
key = hashlib.sha1(str(mtime).encode()).hexdigest()[:7]
return [
Revision(
key=key,
author_name="Local User", # Don't want to leak local data
author_email="-", # as above
date=int(mtime),
message="None",
files=[],
)
def fromdict(d):
"""Instantiate from a dictionary."""
rev = Revision(
key=d["key"],
author_name=d["author_name"],
author_email=d["author_email"],
date=d["date"],
message=d["message"],
files=d["files"] if "files" in d else [],
)
operators = d["operators"]
return IndexedRevision(revision=rev, operators=operators)
def __contains__(self, item):
"""
Check if index contains `item`.
:param item: The item to search for
:type item: ``str``, :class:`Revision` or :class:`LazyRevision`
:return: ``True`` for contains, ``False`` for not.
"""
if isinstance(item, Revision):
return item.key in self._revisions
elif isinstance(item, str):
return item in self._revisions
else:
raise TypeError("Invalid type for __contains__ in Index.")