Differential D4838 Diff 17138 swh/scheduler/tests/test_scheduler.py

Changeset View

Standalone View

swh/scheduler/tests/test_scheduler.py

# Copyright (C) 2017-2019 The Software Heritage developers		# Copyright (C) 2017-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

from collections import defaultdict		from collections import defaultdict
import copy		import copy
import datetime		import datetime
import inspect		import inspect
import random		import random
from typing import Any, Dict, List, Optional		from typing import Any, Dict, List, Optional
import uuid		import uuid

import attr		import attr
import pytest		import pytest

from swh.scheduler.exc import StaleData		from swh.scheduler.exc import StaleData
from swh.scheduler.interface import SchedulerInterface		from swh.scheduler.interface import SchedulerInterface
from swh.scheduler.model import ListedOrigin, ListedOriginPageToken		from swh.scheduler.model import ListedOrigin, ListedOriginPageToken, OriginVisitInfo
from swh.scheduler.utils import utcnow		from swh.scheduler.utils import utcnow

from .common import LISTERS, TASK_TYPES, TEMPLATES, tasks_from_template		from .common import LISTERS, TASK_TYPES, TEMPLATES, tasks_from_template

ONEDAY = datetime.timedelta(days=1)		ONEDAY = datetime.timedelta(days=1)


def subdict(d, keys=None, excl=()):		def subdict(d, keys=None, excl=()):
▲ Show 20 Lines • Show All 696 Lines • ▼ Show 20 Lines	def test_get_listed_origins_all(self, swh_scheduler, listed_origins) -> None:

ret = swh_scheduler.get_listed_origins(limit=len(listed_origins) + 1)		ret = swh_scheduler.get_listed_origins(limit=len(listed_origins) + 1)
assert ret.next_page_token is None		assert ret.next_page_token is None
assert len(ret.origins) == len(listed_origins)		assert len(ret.origins) == len(listed_origins)

def _create_task_types(self, scheduler):		def _create_task_types(self, scheduler):
for tt in TASK_TYPES.values():		for tt in TASK_TYPES.values():
scheduler.create_task_type(tt)		scheduler.create_task_type(tt)

		def test_origin_visit_info_upsert(self, swh_scheduler) -> None:
		eventful_date = utcnow()
		url = "https://github.com/test"

		visit_info = OriginVisitInfo(
		url=url,
		visit_type="git",
		eventful=eventful_date,
		uneventful=None,
		failed=None,
		)
		swh_scheduler.origin_visit_info_upsert(visit_info)
		swh_scheduler.origin_visit_info_upsert(visit_info)

		assert swh_scheduler.origin_visit_info_get(url, "git") == visit_info
		assert swh_scheduler.origin_visit_info_get(url, "svn") is None

		uneventful_date = utcnow()
		visit_info = OriginVisitInfo(
		url=url,
		visit_type="git",
		eventful=None,
		uneventful=uneventful_date,
		failed=None,
		)
		swh_scheduler.origin_visit_info_upsert(visit_info)

		uneventful_visit = swh_scheduler.origin_visit_info_get(url, "git")

		expected_visit_info = OriginVisitInfo(
		url=url,
		visit_type="git",
		eventful=eventful_date,
		uneventful=uneventful_date,
		failed=None,
		)

		assert uneventful_visit == expected_visit_info

		failed_date = utcnow()
		visit_info = OriginVisitInfo(
		url=url,
		visit_type="git",
		eventful=None,
		uneventful=None,
		failed=failed_date,
		)
		swh_scheduler.origin_visit_info_upsert(visit_info)

		failed_visit = swh_scheduler.origin_visit_info_get(url, "git")

		expected_visit_info = OriginVisitInfo(
		url=url,
		visit_type="git",
		eventful=eventful_date,
		uneventful=uneventful_date,
		failed=failed_date,
		)

		assert failed_visit == expected_visit_info