Changeset View
Changeset View
Standalone View
Standalone View
swh/model/tests/test_model.py
Show All 15 Lines | from swh.model.model import ( | ||||
BaseModel, | BaseModel, | ||||
Content, | Content, | ||||
SkippedContent, | SkippedContent, | ||||
Directory, | Directory, | ||||
Revision, | Revision, | ||||
Release, | Release, | ||||
Snapshot, | Snapshot, | ||||
Origin, | Origin, | ||||
OriginVisit, | |||||
OriginVisitStatus, | |||||
Timestamp, | Timestamp, | ||||
TimestampWithTimezone, | TimestampWithTimezone, | ||||
MissingData, | MissingData, | ||||
Person, | Person, | ||||
RawExtrinsicMetadata, | RawExtrinsicMetadata, | ||||
MetadataTargetType, | MetadataTargetType, | ||||
MetadataAuthority, | MetadataAuthority, | ||||
MetadataAuthorityType, | MetadataAuthorityType, | ||||
▲ Show 20 Lines • Show All 60 Lines • ▼ Show 20 Lines | def test_anonymization(objtype_and_obj): | ||||
elif obj_type == "revision": | elif obj_type == "revision": | ||||
assert anon_obj is not None | assert anon_obj is not None | ||||
check_person(anon_obj.author) | check_person(anon_obj.author) | ||||
check_person(anon_obj.committer) | check_person(anon_obj.committer) | ||||
else: | else: | ||||
assert anon_obj is None | assert anon_obj is None | ||||
# Origin, OriginVisit | # Origin, OriginVisit, OriginVisitStatus | ||||
@given(strategies.origins()) | @given(strategies.origins()) | ||||
def test_todict_origins(origin): | def test_todict_origins(origin): | ||||
obj = origin.to_dict() | obj = origin.to_dict() | ||||
assert "type" not in obj | assert "type" not in obj | ||||
assert type(origin)(url=origin.url) == type(origin).from_dict(obj) | assert type(origin)(url=origin.url) == type(origin).from_dict(obj) | ||||
@given(strategies.origin_visits()) | @given(strategies.origin_visits()) | ||||
def test_todict_origin_visits(origin_visit): | def test_todict_origin_visits(origin_visit): | ||||
obj = origin_visit.to_dict() | obj = origin_visit.to_dict() | ||||
assert origin_visit == type(origin_visit).from_dict(obj) | assert origin_visit == type(origin_visit).from_dict(obj) | ||||
def test_origin_visit_naive_datetime(): | |||||
with pytest.raises(ValueError, match="must be a timezone-aware datetime"): | |||||
OriginVisit( | |||||
origin="http://foo/", date=datetime.datetime.now(), type="git", | |||||
) | |||||
@given(strategies.origin_visit_statuses()) | @given(strategies.origin_visit_statuses()) | ||||
def test_todict_origin_visit_statuses(origin_visit_status): | def test_todict_origin_visit_statuses(origin_visit_status): | ||||
obj = origin_visit_status.to_dict() | obj = origin_visit_status.to_dict() | ||||
assert origin_visit_status == type(origin_visit_status).from_dict(obj) | assert origin_visit_status == type(origin_visit_status).from_dict(obj) | ||||
def test_origin_visit_status_naive_datetime(): | |||||
with pytest.raises(ValueError, match="must be a timezone-aware datetime"): | |||||
OriginVisitStatus( | |||||
origin="http://foo/", | |||||
visit=42, | |||||
date=datetime.datetime.now(), | |||||
status="ongoing", | |||||
snapshot=None, | |||||
) | |||||
# Timestamp | # Timestamp | ||||
@given(strategies.timestamps()) | @given(strategies.timestamps()) | ||||
def test_timestamps_strategy(timestamp): | def test_timestamps_strategy(timestamp): | ||||
attr.validate(timestamp) | attr.validate(timestamp) | ||||
▲ Show 20 Lines • Show All 86 Lines • ▼ Show 20 Lines | def test_timestampwithtimezone_from_datetime(): | ||||
assert tstz == TimestampWithTimezone( | assert tstz == TimestampWithTimezone( | ||||
timestamp=Timestamp(seconds=1582810759, microseconds=0,), | timestamp=Timestamp(seconds=1582810759, microseconds=0,), | ||||
offset=60, | offset=60, | ||||
negative_utc=False, | negative_utc=False, | ||||
) | ) | ||||
def test_timestampwithtimezone_from_naive_datetime(): | |||||
date = datetime.datetime(2020, 2, 27, 14, 39, 19) | |||||
with pytest.raises(ValueError, match="datetime without timezone"): | |||||
TimestampWithTimezone.from_datetime(date) | |||||
def test_timestampwithtimezone_from_iso8601(): | def test_timestampwithtimezone_from_iso8601(): | ||||
date = "2020-02-27 14:39:19.123456+0100" | date = "2020-02-27 14:39:19.123456+0100" | ||||
tstz = TimestampWithTimezone.from_iso8601(date) | tstz = TimestampWithTimezone.from_iso8601(date) | ||||
assert tstz == TimestampWithTimezone( | assert tstz == TimestampWithTimezone( | ||||
timestamp=Timestamp(seconds=1582810759, microseconds=123456,), | timestamp=Timestamp(seconds=1582810759, microseconds=123456,), | ||||
offset=60, | offset=60, | ||||
▲ Show 20 Lines • Show All 123 Lines • ▼ Show 20 Lines | def test_content_from_dict(content_d): | ||||
content_d2 = c.to_dict() | content_d2 = c.to_dict() | ||||
c2 = Content.from_dict(content_d2) | c2 = Content.from_dict(content_d2) | ||||
assert c2.ctime == c.ctime | assert c2.ctime == c.ctime | ||||
def test_content_from_dict_str_ctime(): | def test_content_from_dict_str_ctime(): | ||||
# test with ctime as a string | # test with ctime as a string | ||||
n = datetime.datetime(2020, 5, 6, 12, 34) | n = datetime.datetime(2020, 5, 6, 12, 34, tzinfo=datetime.timezone.utc) | ||||
content_d = { | content_d = { | ||||
"ctime": n.isoformat(), | "ctime": n.isoformat(), | ||||
"data": b"", | "data": b"", | ||||
"length": 0, | "length": 0, | ||||
"sha1": b"\x00", | "sha1": b"\x00", | ||||
"sha256": b"\x00", | "sha256": b"\x00", | ||||
"sha1_git": b"\x00", | "sha1_git": b"\x00", | ||||
"blake2s256": b"\x00", | "blake2s256": b"\x00", | ||||
} | } | ||||
c = Content.from_dict(content_d) | c = Content.from_dict(content_d) | ||||
assert c.ctime == n | assert c.ctime == n | ||||
def test_content_from_dict_str_naive_ctime(): | |||||
# test with ctime as a string | |||||
n = datetime.datetime(2020, 5, 6, 12, 34) | |||||
content_d = { | |||||
"ctime": n.isoformat(), | |||||
"data": b"", | |||||
"length": 0, | |||||
"sha1": b"\x00", | |||||
"sha256": b"\x00", | |||||
"sha1_git": b"\x00", | |||||
"blake2s256": b"\x00", | |||||
} | |||||
with pytest.raises(ValueError, match="must be a timezone-aware datetime."): | |||||
Content.from_dict(content_d) | |||||
@given(binary(max_size=4096)) | @given(binary(max_size=4096)) | ||||
def test_content_from_data(data): | def test_content_from_data(data): | ||||
c = Content.from_data(data) | c = Content.from_data(data) | ||||
assert c.data == data | assert c.data == data | ||||
assert c.length == len(data) | assert c.length == len(data) | ||||
assert c.status == "visible" | assert c.status == "visible" | ||||
for key, value in MultiHash.from_data(data).digest().items(): | for key, value in MultiHash.from_data(data).digest().items(): | ||||
assert getattr(c, key) == value | assert getattr(c, key) == value | ||||
@given(binary(max_size=4096)) | @given(binary(max_size=4096)) | ||||
def test_hidden_content_from_data(data): | def test_hidden_content_from_data(data): | ||||
c = Content.from_data(data, status="hidden") | c = Content.from_data(data, status="hidden") | ||||
assert c.data == data | assert c.data == data | ||||
assert c.length == len(data) | assert c.length == len(data) | ||||
assert c.status == "hidden" | assert c.status == "hidden" | ||||
for key, value in MultiHash.from_data(data).digest().items(): | for key, value in MultiHash.from_data(data).digest().items(): | ||||
assert getattr(c, key) == value | assert getattr(c, key) == value | ||||
def test_content_naive_datetime(): | |||||
c = Content.from_data(b"foo") | |||||
with pytest.raises(ValueError, match="must be a timezone-aware datetime"): | |||||
Content( | |||||
**c.to_dict(), ctime=datetime.datetime.now(), | |||||
) | |||||
# SkippedContent | # SkippedContent | ||||
@given(binary(max_size=4096)) | @given(binary(max_size=4096)) | ||||
def test_skipped_content_from_data(data): | def test_skipped_content_from_data(data): | ||||
c = SkippedContent.from_data(data, reason="reason") | c = SkippedContent.from_data(data, reason="reason") | ||||
assert c.reason == "reason" | assert c.reason == "reason" | ||||
assert c.length == len(data) | assert c.length == len(data) | ||||
Show All 9 Lines | def test_skipped_content_origin_is_str(skipped_content_d): | ||||
skipped_content_d["origin"] = "http://path/to/origin" | skipped_content_d["origin"] = "http://path/to/origin" | ||||
assert SkippedContent.from_dict(skipped_content_d) | assert SkippedContent.from_dict(skipped_content_d) | ||||
skipped_content_d["origin"] = Origin(url="http://path/to/origin") | skipped_content_d["origin"] = Origin(url="http://path/to/origin") | ||||
with pytest.raises(ValueError, match="origin"): | with pytest.raises(ValueError, match="origin"): | ||||
SkippedContent.from_dict(skipped_content_d) | SkippedContent.from_dict(skipped_content_d) | ||||
def test_skipped_content_naive_datetime(): | |||||
c = SkippedContent.from_data(b"foo", reason="reason") | |||||
with pytest.raises(ValueError, match="must be a timezone-aware datetime"): | |||||
SkippedContent( | |||||
**c.to_dict(), ctime=datetime.datetime.now(), | |||||
) | |||||
# Revision | # Revision | ||||
def test_revision_extra_headers_no_headers(): | def test_revision_extra_headers_no_headers(): | ||||
rev_dict = revision_example.copy() | rev_dict = revision_example.copy() | ||||
rev_dict.pop("id") | rev_dict.pop("id") | ||||
rev = Revision.from_dict(rev_dict) | rev = Revision.from_dict(rev_dict) | ||||
rev_dict = attr.asdict(rev, recurse=False) | rev_dict = attr.asdict(rev, recurse=False) | ||||
▲ Show 20 Lines • Show All 256 Lines • ▼ Show 20 Lines | |||||
_metadata_authority = MetadataAuthority( | _metadata_authority = MetadataAuthority( | ||||
type=MetadataAuthorityType.FORGE, url="https://forge.softwareheritage.org", | type=MetadataAuthorityType.FORGE, url="https://forge.softwareheritage.org", | ||||
) | ) | ||||
_metadata_fetcher = MetadataFetcher(name="test-fetcher", version="0.0.1",) | _metadata_fetcher = MetadataFetcher(name="test-fetcher", version="0.0.1",) | ||||
_content_swhid = parse_swhid("swh:1:cnt:94a9ed024d3859793618152ea559a168bbcbb5e2") | _content_swhid = parse_swhid("swh:1:cnt:94a9ed024d3859793618152ea559a168bbcbb5e2") | ||||
_origin_url = "https://forge.softwareheritage.org/source/swh-model.git" | _origin_url = "https://forge.softwareheritage.org/source/swh-model.git" | ||||
_common_metadata_fields = dict( | _common_metadata_fields = dict( | ||||
discovery_date=datetime.datetime.now(), | discovery_date=datetime.datetime.now(tz=datetime.timezone.utc), | ||||
authority=_metadata_authority, | authority=_metadata_authority, | ||||
fetcher=_metadata_fetcher, | fetcher=_metadata_fetcher, | ||||
format="json", | format="json", | ||||
metadata=b'{"foo": "bar"}', | metadata=b'{"foo": "bar"}', | ||||
) | ) | ||||
def test_metadata_valid(): | def test_metadata_valid(): | ||||
▲ Show 20 Lines • Show All 91 Lines • ▼ Show 20 Lines | with pytest.raises(ValueError, match="Expected core SWHID"): | ||||
object_type="content", | object_type="content", | ||||
object_id="94a9ed024d3859793618152ea559a168bbcbb5e2", | object_id="94a9ed024d3859793618152ea559a168bbcbb5e2", | ||||
metadata={"foo": "bar"}, | metadata={"foo": "bar"}, | ||||
), | ), | ||||
**_common_metadata_fields, | **_common_metadata_fields, | ||||
) | ) | ||||
def test_metadata_naive_datetime(): | |||||
with pytest.raises(ValueError, match="must be a timezone-aware datetime"): | |||||
RawExtrinsicMetadata( | |||||
type=MetadataTargetType.ORIGIN, | |||||
id=_origin_url, | |||||
**{**_common_metadata_fields, "discovery_date": datetime.datetime.now()}, | |||||
) | |||||
def test_metadata_validate_context_origin(): | def test_metadata_validate_context_origin(): | ||||
"""Checks validation of RawExtrinsicMetadata.origin.""" | """Checks validation of RawExtrinsicMetadata.origin.""" | ||||
# Origins can't have an 'origin' context | # Origins can't have an 'origin' context | ||||
with pytest.raises( | with pytest.raises( | ||||
ValueError, match="Unexpected 'origin' context for origin object" | ValueError, match="Unexpected 'origin' context for origin object" | ||||
): | ): | ||||
RawExtrinsicMetadata( | RawExtrinsicMetadata( | ||||
▲ Show 20 Lines • Show All 307 Lines • Show Last 20 Lines |