Differential D4852 Diff 17179 swh/scheduler/tests/test_journal_client.py

Changeset View

Standalone View

swh/scheduler/tests/test_journal_client.py

This file was added.

				# Copyright (C) 2021 The Software Heritage developers
				# See the AUTHORS file at the top-level directory of this distribution
				# License: GNU General Public License version 3, or any later version
				# See top-level LICENSE file for more information

				import functools

				import pytest

				from swh.model.hashutil import hash_to_bytes
				from swh.model.model import OriginVisitStatus
				from swh.scheduler.journal_client import process_journal_objects
				from swh.scheduler.utils import utcnow


				def test_journal_client_origin_visit_status_from_journal_fail(swh_scheduler):
				process_fn = functools.partial(process_journal_objects, scheduler=swh_scheduler,)

				with pytest.raises(AssertionError, match="Only 'origin_visit_status'"):
				process_fn({"origin_visit": [{"url": "http://foobar.baz"},]})


				def test_journal_client_origin_visit_status_from_journal(swh_scheduler):
				process_fn = functools.partial(process_journal_objects, scheduler=swh_scheduler,)
				visit_statuses = [
				OriginVisitStatus( # eventful
				origin="foo",
				visit=1,
				status="full",
				date=utcnow(),
				type="git",
				snapshot=hash_to_bytes("d81cc0710eb6cf9efd5b920a8453e1e07157b6cd"),
				),
				OriginVisitStatus( # ignored due to status
				origin="zrb",
				visit=2,
				status="created",
				date=utcnow(),
				type="hg",
				snapshot=None,
				),
				OriginVisitStatus( # uneventful
				origin="bar",
				visit=1,
				status="partial",
				date=utcnow(),
				type="svn",
				snapshot=None,
				),
				OriginVisitStatus( # eventful
				origin="blue",
				visit=2,
				status="partial",
				date=utcnow(),
				type="tar",
				snapshot=hash_to_bytes("fffcc0710eb6cf9efd5b920a8453e1e07157b6cd"),
				),
				OriginVisitStatus( # ignored due to status
				origin="zrb",
				visit=2,
				status="ongoing",
				date=utcnow(),
				type="hg",
				snapshot=None,
				),
				]

				process_fn({"origin_visit_status": map(OriginVisitStatus.to_dict, visit_statuses)})

				counter = 0
				for ovs in visit_statuses:
				if ovs.status in ("ongoing", "created"):
				continue # those are skipped
				actual_origin_visit_stats = swh_scheduler.origin_visit_stats_get(
				ovs.origin, ovs.type
				)
				assert actual_origin_visit_stats is not None

				if ovs.status == "partial" and ovs.snapshot is not None:
				assert actual_origin_visit_stats.last_eventful == ovs.date
				assert actual_origin_visit_stats.last_failed is None
				assert actual_origin_visit_stats.last_uneventful is None
				elif ovs.status == "partial" and ovs.snapshot is None:
				assert actual_origin_visit_stats.last_eventful is None
				assert actual_origin_visit_stats.last_failed == ovs.date
				assert actual_origin_visit_stats.last_uneventful is None
				elif ovs.status == "full":
				assert actual_origin_visit_stats.last_eventful == ovs.date
				assert actual_origin_visit_stats.last_failed is None
				assert actual_origin_visit_stats.last_uneventful is None
				# status not-found cannot be tested as it's not part of the model yet

				counter += 1

				assert counter == len(
				[ovs for ovs in visit_statuses if ovs.status not in ("ongoing", "created")]
				)