diff --git a/swh/web/tests/common/test_origin_visits.py b/swh/web/tests/common/test_origin_visits.py index d66a4f2d..d3a54b08 100644 --- a/swh/web/tests/common/test_origin_visits.py +++ b/swh/web/tests/common/test_origin_visits.py @@ -1,225 +1,226 @@ # Copyright (C) 2018-2021 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 +from hypothesis import given, settings import iso8601 import pytest 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 +@settings(max_examples=1) @given(new_origin(), new_snapshots(3)) def test_get_origin_visits(mocker, archive_data, new_origin, new_snapshots): from swh.web.common import archive mocker.patch.object(archive, "MAX_LIMIT", 2) archive_data.origin_add([new_origin]) archive_data.snapshot_add(new_snapshots) for i, snapshot 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] visit_status = OriginVisitStatus( origin=new_origin.url, visit=visit.visit, date=visit_date + timedelta(minutes=5), status="full", snapshot=snapshot.id, ) archive_data.origin_visit_status_add([visit_status]) origin_visits = get_origin_visits(new_origin.to_dict()) assert len(origin_visits) == len(new_snapshots) @given(new_origin(), new_snapshots(5)) def test_get_origin_visit(archive_data, new_origin, new_snapshots): archive_data.origin_add([new_origin]) archive_data.snapshot_add(new_snapshots) visits = [] for i, visit_date in enumerate( map( iso8601.parse_date, [ "2015-07-09T21:09:24+00:00", "2016-02-23T18:05:23.312045+00:00", "2016-03-28T01:35:06.554111+00:00", "2016-06-18T01:22:24.808485+00:00", "2016-08-14T12:10:00.536702+00:00", ], ) ): visit = archive_data.origin_visit_add( [OriginVisit(origin=new_origin.url, date=visit_date, type="git",)] )[0] visits.append(visit) visit_status = OriginVisitStatus( origin=new_origin.url, visit=visit.visit, date=visit_date + timedelta(minutes=5), status="full", snapshot=new_snapshots[i].id, ) archive_data.origin_visit_status_add([visit_status]) origin_info = new_origin.to_dict() visit_id = visits[-1].visit + 1 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_id = visits[1].visit visit = get_origin_visit(origin_info, visit_id=visit_id) assert visit == archive_data.origin_visit_get_by(new_origin.url, visit_id=visit_id) visit = get_origin_visit(origin_info, visit_ts="2016-02-23T18:05:23.312045+00:00") assert visit == archive_data.origin_visit_get_by(new_origin.url, visit_id=visit_id) visit = get_origin_visit(origin_info, visit_ts="2016-02-20") assert visit == archive_data.origin_visit_get_by(new_origin.url, visit_id=visit_id) visit_id = visits[3].visit visit = get_origin_visit(origin_info, visit_ts="2016-06-18T01:22") assert visit == archive_data.origin_visit_get_by(new_origin.url, visit_id=visit_id) visit = get_origin_visit(origin_info, visit_ts="2016-06-18 01:22") assert visit == archive_data.origin_visit_get_by(new_origin.url, visit_id=visit_id) visit_id = visits[0].visit visit = get_origin_visit(origin_info, visit_ts="2014-01-01") assert visit == archive_data.origin_visit_get_by(new_origin.url, visit_id=visit_id) visit_id = visits[-1].visit visit = get_origin_visit(origin_info, visit_ts="2018-01-01") assert visit == archive_data.origin_visit_get_by(new_origin.url, visit_id=visit_id) @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 def test_get_origin_visit_latest_snapshot(mocker, origin_with_multiple_visits): origin_visits = get_origin_visits(origin_with_multiple_visits) 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_with_multiple_visits, snapshot_id=latest_visit["snapshot"] ) assert visit == latest_visit assert not mock_get_origin_visits.called visit = get_origin_visit( origin_with_multiple_visits, snapshot_id=first_visit["snapshot"] ) assert visit == first_visit assert mock_get_origin_visits.called