Differential D6017 Diff 21766 swh/scheduler/tests/test_journal_client.py

Changeset 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
from datetime import timedelta		from datetime import timedelta
import functools		import functools
from itertools import permutations		from itertools import permutations
		from typing import List
from unittest.mock import Mock		from unittest.mock import Mock

import attr		import attr
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 (		from swh.scheduler.journal_client import (
from_position_offset_to_days,		from_position_offset_to_days,
▲ Show 20 Lines • Show All 107 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()


def assert_visit_stats_ok(actual_visit_stats, expected_visit_stats):		def assert_visit_stats_ok(
		actual_visit_stats: List[OriginVisitStats],
		expected_visit_stats: List[OriginVisitStats],
		ignore_fields: List[str] = ["next_visit_queue_position"],
		):
"""Utility test function to ensure visits stats read from the backend are in the right		"""Utility test function to ensure visits stats read from the backend are in the right
shape. The comparison on the next_visit_queue_position will be dealt with in		shape. The comparison on the next_visit_queue_position will be dealt with in
dedicated tests so it's not tested in tests that are calling this function.		dedicated tests so it's not tested in tests that are calling this function.

"""		"""
assert len(actual_visit_stats) == len(expected_visit_stats)		assert len(actual_visit_stats) == len(expected_visit_stats)

for visit_stats in actual_visit_stats:		fields = attr.fields_dict(OriginVisitStats)
visit_stats = attr.evolve(visit_stats, next_visit_queue_position=None)		defaults = {field: fields[field].default for field in ignore_fields}

		for visit_stats in actual_visit_stats:
		visit_stats = attr.evolve(visit_stats, **defaults)
assert visit_stats in expected_visit_stats		assert visit_stats in expected_visit_stats


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",
▲ Show 20 Lines • Show All 758 Lines • Show Last 20 Lines