Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/git/tests/test_converters.py
# Copyright (C) 2015-2018 The Software Heritage developers | # Copyright (C) 2015-2018 The Software Heritage developers | ||||
# See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
# License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
# See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
import copy | |||||
import os | import os | ||||
import shutil | import shutil | ||||
import subprocess | import subprocess | ||||
import tempfile | import tempfile | ||||
import dulwich.repo | import dulwich.repo | ||||
import pytest | import pytest | ||||
import swh.loader.git.converters as converters | import swh.loader.git.converters as converters | ||||
from swh.model.hashutil import bytehex_to_hash, hash_to_bytes | from swh.model.hashutil import bytehex_to_hash, hash_to_bytehex, hash_to_bytes | ||||
from swh.model.model import ( | from swh.model.model import ( | ||||
Content, | Content, | ||||
ObjectType, | ObjectType, | ||||
Person, | Person, | ||||
Release, | Release, | ||||
Revision, | Revision, | ||||
RevisionType, | RevisionType, | ||||
Timestamp, | Timestamp, | ||||
▲ Show 20 Lines • Show All 60 Lines • ▼ Show 20 Lines | |||||
class SWHTag: | class SWHTag: | ||||
"""Dulwich lookalike tag class | """Dulwich lookalike tag class | ||||
""" | """ | ||||
def __init__( | def __init__( | ||||
self, | self, | ||||
sha, | |||||
name, | name, | ||||
type_name, | type_name, | ||||
target, | target, | ||||
target_type, | target_type, | ||||
tagger, | tagger, | ||||
tag_time, | tag_time, | ||||
tag_timezone, | tag_timezone, | ||||
message, | message, | ||||
signature, | signature, | ||||
): | ): | ||||
self._sha = sha | |||||
self.name = name | self.name = name | ||||
self.type_name = type_name | self.type_name = type_name | ||||
self.object = SWHObjectType(target_type), target | self.object = SWHObjectType(target_type), target | ||||
self.tagger = tagger | self.tagger = tagger | ||||
self.message = message | self.message = message | ||||
self.signature = signature | self.signature = signature | ||||
self.tag_time = tag_time | self.tag_time = tag_time | ||||
self.tag_timezone = tag_timezone | self.tag_timezone = tag_timezone | ||||
self._tag_timezone_neg_utc = False | self._tag_timezone_neg_utc = False | ||||
def sha(self): | def sha(self): | ||||
from hashlib import sha1 | class hasher: | ||||
def digest(): | |||||
return self._sha | |||||
return sha1() | return hasher | ||||
ardumont: why not ^ ?
or even simpler:
```
def sha(self):
return lambda: self._sha
```
? | |||||
Done Inline Actionsbecause that's not what dulwich expects. vlorentz: because that's not what dulwich expects. | |||||
Not Done Inline Actionsah! ardumont: ah! | |||||
@pytest.mark.fs | @pytest.mark.fs | ||||
class TestConverters: | class TestConverters: | ||||
@classmethod | @classmethod | ||||
def setup_class(cls): | def setup_class(cls): | ||||
cls.repo_path = tempfile.mkdtemp() | cls.repo_path = tempfile.mkdtemp() | ||||
Show All 32 Lines | def test_blob_to_content(self): | ||||
b"\turl = https://github.com/githubtraining/" | b"\turl = https://github.com/githubtraining/" | ||||
b"example-dependency.git\n" | b"example-dependency.git\n" | ||||
), | ), | ||||
length=124, | length=124, | ||||
status="visible", | status="visible", | ||||
) | ) | ||||
assert content == expected_content | assert content == expected_content | ||||
def test_corrupt_blob(self, mocker): | |||||
# has a signature | |||||
sha1 = hash_to_bytes("28c6f4023d65f74e3b59a2dea3c4277ed9ee07b0") | |||||
blob = copy.deepcopy(self.repo[hash_to_bytehex(sha1)]) | |||||
class hasher: | |||||
def digest(): | |||||
return sha1 | |||||
blob._sha = hasher | |||||
converters.dulwich_blob_to_content(blob) | |||||
converters.dulwich_blob_to_content_id(blob) | |||||
sha1 = hash_to_bytes("1234" * 10) | |||||
with pytest.raises(converters.HashMismatch): | |||||
converters.dulwich_blob_to_content(blob) | |||||
with pytest.raises(converters.HashMismatch): | |||||
converters.dulwich_blob_to_content_id(blob) | |||||
def test_convertion_wrong_input(self): | def test_convertion_wrong_input(self): | ||||
class Something: | class Something: | ||||
type_name = b"something-not-the-right-type" | type_name = b"something-not-the-right-type" | ||||
m = { | m = { | ||||
"blob": converters.dulwich_blob_to_content, | "blob": converters.dulwich_blob_to_content, | ||||
"blob2": converters.dulwich_blob_to_content_id, | |||||
"tree": converters.dulwich_tree_to_directory, | "tree": converters.dulwich_tree_to_directory, | ||||
"commit": converters.dulwich_tree_to_directory, | "commit": converters.dulwich_tree_to_directory, | ||||
"tag": converters.dulwich_tag_to_release, | "tag": converters.dulwich_tag_to_release, | ||||
} | } | ||||
for _callable in m.values(): | for _callable in m.values(): | ||||
with pytest.raises(ValueError): | with pytest.raises(ValueError): | ||||
_callable(Something()) | _callable(Something()) | ||||
def test_corrupt_tree(self): | |||||
# has a signature | |||||
sha1 = b"f0695c2e2fa7ce9d574023c3413761a473e500ca" | |||||
tree = copy.deepcopy(self.repo[sha1]) | |||||
converters.dulwich_tree_to_directory(tree) | |||||
del tree._entries[next(iter(tree._entries))] | |||||
with pytest.raises(converters.HashMismatch): | |||||
converters.dulwich_tree_to_directory(tree) | |||||
def test_commit_to_revision(self): | def test_commit_to_revision(self): | ||||
sha1 = b"9768d0b576dbaaecd80abedad6dfd0d72f1476da" | sha1 = b"9768d0b576dbaaecd80abedad6dfd0d72f1476da" | ||||
revision = converters.dulwich_commit_to_revision(self.repo[sha1]) | revision = converters.dulwich_commit_to_revision(self.repo[sha1]) | ||||
expected_revision = Revision( | expected_revision = Revision( | ||||
id=hash_to_bytes("9768d0b576dbaaecd80abedad6dfd0d72f1476da"), | id=hash_to_bytes("9768d0b576dbaaecd80abedad6dfd0d72f1476da"), | ||||
directory=b"\xf0i\\./\xa7\xce\x9dW@#\xc3A7a\xa4s\xe5\x00\xca", | directory=b"\xf0i\\./\xa7\xce\x9dW@#\xc3A7a\xa4s\xe5\x00\xca", | ||||
type=RevisionType.GIT, | type=RevisionType.GIT, | ||||
▲ Show 20 Lines • Show All 58 Lines • ▼ Show 20 Lines | def test_commit_to_revision_with_extra_headers(self): | ||||
offset=120, | offset=120, | ||||
), | ), | ||||
parents=(bytes.fromhex("c730509025c6e81947102b2d77bc4dc1cade9489"),), | parents=(bytes.fromhex("c730509025c6e81947102b2d77bc4dc1cade9489"),), | ||||
synthetic=False, | synthetic=False, | ||||
) | ) | ||||
assert revision == expected_revision | assert revision == expected_revision | ||||
@pytest.mark.parametrize( | |||||
"attribute", ["_message", "_encoding", "_author", "_gpgsig"] | |||||
) | |||||
def test_corrupt_commit(self, attribute): | |||||
# has a signature | |||||
sha1 = b"322f5bc915e50fc25e85226b5a182bded0e98e4b" | |||||
commit = copy.deepcopy(self.repo[sha1]) | |||||
converters.dulwich_commit_to_revision(commit) | |||||
setattr(commit, attribute, b"abcde") | |||||
with pytest.raises(converters.HashMismatch): | |||||
converters.dulwich_commit_to_revision(commit) | |||||
if attribute == "_gpgsig": | |||||
setattr(commit, attribute, None) | |||||
with pytest.raises(converters.HashMismatch): | |||||
converters.dulwich_commit_to_revision(commit) | |||||
def test_commit_to_revision_with_extra_headers_mergetag(self): | def test_commit_to_revision_with_extra_headers_mergetag(self): | ||||
sha1 = b"3ab3da4bf0f81407be16969df09cd1c8af9ac703" | sha1 = b"3ab3da4bf0f81407be16969df09cd1c8af9ac703" | ||||
revision = converters.dulwich_commit_to_revision(self.repo[sha1]) | revision = converters.dulwich_commit_to_revision(self.repo[sha1]) | ||||
expected_revision = Revision( | expected_revision = Revision( | ||||
id=hash_to_bytes(sha1.decode()), | id=hash_to_bytes(sha1.decode()), | ||||
directory=bytes.fromhex("faa4b64a841ca3e3f07d6501caebda2e3e8e544e"), | directory=bytes.fromhex("faa4b64a841ca3e3f07d6501caebda2e3e8e544e"), | ||||
type=RevisionType.GIT, | type=RevisionType.GIT, | ||||
▲ Show 20 Lines • Show All 52 Lines • ▼ Show 20 Lines | def test_author_line_to_author(self): | ||||
b"something": Person(name=b"something", email=None, fullname=b"something"), | b"something": Person(name=b"something", email=None, fullname=b"something"), | ||||
} | } | ||||
for author in sorted(tests): | for author in sorted(tests): | ||||
parsed_author = tests[author] | parsed_author = tests[author] | ||||
assert parsed_author == converters.parse_author(author) | assert parsed_author == converters.parse_author(author) | ||||
def test_dulwich_tag_to_release_no_author_no_date(self): | def test_dulwich_tag_to_release_no_author_no_date(self): | ||||
target = b"641fb6e08ddb2e4fd096dcf18e80b894bf" | sha = hash_to_bytes("f6e367357b446bd1315276de5e88ba3d0d99e136") | ||||
target = b"641fb6e08ddb2e4fd096dcf18e80b894bf7e25ce" | |||||
message = b"some release message" | message = b"some release message" | ||||
tag = SWHTag( | tag = SWHTag( | ||||
sha=sha, | |||||
name=b"blah", | name=b"blah", | ||||
type_name=b"tag", | type_name=b"tag", | ||||
target=target, | target=target, | ||||
target_type=b"commit", | target_type=b"commit", | ||||
message=message, | message=message, | ||||
signature=None, | signature=None, | ||||
tagger=None, | tagger=None, | ||||
tag_time=None, | tag_time=None, | ||||
tag_timezone=None, | tag_timezone=None, | ||||
) | ) | ||||
# when | # when | ||||
actual_release = converters.dulwich_tag_to_release(tag) | actual_release = converters.dulwich_tag_to_release(tag) | ||||
# then | # then | ||||
expected_release = Release( | expected_release = Release( | ||||
author=None, | author=None, | ||||
date=None, | date=None, | ||||
id=b"\xda9\xa3\xee^kK\r2U\xbf\xef\x95`\x18\x90\xaf\xd8\x07\t", | id=sha, | ||||
message=message, | message=message, | ||||
metadata=None, | metadata=None, | ||||
name=b"blah", | name=b"blah", | ||||
synthetic=False, | synthetic=False, | ||||
target=hash_to_bytes(target.decode()), | target=hash_to_bytes(target.decode()), | ||||
target_type=ObjectType.REVISION, | target_type=ObjectType.REVISION, | ||||
) | ) | ||||
assert actual_release == expected_release | assert actual_release == expected_release | ||||
def test_dulwich_tag_to_release_author_and_date(self): | def test_dulwich_tag_to_release_author_and_date(self): | ||||
sha = hash_to_bytes("fc1e6a4f1e37e93e28e78560e73efd0b12f616ef") | |||||
tagger = b"hey dude <hello@mail.org>" | tagger = b"hey dude <hello@mail.org>" | ||||
target = b"641fb6e08ddb2e4fd096dcf18e80b894bf" | target = b"641fb6e08ddb2e4fd096dcf18e80b894bf7e25ce" | ||||
message = b"some release message" | message = b"some release message" | ||||
import datetime | import datetime | ||||
date = datetime.datetime(2007, 12, 5, tzinfo=datetime.timezone.utc).timestamp() | date = datetime.datetime(2007, 12, 5, tzinfo=datetime.timezone.utc).timestamp() | ||||
tag = SWHTag( | tag = SWHTag( | ||||
sha=sha, | |||||
name=b"blah", | name=b"blah", | ||||
type_name=b"tag", | type_name=b"tag", | ||||
target=target, | target=target, | ||||
target_type=b"commit", | target_type=b"commit", | ||||
message=message, | message=message, | ||||
signature=None, | signature=None, | ||||
tagger=tagger, | tagger=tagger, | ||||
tag_time=date, | tag_time=date, | ||||
Show All 10 Lines | def test_dulwich_tag_to_release_author_and_date(self): | ||||
fullname=b"hey dude <hello@mail.org>", | fullname=b"hey dude <hello@mail.org>", | ||||
name=b"hey dude", | name=b"hey dude", | ||||
), | ), | ||||
date=TimestampWithTimezone( | date=TimestampWithTimezone( | ||||
negative_utc=False, | negative_utc=False, | ||||
offset=0, | offset=0, | ||||
timestamp=Timestamp(seconds=1196812800, microseconds=0,), | timestamp=Timestamp(seconds=1196812800, microseconds=0,), | ||||
), | ), | ||||
id=b"\xda9\xa3\xee^kK\r2U\xbf\xef\x95`\x18\x90\xaf\xd8\x07\t", | id=sha, | ||||
message=message, | message=message, | ||||
metadata=None, | metadata=None, | ||||
name=b"blah", | name=b"blah", | ||||
synthetic=False, | synthetic=False, | ||||
target=hash_to_bytes(target.decode()), | target=hash_to_bytes(target.decode()), | ||||
target_type=ObjectType.REVISION, | target_type=ObjectType.REVISION, | ||||
) | ) | ||||
assert actual_release == expected_release | assert actual_release == expected_release | ||||
def test_dulwich_tag_to_release_author_no_date(self): | def test_dulwich_tag_to_release_author_no_date(self): | ||||
# to reproduce bug T815 (fixed) | # to reproduce bug T815 (fixed) | ||||
sha = hash_to_bytes("41076e970975122dc6b2a878aa9797960bc4781d") | |||||
tagger = b"hey dude <hello@mail.org>" | tagger = b"hey dude <hello@mail.org>" | ||||
target = b"641fb6e08ddb2e4fd096dcf18e80b894bf" | target = b"641fb6e08ddb2e4fd096dcf18e80b894bf7e25ce" | ||||
message = b"some release message" | message = b"some release message" | ||||
tag = SWHTag( | tag = SWHTag( | ||||
sha=sha, | |||||
name=b"blah", | name=b"blah", | ||||
type_name=b"tag", | type_name=b"tag", | ||||
target=target, | target=target, | ||||
target_type=b"commit", | target_type=b"commit", | ||||
message=message, | message=message, | ||||
signature=None, | signature=None, | ||||
tagger=tagger, | tagger=tagger, | ||||
tag_time=None, | tag_time=None, | ||||
tag_timezone=None, | tag_timezone=None, | ||||
) | ) | ||||
# when | # when | ||||
actual_release = converters.dulwich_tag_to_release(tag) | actual_release = converters.dulwich_tag_to_release(tag) | ||||
# then | # then | ||||
expected_release = Release( | expected_release = Release( | ||||
author=Person( | author=Person( | ||||
email=b"hello@mail.org", | email=b"hello@mail.org", | ||||
fullname=b"hey dude <hello@mail.org>", | fullname=b"hey dude <hello@mail.org>", | ||||
name=b"hey dude", | name=b"hey dude", | ||||
), | ), | ||||
date=None, | date=None, | ||||
id=b"\xda9\xa3\xee^kK\r2U\xbf\xef\x95`\x18\x90\xaf\xd8\x07\t", | id=sha, | ||||
message=message, | message=message, | ||||
metadata=None, | metadata=None, | ||||
name=b"blah", | name=b"blah", | ||||
synthetic=False, | synthetic=False, | ||||
target=hash_to_bytes(target.decode()), | target=hash_to_bytes(target.decode()), | ||||
target_type=ObjectType.REVISION, | target_type=ObjectType.REVISION, | ||||
) | ) | ||||
assert actual_release == expected_release | assert actual_release == expected_release | ||||
def test_dulwich_tag_to_release_signature(self): | def test_dulwich_tag_to_release_signature(self): | ||||
target = b"641fb6e08ddb2e4fd096dcf18e80b894bf" | target = b"641fb6e08ddb2e4fd096dcf18e80b894bf7e25ce" | ||||
message = b"some release message" | message = b"some release message" | ||||
sha = hash_to_bytes("46fff489610ed733d2cc904e363070dadee05c71") | |||||
tag = SWHTag( | tag = SWHTag( | ||||
sha=sha, | |||||
name=b"blah", | name=b"blah", | ||||
type_name=b"tag", | type_name=b"tag", | ||||
target=target, | target=target, | ||||
target_type=b"commit", | target_type=b"commit", | ||||
message=message, | message=message, | ||||
signature=GPGSIG, | signature=GPGSIG, | ||||
tagger=None, | tagger=None, | ||||
tag_time=None, | tag_time=None, | ||||
tag_timezone=None, | tag_timezone=None, | ||||
) | ) | ||||
# when | # when | ||||
actual_release = converters.dulwich_tag_to_release(tag) | actual_release = converters.dulwich_tag_to_release(tag) | ||||
# then | # then | ||||
expected_release = Release( | expected_release = Release( | ||||
author=None, | author=None, | ||||
date=None, | date=None, | ||||
id=b"\xda9\xa3\xee^kK\r2U\xbf\xef\x95`\x18\x90\xaf\xd8\x07\t", | id=sha, | ||||
message=message + GPGSIG, | message=message + GPGSIG, | ||||
metadata=None, | metadata=None, | ||||
name=b"blah", | name=b"blah", | ||||
synthetic=False, | synthetic=False, | ||||
target=hash_to_bytes(target.decode()), | target=hash_to_bytes(target.decode()), | ||||
target_type=ObjectType.REVISION, | target_type=ObjectType.REVISION, | ||||
) | ) | ||||
assert actual_release == expected_release | assert actual_release == expected_release | ||||
@pytest.mark.parametrize("attribute", ["name", "message", "signature"]) | |||||
def test_corrupt_tag(self, attribute): | |||||
# has a signature | |||||
sha = hash_to_bytes("46fff489610ed733d2cc904e363070dadee05c71") | |||||
target = b"641fb6e08ddb2e4fd096dcf18e80b894bf7e25ce" | |||||
message = b"some release message" | |||||
tag = SWHTag( | |||||
sha=sha, | |||||
name=b"blah", | |||||
type_name=b"tag", | |||||
target=target, | |||||
target_type=b"commit", | |||||
message=message, | |||||
signature=GPGSIG, | |||||
tagger=None, | |||||
tag_time=None, | |||||
tag_timezone=None, | |||||
) | |||||
converters.dulwich_tag_to_release(tag) | |||||
setattr(tag, attribute, b"abcde") | |||||
with pytest.raises(converters.HashMismatch): | |||||
converters.dulwich_tag_to_release(tag) | |||||
if attribute == "signature": | |||||
setattr(tag, attribute, None) | |||||
with pytest.raises(converters.HashMismatch): | |||||
converters.dulwich_tag_to_release(tag) |
why not ^ ?
or even simpler:
?