Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
# Copyright (C) 2015-2020 The Software Heritage developers | # Copyright (C) 2015-2020 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 attr | |||||
import copy | import copy | ||||
from contextlib import contextmanager | from contextlib import contextmanager | ||||
import datetime | import datetime | ||||
import inspect | import inspect | ||||
import itertools | import itertools | ||||
import math | import math | ||||
import queue | import queue | ||||
import random | import random | ||||
▲ Show 20 Lines • Show All 1,425 Lines • ▼ Show 20 Lines | def test_origin_visit_get_random(self, swh_storage): | ||||
visits = self._generate_random_visits() | visits = self._generate_random_visits() | ||||
visit_type = "git" | visit_type = "git" | ||||
# Add visits to those origins | # Add visits to those origins | ||||
for origin in data.origins: | for origin in data.origins: | ||||
origin_url = origin["url"] | origin_url = origin["url"] | ||||
for date_visit in visits: | for date_visit in visits: | ||||
visit = swh_storage.origin_visit_add( | visit = swh_storage.origin_visit_add( | ||||
[ | [OriginVisit(origin=origin_url, date=date_visit, type=visit_type,)] | ||||
OriginVisit( | |||||
origin=origin_url, | |||||
date=date_visit, | |||||
type=visit_type, | |||||
status="ongoing", | |||||
snapshot=None, | |||||
) | |||||
] | |||||
)[0] | )[0] | ||||
swh_storage.origin_visit_status_add( | swh_storage.origin_visit_status_add( | ||||
[ | [ | ||||
OriginVisitStatus( | OriginVisitStatus( | ||||
origin=origin_url, | origin=origin_url, | ||||
visit=visit.visit, | visit=visit.visit, | ||||
date=now(), | date=now(), | ||||
status="full", | status="full", | ||||
Show All 19 Lines | def test_origin_visit_get_random_nothing_found(self, swh_storage): | ||||
visit_type = "hg" | visit_type = "hg" | ||||
# Add some visits outside of the random generation selection so nothing | # Add some visits outside of the random generation selection so nothing | ||||
# will be found by the random selection | # will be found by the random selection | ||||
visits = self._generate_random_visits(nb_visits=3, start=13, end=24) | visits = self._generate_random_visits(nb_visits=3, start=13, end=24) | ||||
for origin in data.origins: | for origin in data.origins: | ||||
origin_url = origin["url"] | origin_url = origin["url"] | ||||
for date_visit in visits: | for date_visit in visits: | ||||
visit = swh_storage.origin_visit_add( | visit = swh_storage.origin_visit_add( | ||||
[ | [OriginVisit(origin=origin_url, date=date_visit, type=visit_type,)] | ||||
OriginVisit( | |||||
origin=origin_url, | |||||
date=date_visit, | |||||
type=visit_type, | |||||
status="ongoing", | |||||
snapshot=None, | |||||
) | |||||
] | |||||
)[0] | )[0] | ||||
swh_storage.origin_visit_status_add( | swh_storage.origin_visit_status_add( | ||||
[ | [ | ||||
OriginVisitStatus( | OriginVisitStatus( | ||||
origin=origin_url, | origin=origin_url, | ||||
visit=visit.visit, | visit=visit.visit, | ||||
date=now(), | date=now(), | ||||
status="full", | status="full", | ||||
▲ Show 20 Lines • Show All 146 Lines • ▼ Show 20 Lines | def test_origin_visit_add(self, swh_storage): | ||||
date_visit = now() | date_visit = now() | ||||
date_visit2 = date_visit + datetime.timedelta(minutes=1) | date_visit2 = date_visit + datetime.timedelta(minutes=1) | ||||
date_visit = round_to_milliseconds(date_visit) | date_visit = round_to_milliseconds(date_visit) | ||||
date_visit2 = round_to_milliseconds(date_visit2) | date_visit2 = round_to_milliseconds(date_visit2) | ||||
visit1 = OriginVisit( | visit1 = OriginVisit( | ||||
origin=origin1.url, | origin=origin1.url, date=date_visit, type=data.type_visit1, | ||||
date=date_visit, | |||||
type=data.type_visit1, | |||||
status="ongoing", | |||||
snapshot=None, | |||||
) | ) | ||||
visit2 = OriginVisit( | visit2 = OriginVisit( | ||||
origin=origin1.url, | origin=origin1.url, date=date_visit2, type=data.type_visit2, | ||||
date=date_visit2, | |||||
type=data.type_visit2, | |||||
status="partial", | |||||
snapshot=None, | |||||
) | ) | ||||
# add once | # add once | ||||
ov1, ov2 = swh_storage.origin_visit_add([visit1, visit2]) | ov1, ov2 = swh_storage.origin_visit_add([visit1, visit2]) | ||||
# then again (will be ignored as they already exist) | # then again (will be ignored as they already exist) | ||||
origin_visit1, origin_visit2 = swh_storage.origin_visit_add([ov1, ov2]) | origin_visit1, origin_visit2 = swh_storage.origin_visit_add([ov1, ov2]) | ||||
assert ov1 == origin_visit1 | assert ov1 == origin_visit1 | ||||
assert ov2 == origin_visit2 | assert ov2 == origin_visit2 | ||||
ovs1 = OriginVisitStatus( | |||||
origin=origin1.url, | |||||
visit=ov1.visit, | |||||
date=date_visit, | |||||
status="created", | |||||
snapshot=None, | |||||
) | |||||
ovs2 = OriginVisitStatus( | |||||
origin=origin1.url, | |||||
visit=ov2.visit, | |||||
date=date_visit2, | |||||
status="created", | |||||
snapshot=None, | |||||
) | |||||
actual_origin_visits = list(swh_storage.origin_visit_get(origin1.url)) | actual_origin_visits = list(swh_storage.origin_visit_get(origin1.url)) | ||||
expected_visits = [ | expected_visits = [ | ||||
attr.evolve(visit1, visit=ov1.visit), | {**ovs1.to_dict(), "type": ov1.type}, | ||||
attr.evolve(visit2, visit=ov2.visit), | {**ovs2.to_dict(), "type": ov2.type}, | ||||
] | ] | ||||
expected_visit_statuses = [] | |||||
for visit in expected_visits: | |||||
visit_status = visit.to_dict() | |||||
visit_status.pop("type") | |||||
expected_visit_statuses.append(OriginVisitStatus.from_dict(visit_status)) | |||||
assert len(expected_visits) == len(actual_origin_visits) | assert len(expected_visits) == len(actual_origin_visits) | ||||
for visit in expected_visits: | for visit in expected_visits: | ||||
assert visit.to_dict() in actual_origin_visits | assert visit in actual_origin_visits | ||||
actual_objects = set(swh_storage.journal_writer.journal.objects) | actual_objects = set(swh_storage.journal_writer.journal.objects) | ||||
# we write to the journal as many times as we call the endpoint | # we write to the journal as many times as we call the endpoint | ||||
assert actual_objects == set( | assert actual_objects == set( | ||||
[("origin", origin1)] | [("origin", origin1)] | ||||
+ [("origin_visit", visit) for visit in expected_visits] * 2 | + [("origin_visit", visit) for visit in [ov1, ov2]] * 2 | ||||
+ [("origin_visit_status", ovs) for ovs in expected_visit_statuses] * 2 | + [("origin_visit_status", ovs) for ovs in [ovs1, ovs2]] | ||||
) | ) | ||||
def test_origin_visit_add_validation(self, swh_storage): | def test_origin_visit_add_validation(self, swh_storage): | ||||
"""Unknown origin when adding visits should raise""" | """Unknown origin when adding visits should raise""" | ||||
visit = OriginVisit( | visit = OriginVisit( | ||||
origin="something-unknown", | origin="something-unknown", date=now(), type=data.type_visit1, | ||||
date=now(), | |||||
type=data.type_visit1, | |||||
status="ongoing", | |||||
snapshot=None, | |||||
) | ) | ||||
with pytest.raises(StorageArgumentException, match="Unknown origin"): | with pytest.raises(StorageArgumentException, match="Unknown origin"): | ||||
swh_storage.origin_visit_add([visit]) | swh_storage.origin_visit_add([visit]) | ||||
objects = list(swh_storage.journal_writer.journal.objects) | objects = list(swh_storage.journal_writer.journal.objects) | ||||
assert not objects | assert not objects | ||||
def test_origin_visit_status_add_validation(self, swh_storage): | def test_origin_visit_status_add_validation(self, swh_storage): | ||||
Show All 18 Lines | def test_origin_visit_status_add(self, swh_storage): | ||||
""" | """ | ||||
origin1 = Origin.from_dict(data.origin2) | origin1 = Origin.from_dict(data.origin2) | ||||
origin2 = Origin(url="new-origin") | origin2 = Origin(url="new-origin") | ||||
swh_storage.origin_add([origin1, origin2]) | swh_storage.origin_add([origin1, origin2]) | ||||
ov1, ov2 = swh_storage.origin_visit_add( | ov1, ov2 = swh_storage.origin_visit_add( | ||||
[ | [ | ||||
OriginVisit( | OriginVisit( | ||||
origin=origin1.url, date=data.date_visit1, type=data.type_visit1, | |||||
), | |||||
OriginVisit( | |||||
origin=origin2.url, date=data.date_visit2, type=data.type_visit2, | |||||
), | |||||
] | |||||
) | |||||
ovs1 = OriginVisitStatus( | |||||
origin=origin1.url, | origin=origin1.url, | ||||
visit=ov1.visit, | |||||
date=data.date_visit1, | date=data.date_visit1, | ||||
type=data.type_visit1, | status="created", | ||||
status="ongoing", | |||||
snapshot=None, | snapshot=None, | ||||
), | ) | ||||
OriginVisit( | ovs2 = OriginVisitStatus( | ||||
origin=origin2.url, | origin=origin2.url, | ||||
visit=ov2.visit, | |||||
date=data.date_visit2, | date=data.date_visit2, | ||||
type=data.type_visit2, | status="created", | ||||
status="ongoing", | |||||
snapshot=None, | snapshot=None, | ||||
), | |||||
] | |||||
) | ) | ||||
snapshot_id = data.snapshot["id"] | snapshot_id = data.snapshot["id"] | ||||
date_visit_now = now() | date_visit_now = now() | ||||
visit_status1 = OriginVisitStatus( | visit_status1 = OriginVisitStatus( | ||||
origin=ov1.origin, | origin=ov1.origin, | ||||
visit=ov1.visit, | visit=ov1.visit, | ||||
date=date_visit_now, | date=date_visit_now, | ||||
Show All 27 Lines | def test_origin_visit_status_add(self, swh_storage): | ||||
assert origin_visit2["status"] == "ongoing" | assert origin_visit2["status"] == "ongoing" | ||||
assert origin_visit2["snapshot"] is None | assert origin_visit2["snapshot"] is None | ||||
assert origin_visit2["metadata"] == {"intrinsic": "something"} | assert origin_visit2["metadata"] == {"intrinsic": "something"} | ||||
actual_objects = list(swh_storage.journal_writer.journal.objects) | actual_objects = list(swh_storage.journal_writer.journal.objects) | ||||
expected_origins = [origin1, origin2] | expected_origins = [origin1, origin2] | ||||
expected_visits = [ov1, ov2] | expected_visits = [ov1, ov2] | ||||
expected_visit_statuses = [] | expected_visit_statuses = [ovs1, ovs2, visit_status1, visit_status2] | ||||
for visit in expected_visits: # out of origin-visit-add calls | |||||
visit_status = visit.to_dict() | |||||
visit_status.pop("type") | |||||
expected_visit_statuses.append(OriginVisitStatus.from_dict(visit_status)) | |||||
# out of origin-visit-status add calls | |||||
expected_visit_statuses += [visit_status1, visit_status2] | |||||
expected_objects = ( | expected_objects = ( | ||||
[("origin", o) for o in expected_origins] | [("origin", o) for o in expected_origins] | ||||
+ [("origin_visit", v) for v in expected_visits] | + [("origin_visit", v) for v in expected_visits] | ||||
+ [("origin_visit_status", ovs) for ovs in expected_visit_statuses] | + [("origin_visit_status", ovs) for ovs in expected_visit_statuses] | ||||
) | ) | ||||
assert len(actual_objects) == len(expected_objects) | |||||
for obj in expected_objects: | for obj in expected_objects: | ||||
assert obj in actual_objects | assert obj in actual_objects | ||||
def test_origin_visit_status_add_twice(self, swh_storage): | def test_origin_visit_status_add_twice(self, swh_storage): | ||||
"""Correct origin visit statuses should add a new visit status | """Correct origin visit statuses should add a new visit status | ||||
""" | """ | ||||
origin1 = Origin.from_dict(data.origin2) | origin1 = Origin.from_dict(data.origin2) | ||||
swh_storage.origin_add([origin1]) | swh_storage.origin_add([origin1]) | ||||
ov1 = swh_storage.origin_visit_add( | ov1 = swh_storage.origin_visit_add( | ||||
[ | [ | ||||
OriginVisit( | OriginVisit( | ||||
origin=origin1.url, | origin=origin1.url, date=data.date_visit1, type=data.type_visit1, | ||||
date=data.date_visit1, | |||||
type=data.type_visit1, | |||||
status="ongoing", | |||||
snapshot=None, | |||||
), | ), | ||||
] | ] | ||||
)[0] | )[0] | ||||
ovs1 = OriginVisitStatus( | |||||
origin=origin1.url, | |||||
visit=ov1.visit, | |||||
date=data.date_visit1, | |||||
status="created", | |||||
snapshot=None, | |||||
) | |||||
snapshot_id = data.snapshot["id"] | snapshot_id = data.snapshot["id"] | ||||
date_visit_now = now() | date_visit_now = now() | ||||
visit_status1 = OriginVisitStatus( | visit_status1 = OriginVisitStatus( | ||||
origin=ov1.origin, | origin=ov1.origin, | ||||
visit=ov1.visit, | visit=ov1.visit, | ||||
date=date_visit_now, | date=date_visit_now, | ||||
status="full", | status="full", | ||||
snapshot=snapshot_id, | snapshot=snapshot_id, | ||||
Show All 10 Lines | def test_origin_visit_status_add_twice(self, swh_storage): | ||||
assert origin_visit1 | assert origin_visit1 | ||||
assert origin_visit1["status"] == "full" | assert origin_visit1["status"] == "full" | ||||
assert origin_visit1["snapshot"] == snapshot_id | assert origin_visit1["snapshot"] == snapshot_id | ||||
actual_objects = list(swh_storage.journal_writer.journal.objects) | actual_objects = list(swh_storage.journal_writer.journal.objects) | ||||
expected_origins = [origin1] | expected_origins = [origin1] | ||||
expected_visits = [ov1] | expected_visits = [ov1] | ||||
expected_visit_statuses = [] | expected_visit_statuses = [ovs1, visit_status1, visit_status1] | ||||
for visit in expected_visits: # out of origin-visit-add calls | |||||
visit_status = visit.to_dict() | |||||
visit_status.pop("type") | |||||
expected_visit_statuses.append(OriginVisitStatus.from_dict(visit_status)) | |||||
# write twice in the journal | # write twice in the journal | ||||
expected_visit_statuses += [visit_status1] * 2 | |||||
expected_objects = ( | expected_objects = ( | ||||
[("origin", o) for o in expected_origins] | [("origin", o) for o in expected_origins] | ||||
+ [("origin_visit", v) for v in expected_visits] | + [("origin_visit", v) for v in expected_visits] | ||||
+ [("origin_visit_status", ovs) for ovs in expected_visit_statuses] | + [("origin_visit_status", ovs) for ovs in expected_visit_statuses] | ||||
) | ) | ||||
assert len(actual_objects) == len(expected_objects) | |||||
for obj in expected_objects: | for obj in expected_objects: | ||||
assert obj in actual_objects | assert obj in actual_objects | ||||
def test_origin_visit_find_by_date(self, swh_storage): | def test_origin_visit_find_by_date(self, swh_storage): | ||||
# given | # given | ||||
origin_url = swh_storage.origin_add_one(data.origin) | origin = Origin.from_dict(data.origin) | ||||
swh_storage.origin_add_one(data.origin) | |||||
visit1 = OriginVisit( | visit1 = OriginVisit( | ||||
origin=origin_url, | origin=origin.url, date=data.date_visit2, type=data.type_visit1, | ||||
) | |||||
visit2 = OriginVisit( | |||||
origin=origin.url, date=data.date_visit3, type=data.type_visit2, | |||||
) | |||||
visit3 = OriginVisit( | |||||
origin=origin.url, date=data.date_visit2, type=data.type_visit3, | |||||
) | |||||
ov1, ov2, ov3 = swh_storage.origin_visit_add([visit1, visit2, visit3]) | |||||
ovs1 = OriginVisitStatus( | |||||
origin=origin.url, | |||||
visit=ov1.visit, | |||||
date=data.date_visit2, | date=data.date_visit2, | ||||
type=data.type_visit1, | |||||
status="ongoing", | status="ongoing", | ||||
snapshot=None, | snapshot=None, | ||||
) | ) | ||||
visit2 = OriginVisit( | ovs2 = OriginVisitStatus( | ||||
origin=origin_url, | origin=origin.url, | ||||
visit=ov2.visit, | |||||
date=data.date_visit3, | date=data.date_visit3, | ||||
type=data.type_visit2, | |||||
status="ongoing", | status="ongoing", | ||||
snapshot=None, | snapshot=None, | ||||
) | ) | ||||
visit3 = OriginVisit( | ovs3 = OriginVisitStatus( | ||||
origin=origin_url, | origin=origin.url, | ||||
visit=ov3.visit, | |||||
date=data.date_visit2, | date=data.date_visit2, | ||||
type=data.type_visit3, | |||||
status="ongoing", | status="ongoing", | ||||
snapshot=None, | snapshot=None, | ||||
) | ) | ||||
swh_storage.origin_visit_status_add([ovs1, ovs2, ovs3]) | |||||
_, ov2, ov3 = swh_storage.origin_visit_add([visit1, visit2, visit3]) | |||||
# Simple case | # Simple case | ||||
visit = swh_storage.origin_visit_find_by_date(origin_url, data.date_visit3) | visit = swh_storage.origin_visit_find_by_date(origin.url, data.date_visit3) | ||||
assert visit["visit"] == ov2.visit | assert visit["visit"] == ov2.visit | ||||
# There are two visits at the same date, the latest must be returned | # There are two visits at the same date, the latest must be returned | ||||
visit = swh_storage.origin_visit_find_by_date(origin_url, data.date_visit2) | visit = swh_storage.origin_visit_find_by_date(origin.url, data.date_visit2) | ||||
assert visit["visit"] == ov3.visit | assert visit["visit"] == ov3.visit | ||||
def test_origin_visit_find_by_date__unknown_origin(self, swh_storage): | def test_origin_visit_find_by_date__unknown_origin(self, swh_storage): | ||||
swh_storage.origin_visit_find_by_date("foo", data.date_visit2) | swh_storage.origin_visit_find_by_date("foo", data.date_visit2) | ||||
def test_origin_visit_get_by(self, swh_storage): | def test_origin_visit_get_by(self, swh_storage): | ||||
origin_url = swh_storage.origin_add_one(data.origin) | origin_url = swh_storage.origin_add_one(data.origin) | ||||
origin_url2 = swh_storage.origin_add_one(data.origin2) | origin_url2 = swh_storage.origin_add_one(data.origin2) | ||||
visit = OriginVisit( | visit = OriginVisit( | ||||
origin=origin_url, | origin=origin_url, date=data.date_visit2, type=data.type_visit2, | ||||
date=data.date_visit2, | |||||
type=data.type_visit2, | |||||
status="ongoing", | |||||
snapshot=None, | |||||
) | ) | ||||
origin_visit1 = swh_storage.origin_visit_add([visit])[0] | origin_visit1 = swh_storage.origin_visit_add([visit])[0] | ||||
swh_storage.snapshot_add([data.snapshot]) | swh_storage.snapshot_add([data.snapshot]) | ||||
swh_storage.origin_visit_status_add( | swh_storage.origin_visit_status_add( | ||||
[ | [ | ||||
OriginVisitStatus( | OriginVisitStatus( | ||||
origin=origin_url, | origin=origin_url, | ||||
▲ Show 20 Lines • Show All 84 Lines • ▼ Show 20 Lines | |||||
def test_origin_visit_get_latest_filter_type(self, swh_storage): | def test_origin_visit_get_latest_filter_type(self, swh_storage): | ||||
"""Filtering origin visit get latest with filter type should be ok | """Filtering origin visit get latest with filter type should be ok | ||||
""" | """ | ||||
origin = Origin.from_dict(data.origin) | origin = Origin.from_dict(data.origin) | ||||
swh_storage.origin_add_one(origin) | swh_storage.origin_add_one(origin) | ||||
visit1 = OriginVisit( | visit1 = OriginVisit( | ||||
origin=origin.url, | origin=origin.url, date=data.date_visit1, type=data.type_visit1, | ||||
date=data.date_visit1, | |||||
type=data.type_visit1, | |||||
status="ongoing", | |||||
snapshot=None, | |||||
) | ) | ||||
visit2 = OriginVisit( | visit2 = OriginVisit( | ||||
origin=origin.url, | origin=origin.url, date=data.date_visit2, type=data.type_visit2, | ||||
date=data.date_visit2, | |||||
type=data.type_visit2, | |||||
status="ongoing", | |||||
snapshot=None, | |||||
) | ) | ||||
# Add a visit with the same date as the previous one | # Add a visit with the same date as the previous one | ||||
visit3 = OriginVisit( | visit3 = OriginVisit( | ||||
origin=origin.url, | origin=origin.url, date=data.date_visit2, type=data.type_visit2, | ||||
date=data.date_visit2, | |||||
type=data.type_visit2, | |||||
status="ongoing", | |||||
snapshot=None, | |||||
) | ) | ||||
assert data.type_visit1 != data.type_visit2 | assert data.type_visit1 != data.type_visit2 | ||||
assert data.date_visit1 < data.date_visit2 | assert data.date_visit1 < data.date_visit2 | ||||
ov1, ov2, ov3 = swh_storage.origin_visit_add([visit1, visit2, visit3]) | ov1, ov2, ov3 = swh_storage.origin_visit_add([visit1, visit2, visit3]) | ||||
origin_visit1 = swh_storage.origin_visit_get_by(origin.url, ov1.visit) | origin_visit1 = swh_storage.origin_visit_get_by(origin.url, ov1.visit) | ||||
origin_visit3 = swh_storage.origin_visit_get_by(origin.url, ov3.visit) | origin_visit3 = swh_storage.origin_visit_get_by(origin.url, ov3.visit) | ||||
assert data.type_visit1 != data.type_visit2 | assert data.type_visit1 != data.type_visit2 | ||||
# Check type filter is ok | # Check type filter is ok | ||||
actual_ov1 = swh_storage.origin_visit_get_latest( | actual_ov1 = swh_storage.origin_visit_get_latest( | ||||
origin.url, type=data.type_visit1, | origin.url, type=data.type_visit1, | ||||
Show All 14 Lines | def test_origin_visit_get_latest_filter_type(self, swh_storage): | ||||
) | ) | ||||
is None | is None | ||||
) | ) | ||||
def test_origin_visit_get_latest(self, swh_storage): | def test_origin_visit_get_latest(self, swh_storage): | ||||
origin = Origin.from_dict(data.origin) | origin = Origin.from_dict(data.origin) | ||||
swh_storage.origin_add_one(origin) | swh_storage.origin_add_one(origin) | ||||
visit1 = OriginVisit( | visit1 = OriginVisit( | ||||
origin=origin.url, | origin=origin.url, date=data.date_visit1, type=data.type_visit1, | ||||
date=data.date_visit1, | |||||
type=data.type_visit1, | |||||
status="ongoing", | |||||
snapshot=None, | |||||
) | ) | ||||
visit2 = OriginVisit( | visit2 = OriginVisit( | ||||
origin=origin.url, | origin=origin.url, date=data.date_visit2, type=data.type_visit2, | ||||
date=data.date_visit2, | |||||
type=data.type_visit2, | |||||
status="ongoing", | |||||
snapshot=None, | |||||
) | ) | ||||
# Add a visit with the same date as the previous one | # Add a visit with the same date as the previous one | ||||
visit3 = OriginVisit( | visit3 = OriginVisit( | ||||
origin=origin.url, | origin=origin.url, date=data.date_visit2, type=data.type_visit2, | ||||
date=data.date_visit2, | |||||
type=data.type_visit2, | |||||
status="ongoing", | |||||
snapshot=None, | |||||
) | ) | ||||
ov1, ov2, ov3 = swh_storage.origin_visit_add([visit1, visit2, visit3]) | ov1, ov2, ov3 = swh_storage.origin_visit_add([visit1, visit2, visit3]) | ||||
origin_visit1 = swh_storage.origin_visit_get_by(origin.url, ov1.visit) | origin_visit1 = swh_storage.origin_visit_get_by(origin.url, ov1.visit) | ||||
origin_visit2 = swh_storage.origin_visit_get_by(origin.url, ov2.visit) | origin_visit2 = swh_storage.origin_visit_get_by(origin.url, ov2.visit) | ||||
origin_visit3 = swh_storage.origin_visit_get_by(origin.url, ov3.visit) | origin_visit3 = swh_storage.origin_visit_get_by(origin.url, ov3.visit) | ||||
# Two visits, both with no snapshot | # Two visits, both with no snapshot | ||||
Show All 13 Lines | def test_origin_visit_get_latest(self, swh_storage): | ||||
origin=origin.url, | origin=origin.url, | ||||
visit=ov1.visit, | visit=ov1.visit, | ||||
date=now(), | date=now(), | ||||
status="ongoing", | status="ongoing", | ||||
snapshot=complete_snapshot.id, | snapshot=complete_snapshot.id, | ||||
) | ) | ||||
] | ] | ||||
) | ) | ||||
assert { | actual_visit = swh_storage.origin_visit_get_latest( | ||||
origin.url, require_snapshot=True | |||||
) | |||||
assert actual_visit == { | |||||
**origin_visit1, | **origin_visit1, | ||||
"snapshot": complete_snapshot.id, | "snapshot": complete_snapshot.id, | ||||
} == swh_storage.origin_visit_get_latest(origin.url, require_snapshot=True) | "status": "ongoing", # visit1 has status created now | ||||
} | |||||
assert origin_visit3 == swh_storage.origin_visit_get_latest(origin.url) | assert origin_visit3 == swh_storage.origin_visit_get_latest(origin.url) | ||||
# Status filter: all three visits are status=ongoing, so no visit | # Status filter: all three visits are status=ongoing, so no visit | ||||
# returned | # returned | ||||
assert ( | assert ( | ||||
swh_storage.origin_visit_get_latest(origin.url, allowed_statuses=["full"]) | swh_storage.origin_visit_get_latest(origin.url, allowed_statuses=["full"]) | ||||
is None | is None | ||||
Show All 33 Lines | def test_origin_visit_get_latest(self, swh_storage): | ||||
status="ongoing", | status="ongoing", | ||||
snapshot=empty_snapshot.id, | snapshot=empty_snapshot.id, | ||||
) | ) | ||||
] | ] | ||||
) | ) | ||||
assert { | assert { | ||||
**origin_visit2, | **origin_visit2, | ||||
"snapshot": empty_snapshot.id, | "snapshot": empty_snapshot.id, | ||||
"status": "ongoing", | |||||
} == swh_storage.origin_visit_get_latest(origin.url, require_snapshot=True) | } == swh_storage.origin_visit_get_latest(origin.url, require_snapshot=True) | ||||
assert origin_visit3 == swh_storage.origin_visit_get_latest(origin.url) | assert origin_visit3 == swh_storage.origin_visit_get_latest(origin.url) | ||||
# Check that the status filter is still working | # Check that the status filter is still working | ||||
assert { | assert { | ||||
**origin_visit1, | **origin_visit1, | ||||
"snapshot": complete_snapshot.id, | "snapshot": complete_snapshot.id, | ||||
Show All 22 Lines | def test_origin_visit_get_latest(self, swh_storage): | ||||
"snapshot": complete_snapshot.id, | "snapshot": complete_snapshot.id, | ||||
"status": "full", | "status": "full", | ||||
} == swh_storage.origin_visit_get_latest( | } == swh_storage.origin_visit_get_latest( | ||||
origin.url, allowed_statuses=["full"], require_snapshot=True | origin.url, allowed_statuses=["full"], require_snapshot=True | ||||
) | ) | ||||
assert { | assert { | ||||
**origin_visit3, | **origin_visit3, | ||||
"snapshot": complete_snapshot.id, | "snapshot": complete_snapshot.id, | ||||
"status": "ongoing", | |||||
} == swh_storage.origin_visit_get_latest(origin.url) | } == swh_storage.origin_visit_get_latest(origin.url) | ||||
assert { | assert { | ||||
**origin_visit3, | **origin_visit3, | ||||
"snapshot": complete_snapshot.id, | "snapshot": complete_snapshot.id, | ||||
"status": "ongoing", | |||||
} == swh_storage.origin_visit_get_latest(origin.url, require_snapshot=True) | } == swh_storage.origin_visit_get_latest(origin.url, require_snapshot=True) | ||||
def test_origin_visit_status_get_latest(self, swh_storage): | def test_origin_visit_status_get_latest(self, swh_storage): | ||||
origin1 = Origin.from_dict(data.origin) | origin1 = Origin.from_dict(data.origin) | ||||
swh_storage.origin_add_one(data.origin) | swh_storage.origin_add_one(data.origin) | ||||
# to have some reference visits | # to have some reference visits | ||||
ov1, ov2 = swh_storage.origin_visit_add( | ov1, ov2 = swh_storage.origin_visit_add( | ||||
[ | [ | ||||
OriginVisit( | OriginVisit( | ||||
origin=origin1.url, | origin=origin1.url, date=data.date_visit1, type=data.type_visit1, | ||||
date=data.date_visit1, | |||||
type=data.type_visit1, | |||||
status="ongoing", | |||||
snapshot=None, | |||||
), | ), | ||||
OriginVisit( | OriginVisit( | ||||
origin=origin1.url, | origin=origin1.url, date=data.date_visit2, type=data.type_visit2, | ||||
date=data.date_visit2, | |||||
type=data.type_visit2, | |||||
status="ongoing", | |||||
snapshot=None, | |||||
), | ), | ||||
] | ] | ||||
) | ) | ||||
snapshot = Snapshot.from_dict(data.complete_snapshot) | snapshot = Snapshot.from_dict(data.complete_snapshot) | ||||
swh_storage.snapshot_add([snapshot]) | swh_storage.snapshot_add([snapshot]) | ||||
date_now = now() | date_now = now() | ||||
Show All 13 Lines | def test_origin_visit_status_get_latest(self, swh_storage): | ||||
visit=ov1.visit, | visit=ov1.visit, | ||||
date=data.date_visit2, | date=data.date_visit2, | ||||
status="ongoing", | status="ongoing", | ||||
snapshot=None, | snapshot=None, | ||||
) | ) | ||||
ovs3 = OriginVisitStatus( | ovs3 = OriginVisitStatus( | ||||
origin=origin1.url, | origin=origin1.url, | ||||
visit=ov2.visit, | visit=ov2.visit, | ||||
date=data.date_visit2, | date=data.date_visit2 + datetime.timedelta(minutes=1), # to not be ignored | ||||
ardumont: at it was initially the same date as the origin-visit-status from the origin-add call.
This one… | |||||
status="ongoing", | status="ongoing", | ||||
snapshot=None, | snapshot=None, | ||||
) | ) | ||||
ovs4 = OriginVisitStatus( | ovs4 = OriginVisitStatus( | ||||
origin=origin1.url, | origin=origin1.url, | ||||
visit=ov2.visit, | visit=ov2.visit, | ||||
date=date_now, | date=date_now, | ||||
status="full", | status="full", | ||||
▲ Show 20 Lines • Show All 74 Lines • ▼ Show 20 Lines | def test_person_fullname_unicity(self, swh_storage): | ||||
# check committers are the same | # check committers are the same | ||||
assert revisions[0]["committer"] == revisions[1]["committer"] | assert revisions[0]["committer"] == revisions[1]["committer"] | ||||
def test_snapshot_add_get_empty(self, swh_storage): | def test_snapshot_add_get_empty(self, swh_storage): | ||||
origin_url = swh_storage.origin_add_one(data.origin) | origin_url = swh_storage.origin_add_one(data.origin) | ||||
ov1 = swh_storage.origin_visit_add( | ov1 = swh_storage.origin_visit_add( | ||||
[ | [ | ||||
OriginVisit( | OriginVisit( | ||||
origin=origin_url, | origin=origin_url, date=data.date_visit1, type=data.type_visit1, | ||||
date=data.date_visit1, | |||||
type=data.type_visit1, | |||||
status="ongoing", | |||||
snapshot=None, | |||||
) | ) | ||||
] | ] | ||||
)[0] | )[0] | ||||
actual_result = swh_storage.snapshot_add([data.empty_snapshot]) | actual_result = swh_storage.snapshot_add([data.empty_snapshot]) | ||||
assert actual_result == {"snapshot:add": 1} | assert actual_result == {"snapshot:add": 1} | ||||
date_now = now() | date_now = now() | ||||
Show All 11 Lines | def test_snapshot_add_get_empty(self, swh_storage): | ||||
) | ) | ||||
by_id = swh_storage.snapshot_get(data.empty_snapshot["id"]) | by_id = swh_storage.snapshot_get(data.empty_snapshot["id"]) | ||||
assert by_id == {**data.empty_snapshot, "next_branch": None} | assert by_id == {**data.empty_snapshot, "next_branch": None} | ||||
by_ov = swh_storage.snapshot_get_by_origin_visit(origin_url, ov1.visit) | by_ov = swh_storage.snapshot_get_by_origin_visit(origin_url, ov1.visit) | ||||
assert by_ov == {**data.empty_snapshot, "next_branch": None} | assert by_ov == {**data.empty_snapshot, "next_branch": None} | ||||
data1 = { | ovs1 = OriginVisitStatus.from_dict( | ||||
{ | |||||
"origin": origin_url, | "origin": origin_url, | ||||
"date": data.date_visit1, | "date": data.date_visit1, | ||||
"visit": ov1.visit, | "visit": ov1.visit, | ||||
"status": "ongoing", | "status": "created", | ||||
"metadata": None, | |||||
"snapshot": None, | "snapshot": None, | ||||
"metadata": None, | |||||
} | } | ||||
data2 = { | ) | ||||
ovs2 = OriginVisitStatus.from_dict( | |||||
{ | |||||
"origin": origin_url, | "origin": origin_url, | ||||
"date": date_now, | "date": date_now, | ||||
"visit": ov1.visit, | "visit": ov1.visit, | ||||
"status": "full", | "status": "full", | ||||
"metadata": None, | "metadata": None, | ||||
"snapshot": data.empty_snapshot["id"], | "snapshot": data.empty_snapshot["id"], | ||||
} | } | ||||
) | |||||
actual_objects = list(swh_storage.journal_writer.journal.objects) | actual_objects = list(swh_storage.journal_writer.journal.objects) | ||||
assert actual_objects == [ | |||||
expected_objects = [ | |||||
("origin", Origin.from_dict(data.origin)), | ("origin", Origin.from_dict(data.origin)), | ||||
( | ("origin_visit", ov1), | ||||
"origin_visit", | ("origin_visit_status", ovs1,), | ||||
OriginVisit.from_dict({**data1, "type": data.type_visit1},), | |||||
), | |||||
("origin_visit_status", OriginVisitStatus.from_dict(data1)), | |||||
("snapshot", Snapshot.from_dict(data.empty_snapshot)), | ("snapshot", Snapshot.from_dict(data.empty_snapshot)), | ||||
("origin_visit_status", OriginVisitStatus.from_dict(data2),), | ("origin_visit_status", ovs2,), | ||||
] | ] | ||||
for obj in expected_objects: | |||||
assert obj in actual_objects | |||||
def test_snapshot_add_get_complete(self, swh_storage): | def test_snapshot_add_get_complete(self, swh_storage): | ||||
origin_url = data.origin["url"] | origin_url = data.origin["url"] | ||||
origin_url = swh_storage.origin_add_one(data.origin) | origin_url = swh_storage.origin_add_one(data.origin) | ||||
visit = OriginVisit( | visit = OriginVisit( | ||||
origin=origin_url, | origin=origin_url, | ||||
date=data.date_visit1, | date=data.date_visit1, | ||||
type=data.type_visit1, | type=data.type_visit1, | ||||
▲ Show 20 Lines • Show All 161 Lines • ▼ Show 20 Lines | def test_snapshot_add_get_paginated(self, swh_storage): | ||||
"next_branch": branch_names[dir_idx + 3], | "next_branch": branch_names[dir_idx + 3], | ||||
} | } | ||||
assert snapshot == expected_snapshot | assert snapshot == expected_snapshot | ||||
def test_snapshot_add_get_filtered(self, swh_storage): | def test_snapshot_add_get_filtered(self, swh_storage): | ||||
origin_url = swh_storage.origin_add_one(data.origin) | origin_url = swh_storage.origin_add_one(data.origin) | ||||
visit = OriginVisit( | visit = OriginVisit( | ||||
origin=origin_url, | origin=origin_url, date=data.date_visit1, type=data.type_visit1, | ||||
date=data.date_visit1, | |||||
type=data.type_visit1, | |||||
status="ongoing", | |||||
snapshot=None, | |||||
) | ) | ||||
origin_visit1 = swh_storage.origin_visit_add([visit])[0] | origin_visit1 = swh_storage.origin_visit_add([visit])[0] | ||||
swh_storage.snapshot_add([data.complete_snapshot]) | swh_storage.snapshot_add([data.complete_snapshot]) | ||||
swh_storage.origin_visit_status_add( | swh_storage.origin_visit_status_add( | ||||
[ | [ | ||||
OriginVisitStatus( | OriginVisitStatus( | ||||
origin=origin_url, | origin=origin_url, | ||||
▲ Show 20 Lines • Show All 103 Lines • ▼ Show 20 Lines | def test_snapshot_add_get_filtered_and_paginated(self, swh_storage): | ||||
"next_branch": b"release", | "next_branch": b"release", | ||||
} | } | ||||
assert snapshot == expected_snapshot | assert snapshot == expected_snapshot | ||||
def test_snapshot_add_get(self, swh_storage): | def test_snapshot_add_get(self, swh_storage): | ||||
origin_url = swh_storage.origin_add_one(data.origin) | origin_url = swh_storage.origin_add_one(data.origin) | ||||
visit = OriginVisit( | visit = OriginVisit( | ||||
origin=origin_url, | origin=origin_url, date=data.date_visit1, type=data.type_visit1, | ||||
date=data.date_visit1, | |||||
type=data.type_visit1, | |||||
status="ongoing", | |||||
snapshot=None, | |||||
) | ) | ||||
origin_visit1 = swh_storage.origin_visit_add([visit])[0] | origin_visit1 = swh_storage.origin_visit_add([visit])[0] | ||||
visit_id = origin_visit1.visit | visit_id = origin_visit1.visit | ||||
swh_storage.snapshot_add([data.snapshot]) | swh_storage.snapshot_add([data.snapshot]) | ||||
swh_storage.origin_visit_status_add( | swh_storage.origin_visit_status_add( | ||||
[ | [ | ||||
OriginVisitStatus( | OriginVisitStatus( | ||||
Show All 15 Lines | def test_snapshot_add_get(self, swh_storage): | ||||
origin_visit_info = swh_storage.origin_visit_get_by(origin_url, visit_id) | origin_visit_info = swh_storage.origin_visit_get_by(origin_url, visit_id) | ||||
assert origin_visit_info["snapshot"] == data.snapshot["id"] | assert origin_visit_info["snapshot"] == data.snapshot["id"] | ||||
def test_snapshot_add_twice__by_origin_visit(self, swh_storage): | def test_snapshot_add_twice__by_origin_visit(self, swh_storage): | ||||
origin_url = swh_storage.origin_add_one(data.origin) | origin_url = swh_storage.origin_add_one(data.origin) | ||||
ov1 = swh_storage.origin_visit_add( | ov1 = swh_storage.origin_visit_add( | ||||
[ | [ | ||||
OriginVisit( | OriginVisit( | ||||
origin=origin_url, | origin=origin_url, date=data.date_visit1, type=data.type_visit1, | ||||
date=data.date_visit1, | |||||
type=data.type_visit1, | |||||
status="ongoing", | |||||
snapshot=None, | |||||
) | ) | ||||
] | ] | ||||
)[0] | )[0] | ||||
swh_storage.snapshot_add([data.snapshot]) | swh_storage.snapshot_add([data.snapshot]) | ||||
date_now2 = now() | date_now2 = now() | ||||
swh_storage.origin_visit_status_add( | swh_storage.origin_visit_status_add( | ||||
[ | [ | ||||
OriginVisitStatus( | OriginVisitStatus( | ||||
origin=origin_url, | origin=origin_url, | ||||
visit=ov1.visit, | visit=ov1.visit, | ||||
date=date_now2, | date=date_now2, | ||||
status="ongoing", | status="ongoing", | ||||
snapshot=data.snapshot["id"], | snapshot=data.snapshot["id"], | ||||
) | ) | ||||
] | ] | ||||
) | ) | ||||
by_ov1 = swh_storage.snapshot_get_by_origin_visit(origin_url, ov1.visit) | by_ov1 = swh_storage.snapshot_get_by_origin_visit(origin_url, ov1.visit) | ||||
assert by_ov1 == {**data.snapshot, "next_branch": None} | assert by_ov1 == {**data.snapshot, "next_branch": None} | ||||
ov2 = swh_storage.origin_visit_add( | ov2 = swh_storage.origin_visit_add( | ||||
[ | [ | ||||
OriginVisit( | OriginVisit( | ||||
origin=origin_url, | origin=origin_url, date=data.date_visit2, type=data.type_visit2, | ||||
date=data.date_visit2, | |||||
type=data.type_visit2, | |||||
status="ongoing", | |||||
snapshot=None, | |||||
) | ) | ||||
] | ] | ||||
)[0] | )[0] | ||||
swh_storage.snapshot_add([data.snapshot]) | swh_storage.snapshot_add([data.snapshot]) | ||||
date_now4 = now() | date_now4 = now() | ||||
swh_storage.origin_visit_status_add( | swh_storage.origin_visit_status_add( | ||||
[ | [ | ||||
OriginVisitStatus( | OriginVisitStatus( | ||||
origin=origin_url, | origin=origin_url, | ||||
visit=ov2.visit, | visit=ov2.visit, | ||||
date=date_now4, | date=date_now4, | ||||
status="ongoing", | status="ongoing", | ||||
snapshot=data.snapshot["id"], | snapshot=data.snapshot["id"], | ||||
) | ) | ||||
] | ] | ||||
) | ) | ||||
by_ov2 = swh_storage.snapshot_get_by_origin_visit(origin_url, ov2.visit) | by_ov2 = swh_storage.snapshot_get_by_origin_visit(origin_url, ov2.visit) | ||||
assert by_ov2 == {**data.snapshot, "next_branch": None} | assert by_ov2 == {**data.snapshot, "next_branch": None} | ||||
data1 = { | ovs1 = OriginVisitStatus.from_dict( | ||||
{ | |||||
"origin": origin_url, | "origin": origin_url, | ||||
"date": data.date_visit1, | "date": data.date_visit1, | ||||
"visit": ov1.visit, | "visit": ov1.visit, | ||||
"status": "ongoing", | "status": "created", | ||||
"metadata": None, | "metadata": None, | ||||
"snapshot": None, | "snapshot": None, | ||||
} | } | ||||
data2 = { | ) | ||||
ovs2 = OriginVisitStatus.from_dict( | |||||
{ | |||||
"origin": origin_url, | "origin": origin_url, | ||||
"date": date_now2, | "date": date_now2, | ||||
"visit": ov1.visit, | "visit": ov1.visit, | ||||
"status": "ongoing", | "status": "ongoing", | ||||
"metadata": None, | "metadata": None, | ||||
"snapshot": data.snapshot["id"], | "snapshot": data.snapshot["id"], | ||||
} | } | ||||
data3 = { | ) | ||||
ovs3 = OriginVisitStatus.from_dict( | |||||
{ | |||||
"origin": origin_url, | "origin": origin_url, | ||||
"date": data.date_visit2, | "date": data.date_visit2, | ||||
"visit": ov2.visit, | "visit": ov2.visit, | ||||
"status": "ongoing", | "status": "created", | ||||
"metadata": None, | "metadata": None, | ||||
"snapshot": None, | "snapshot": None, | ||||
} | } | ||||
data4 = { | ) | ||||
ovs4 = OriginVisitStatus.from_dict( | |||||
{ | |||||
"origin": origin_url, | "origin": origin_url, | ||||
"date": date_now4, | "date": date_now4, | ||||
"visit": ov2.visit, | "visit": ov2.visit, | ||||
"status": "ongoing", | "status": "ongoing", | ||||
"metadata": None, | "metadata": None, | ||||
"snapshot": data.snapshot["id"], | "snapshot": data.snapshot["id"], | ||||
} | } | ||||
) | |||||
actual_objects = list(swh_storage.journal_writer.journal.objects) | actual_objects = list(swh_storage.journal_writer.journal.objects) | ||||
assert actual_objects == [ | expected_objects = [ | ||||
("origin", Origin.from_dict(data.origin)), | ("origin", Origin.from_dict(data.origin)), | ||||
( | ("origin_visit", ov1), | ||||
"origin_visit", | ("origin_visit_status", ovs1), | ||||
OriginVisit.from_dict({**data1, "type": data.type_visit1}), | |||||
), | |||||
("origin_visit_status", OriginVisitStatus.from_dict(data1)), | |||||
("snapshot", Snapshot.from_dict(data.snapshot)), | ("snapshot", Snapshot.from_dict(data.snapshot)), | ||||
("origin_visit_status", OriginVisitStatus.from_dict(data2),), | ("origin_visit_status", ovs2), | ||||
( | ("origin_visit", ov2), | ||||
"origin_visit", | ("origin_visit_status", ovs3), | ||||
OriginVisit.from_dict({**data3, "type": data.type_visit2}), | ("origin_visit_status", ovs4), | ||||
), | |||||
("origin_visit_status", OriginVisitStatus.from_dict(data3)), | |||||
("origin_visit_status", OriginVisitStatus.from_dict(data4),), | |||||
] | ] | ||||
for obj in expected_objects: | |||||
assert obj in actual_objects | |||||
def test_snapshot_get_random(self, swh_storage): | def test_snapshot_get_random(self, swh_storage): | ||||
swh_storage.snapshot_add( | swh_storage.snapshot_add( | ||||
[data.snapshot, data.empty_snapshot, data.complete_snapshot] | [data.snapshot, data.empty_snapshot, data.complete_snapshot] | ||||
) | ) | ||||
assert swh_storage.snapshot_get_random() in { | assert swh_storage.snapshot_get_random() in { | ||||
data.snapshot["id"], | data.snapshot["id"], | ||||
▲ Show 20 Lines • Show All 759 Lines • ▼ Show 20 Lines | def test_origin_count_with_visit_no_visits(self, swh_storage): | ||||
assert swh_storage.origin_count(".*user.*", regexp=False, with_visit=True) == 0 | assert swh_storage.origin_count(".*user.*", regexp=False, with_visit=True) == 0 | ||||
assert swh_storage.origin_count(".*user1.*", regexp=True, with_visit=True) == 0 | assert swh_storage.origin_count(".*user1.*", regexp=True, with_visit=True) == 0 | ||||
assert swh_storage.origin_count(".*user1.*", regexp=False, with_visit=True) == 0 | assert swh_storage.origin_count(".*user1.*", regexp=False, with_visit=True) == 0 | ||||
def test_origin_count_with_visit_with_visits_no_snapshot(self, swh_storage): | def test_origin_count_with_visit_with_visits_no_snapshot(self, swh_storage): | ||||
swh_storage.origin_add([{"url": url} for url in self.ORIGINS]) | swh_storage.origin_add([{"url": url} for url in self.ORIGINS]) | ||||
origin_url = "https://github.com/user1/repo1" | origin_url = "https://github.com/user1/repo1" | ||||
visit = OriginVisit( | visit = OriginVisit(origin=origin_url, date=now(), type="git",) | ||||
origin=origin_url, date=now(), type="git", status="ongoing", snapshot=None | |||||
) | |||||
swh_storage.origin_visit_add([visit]) | swh_storage.origin_visit_add([visit]) | ||||
assert swh_storage.origin_count("github", with_visit=False) == 3 | assert swh_storage.origin_count("github", with_visit=False) == 3 | ||||
# it has a visit, but no snapshot, so with_visit=True => 0 | # it has a visit, but no snapshot, so with_visit=True => 0 | ||||
assert swh_storage.origin_count("github", with_visit=True) == 0 | assert swh_storage.origin_count("github", with_visit=True) == 0 | ||||
assert swh_storage.origin_count("gitlab", with_visit=False) == 2 | assert swh_storage.origin_count("gitlab", with_visit=False) == 2 | ||||
# these gitlab origins have no visit | # these gitlab origins have no visit | ||||
assert swh_storage.origin_count("gitlab", with_visit=True) == 0 | assert swh_storage.origin_count("gitlab", with_visit=True) == 0 | ||||
assert ( | assert ( | ||||
swh_storage.origin_count("github.*user1", regexp=True, with_visit=False) | swh_storage.origin_count("github.*user1", regexp=True, with_visit=False) | ||||
== 1 | == 1 | ||||
) | ) | ||||
assert ( | assert ( | ||||
swh_storage.origin_count("github.*user1", regexp=True, with_visit=True) == 0 | swh_storage.origin_count("github.*user1", regexp=True, with_visit=True) == 0 | ||||
) | ) | ||||
assert swh_storage.origin_count("github", regexp=True, with_visit=True) == 0 | assert swh_storage.origin_count("github", regexp=True, with_visit=True) == 0 | ||||
def test_origin_count_with_visit_with_visits_and_snapshot(self, swh_storage): | def test_origin_count_with_visit_with_visits_and_snapshot(self, swh_storage): | ||||
swh_storage.origin_add([{"url": url} for url in self.ORIGINS]) | swh_storage.origin_add([{"url": url} for url in self.ORIGINS]) | ||||
swh_storage.snapshot_add([data.snapshot]) | swh_storage.snapshot_add([data.snapshot]) | ||||
origin_url = "https://github.com/user1/repo1" | origin_url = "https://github.com/user1/repo1" | ||||
visit = OriginVisit( | visit = OriginVisit(origin=origin_url, date=now(), type="git",) | ||||
origin=origin_url, date=now(), type="git", status="ongoing", snapshot=None | |||||
) | |||||
visit = swh_storage.origin_visit_add([visit])[0] | visit = swh_storage.origin_visit_add([visit])[0] | ||||
swh_storage.origin_visit_status_add( | swh_storage.origin_visit_status_add( | ||||
[ | [ | ||||
OriginVisitStatus( | OriginVisitStatus( | ||||
origin=origin_url, | origin=origin_url, | ||||
visit=visit.visit, | visit=visit.visit, | ||||
date=now(), | date=now(), | ||||
status="ongoing", | status="ongoing", | ||||
Show All 21 Lines | def test_add_arbitrary(self, swh_storage, objects): | ||||
for (obj_type, obj) in objects: | for (obj_type, obj) in objects: | ||||
obj = obj.to_dict() | obj = obj.to_dict() | ||||
if obj_type == "origin_visit": | if obj_type == "origin_visit": | ||||
origin_url = obj.pop("origin") | origin_url = obj.pop("origin") | ||||
swh_storage.origin_add_one({"url": origin_url}) | swh_storage.origin_add_one({"url": origin_url}) | ||||
if "visit" in obj: | if "visit" in obj: | ||||
del obj["visit"] | del obj["visit"] | ||||
visit = OriginVisit( | visit = OriginVisit( | ||||
origin=origin_url, | origin=origin_url, date=obj["date"], type=obj["type"], | ||||
date=obj["date"], | |||||
type=obj["type"], | |||||
status="ongoing", | |||||
snapshot=None, | |||||
) | ) | ||||
swh_storage.origin_visit_add([visit]) | swh_storage.origin_visit_add([visit]) | ||||
else: | else: | ||||
if obj_type == "content" and obj["status"] == "absent": | if obj_type == "content" and obj["status"] == "absent": | ||||
obj_type = "skipped_content" | obj_type = "skipped_content" | ||||
method = getattr(swh_storage, obj_type + "_add") | method = getattr(swh_storage, obj_type + "_add") | ||||
try: | try: | ||||
method([obj]) | method([obj]) | ||||
▲ Show 20 Lines • Show All 240 Lines • Show Last 20 Lines |
at it was initially the same date as the origin-visit-status from the origin-add call.
This one got dropped silently on conflict (origin, visit, date) ignore style.