Differential D5068 Diff 18095 swh/scheduler/tests/test_scheduler.py

Changeset View

Standalone View

swh/scheduler/tests/test_scheduler.py

# Copyright (C) 2017-2021 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, Tuple		from typing import Any, Dict, List, Optional, Tuple
import uuid		import uuid

import attr		import attr
		from psycopg2.extras import execute_values
import pytest		import pytest

from swh.model.hashutil import hash_to_bytes		from swh.model.hashutil import hash_to_bytes
from swh.scheduler.exc import SchedulerException, StaleData, UnknownPolicy		from swh.scheduler.exc import SchedulerException, StaleData, UnknownPolicy
from swh.scheduler.interface import ListedOriginPageToken, SchedulerInterface		from swh.scheduler.interface import ListedOriginPageToken, SchedulerInterface
from swh.scheduler.model import ListedOrigin, OriginVisitStats, SchedulerMetrics		from swh.scheduler.model import ListedOrigin, OriginVisitStats, SchedulerMetrics
from swh.scheduler.utils import utcnow		from swh.scheduler.utils import utcnow

▲ Show 20 Lines • Show All 915 Lines • ▼ Show 20 Lines	class TestScheduler:

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_stats_get_empty(self, swh_scheduler) -> None:		def test_origin_visit_stats_get_empty(self, swh_scheduler) -> None:
assert swh_scheduler.origin_visit_stats_get([]) == []		assert swh_scheduler.origin_visit_stats_get([]) == []

		def test_origin_visit_stats_get_pagination(self, swh_scheduler) -> None:
		page_size = inspect.signature(execute_values).parameters["page_size"].default

		visit_stats = [
		OriginVisitStats(
		url=f"https://example.com/origin-{i:03d}",
		visit_type="git",
		last_eventful=utcnow(),
		last_uneventful=None,
		last_failed=None,
		last_notfound=None,
		)
		for i in range(
		page_size + 1
		) # Ensure overflow of the psycopg2.extras.execute_values page_size
		]

		swh_scheduler.origin_visit_stats_upsert(visit_stats)

		assert set(
		swh_scheduler.origin_visit_stats_get(
		[(ovs.url, ovs.visit_type) for ovs in visit_stats]
		)
		) == set(visit_stats)

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

visit_stats = OriginVisitStats(		visit_stats = OriginVisitStats(
url=url,		url=url,
visit_type="git",		visit_type="git",
last_eventful=eventful_date,		last_eventful=eventful_date,
▲ Show 20 Lines • Show All 282 Lines • Show Last 20 Lines