Changeset View
Changeset View
Standalone View
Standalone View
swh/scheduler/tests/test_journal_client.py
# Copyright (C) 2021 The Software Heritage developers | # Copyright (C) 2021 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 datetime | import datetime | ||||
import functools | import functools | ||||
from itertools import permutations | from itertools import permutations | ||||
from unittest.mock import Mock | from unittest.mock import Mock | ||||
import pytest | import pytest | ||||
from swh.model.hashutil import hash_to_bytes | from swh.model.hashutil import hash_to_bytes | ||||
from swh.scheduler.journal_client import max_date, process_journal_objects | from swh.scheduler.journal_client import ( | ||||
from_position_offset_to_days, | |||||
max_date, | |||||
process_journal_objects, | |||||
) | |||||
from swh.scheduler.model import ListedOrigin, OriginVisitStats | from swh.scheduler.model import ListedOrigin, OriginVisitStats | ||||
from swh.scheduler.utils import utcnow | from swh.scheduler.utils import utcnow | ||||
def test_journal_client_origin_visit_status_from_journal_fail(swh_scheduler): | def test_journal_client_origin_visit_status_from_journal_fail(swh_scheduler): | ||||
process_fn = functools.partial(process_journal_objects, scheduler=swh_scheduler,) | process_fn = functools.partial(process_journal_objects, scheduler=swh_scheduler,) | ||||
with pytest.raises(AssertionError, match="Got unexpected origin_visit"): | with pytest.raises(AssertionError, match="Got unexpected origin_visit"): | ||||
▲ Show 20 Lines • Show All 94 Lines • ▼ Show 20 Lines | def test_journal_client_ignore_missing_type(swh_scheduler): | ||||
process_journal_objects( | process_journal_objects( | ||||
{"origin_visit_status": visit_statuses}, scheduler=swh_scheduler | {"origin_visit_status": visit_statuses}, scheduler=swh_scheduler | ||||
) | ) | ||||
# The message has been ignored: no stats have been upserted | # The message has been ignored: no stats have been upserted | ||||
swh_scheduler.origin_visit_stats_upsert.assert_not_called() | swh_scheduler.origin_visit_stats_upsert.assert_not_called() | ||||
olasd: (We'll add typing to this in a further diff) | |||||
def test_journal_client_origin_visit_status_from_journal_last_notfound(swh_scheduler): | def test_journal_client_origin_visit_status_from_journal_last_notfound(swh_scheduler): | ||||
visit_status = { | visit_status = { | ||||
"origin": "foo", | "origin": "foo", | ||||
"visit": 1, | "visit": 1, | ||||
"status": "not_found", | "status": "not_found", | ||||
"date": DATE1, | "date": DATE1, | ||||
"type": "git", | "type": "git", | ||||
"snapshot": None, | "snapshot": None, | ||||
▲ Show 20 Lines • Show All 625 Lines • ▼ Show 20 Lines | assert swh_scheduler.origin_visit_stats_get([("cavabarder", "hg")]) == [ | ||||
last_uneventful=DATE1 + 2 * ONE_YEAR, | last_uneventful=DATE1 + 2 * ONE_YEAR, | ||||
last_failed=None, | last_failed=None, | ||||
last_notfound=None, | last_notfound=None, | ||||
last_snapshot=hash_to_bytes("aaaaaabbbeb6cf9efd5b920a8453e1e07157b6cd"), | last_snapshot=hash_to_bytes("aaaaaabbbeb6cf9efd5b920a8453e1e07157b6cd"), | ||||
next_visit_queue_position=None, | next_visit_queue_position=None, | ||||
next_position_offset=6, # 2 uneventful visits, whatever the permutation | next_position_offset=6, # 2 uneventful visits, whatever the permutation | ||||
) | ) | ||||
] | ] | ||||
@pytest.mark.parametrize( | |||||
"position_offset, interval", | |||||
[ | |||||
(0, 1), | |||||
(1, 1), | |||||
(2, 2), | |||||
(3, 2), | |||||
(4, 2), | |||||
(5, 4), | |||||
(6, 16), | |||||
(7, 64), | |||||
(8, 256), | |||||
(9, 1024), | |||||
(10, 4096), | |||||
], | |||||
) | |||||
def test_journal_client_from_position_offset_to_days(position_offset, interval): | |||||
assert from_position_offset_to_days(position_offset) == interval | |||||
def test_journal_client_from_position_offset_to_days_only_positive_input(): | |||||
with pytest.raises(AssertionError): | |||||
from_position_offset_to_days(-1) |
(We'll add typing to this in a further diff)