Page MenuHomeSoftware Heritage
Files F9346921

test_origin_visits.py
No OneTemporary
Actions

test_origin_visits.py
View Options

# Copyright (C) 2018-2019 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
from datetime import timedelta
from hypothesis import given
import pytest
from swh.model.hashutil import hash_to_hex
from swh.model.model import OriginVisit, OriginVisitStatus
from swh.storage.utils import now
from swh.web.common.exc import NotFoundExc
from swh.web.common.origin_visits import get_origin_visit, get_origin_visits
from swh.web.common.typing import OriginInfo
from swh.web.tests.strategies import (
new_origin,
new_snapshots,
origin_with_multiple_visits,
)
@given(new_snapshots(3))
def test_get_origin_visits(mocker, snapshots):
mock_archive = mocker.patch("swh.web.common.archive")
mock_archive.MAX_LIMIT = 2
def _lookup_origin_visits(*args, **kwargs):
if kwargs["last_visit"] is None:
return [
{
"visit": 1,
"date": "2017-05-06T00:59:10+00:00",
"status": "full",
"snapshot": hash_to_hex(snapshots[0].id),
"type": "git",
},
{
"visit": 2,
"date": "2017-08-06T00:59:10+00:00",
"status": "full",
"snapshot": hash_to_hex(snapshots[1].id),
"type": "git",
},
]
else:
return [
{
"visit": 3,
"date": "2017-09-06T00:59:10+00:00",
"status": "full",
"snapshot": hash_to_hex(snapshots[2].id),
"type": "git",
}
]
mock_archive.lookup_origin_visits.side_effect = _lookup_origin_visits
origin_info = {
"url": "https://github.com/foo/bar",
}
origin_visits = get_origin_visits(origin_info)
assert len(origin_visits) == 3
@given(new_snapshots(5))
def test_get_origin_visit(mocker, snapshots):
mock_origin_visits = mocker.patch("swh.web.common.origin_visits.get_origin_visits")
origin_info = {
"url": "https://github.com/foo/bar",
}
visits = [
{
"status": "full",
"date": "2015-07-09T21:09:24+00:00",
"visit": 1,
"origin": "https://github.com/foo/bar",
"type": "git",
"snapshot": hash_to_hex(snapshots[0].id),
},
{
"status": "full",
"date": "2016-02-23T18:05:23.312045+00:00",
"visit": 2,
"origin": "https://github.com/foo/bar",
"type": "git",
"snapshot": hash_to_hex(snapshots[1].id),
},
{
"status": "full",
"date": "2016-03-28T01:35:06.554111+00:00",
"visit": 3,
"origin": "https://github.com/foo/bar",
"type": "git",
"snapshot": hash_to_hex(snapshots[2].id),
},
{
"status": "full",
"date": "2016-06-18T01:22:24.808485+00:00",
"visit": 4,
"origin": "https://github.com/foo/bar",
"type": "git",
"snapshot": hash_to_hex(snapshots[3].id),
},
{
"status": "full",
"date": "2016-08-14T12:10:00.536702+00:00",
"visit": 5,
"origin": "https://github.com/foo/bar",
"type": "git",
"snapshot": hash_to_hex(snapshots[4].id),
},
]
mock_origin_visits.return_value = visits
visit_id = 12
with pytest.raises(NotFoundExc) as e:
visit = get_origin_visit(origin_info, visit_id=visit_id)
assert e.match("Visit with id %s" % visit_id)
assert e.match("url %s" % origin_info["url"])
visit = get_origin_visit(origin_info, visit_id=2)
assert visit == visits[1]
visit = get_origin_visit(origin_info, visit_ts="2016-02-23T18:05:23.312045+00:00")
assert visit == visits[1]
visit = get_origin_visit(origin_info, visit_ts="2016-02-20")
assert visit == visits[1]
visit = get_origin_visit(origin_info, visit_ts="2016-06-18T01:22")
assert visit == visits[3]
visit = get_origin_visit(origin_info, visit_ts="2016-06-18 01:22")
assert visit == visits[3]
visit = get_origin_visit(origin_info, visit_ts="2014-01-01")
assert visit == visits[0]
visit = get_origin_visit(origin_info, visit_ts="2018-01-01")
assert visit == visits[-1]
@given(new_origin(), new_snapshots(6))
def test_get_origin_visit_return_first_valid_full_visit(
archive_data, new_origin, new_snapshots
):
visits = []
archive_data.origin_add([new_origin])
# create 6 visits, the first three have full status while the
# last three have partial status and set a null snapshot for
# the last four visits
for i, snp in enumerate(new_snapshots):
visit_date = now() + timedelta(days=i * 10)
visit = archive_data.origin_visit_add(
[OriginVisit(origin=new_origin.url, date=visit_date, type="git",)]
)[0]
archive_data.snapshot_add([new_snapshots[i]])
visit_status = OriginVisitStatus(
origin=new_origin.url,
visit=visit.visit,
date=visit_date + timedelta(minutes=5),
status="full" if i < 3 else "partial",
snapshot=new_snapshots[i].id if i < 2 else None,
)
if i < 2:
archive_data.origin_visit_status_add([visit_status])
visits.append(visit.visit)
# should return the second visit
expected_visit = archive_data.origin_visit_get_by(new_origin.url, visits[1])
assert get_origin_visit((OriginInfo(url=new_origin.url))) == expected_visit
@given(new_origin(), new_snapshots(6))
def test_get_origin_visit_non_resolvable_snapshots(
archive_data, new_origin, new_snapshots
):
visits = []
archive_data.origin_add([new_origin])
# create 6 full visits, the first three have resolvable snapshots
# while the last three have non resolvable snapshots
for i, snp in enumerate(new_snapshots):
visit_date = now() + timedelta(days=i * 10)
visit = archive_data.origin_visit_add(
[OriginVisit(origin=new_origin.url, date=visit_date, type="git",)]
)[0]
archive_data.snapshot_add([new_snapshots[i]])
visit_status = OriginVisitStatus(
origin=new_origin.url,
visit=visit.visit,
date=visit_date + timedelta(minutes=5),
status="full",
snapshot=new_snapshots[i].id,
)
if i < 3:
archive_data.origin_visit_status_add([visit_status])
visits.append(visit.visit)
# should return the third visit
expected_visit = archive_data.origin_visit_get_by(new_origin.url, visits[2])
assert get_origin_visit((OriginInfo(url=new_origin.url))) == expected_visit
@given(new_origin(), new_snapshots(6))
def test_get_origin_visit_return_first_valid_partial_visit(
archive_data, new_origin, new_snapshots
):
visits = []
archive_data.origin_add([new_origin])
# create 6 visits, the first three have full status but null snapshot
# while the last three have partial status with valid snapshot
for i, snp in enumerate(new_snapshots):
visit_date = now() + timedelta(days=i * 10)
visit = archive_data.origin_visit_add(
[OriginVisit(origin=new_origin.url, date=visit_date, type="git",)]
)[0]
archive_data.snapshot_add([new_snapshots[i]])
visit_status = OriginVisitStatus(
origin=new_origin.url,
visit=visit.visit,
date=visit_date + timedelta(minutes=5),
status="full" if i < 3 else "partial",
snapshot=new_snapshots[i].id if i > 2 else None,
)
if i > 2:
archive_data.origin_visit_status_add([visit_status])
visits.append(visit.visit)
# should return the last visit
expected_visit = archive_data.origin_visit_get_by(new_origin.url, visits[-1])
assert get_origin_visit((OriginInfo(url=new_origin.url))) == expected_visit
@given(origin_with_multiple_visits())
def test_get_origin_visit_latest_snapshot(mocker, origin):
origin_visits = get_origin_visits(origin)
first_visit = origin_visits[0]
latest_visit = origin_visits[-1]
mock_get_origin_visits = mocker.patch(
"swh.web.common.origin_visits.get_origin_visits"
)
mock_get_origin_visits.return_value = origin_visits
visit = get_origin_visit(origin, snapshot_id=latest_visit["snapshot"])
assert visit == latest_visit
assert not mock_get_origin_visits.called
visit = get_origin_visit(origin, snapshot_id=first_visit["snapshot"])
assert visit == first_visit
assert mock_get_origin_visits.called

File Metadata

Mime Type
text/x-python
Expires
Fri, Jul 4, 4:39 PM (2 w, 5 d ago)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
3293497

Event Timeline

About · Developer information · Legal