Changeset View
Changeset View
Standalone 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 | from typing import Any, Dict, List, Optional | ||||
| import uuid | import uuid | ||||
| import attr | import attr | ||||
| import pytest | import pytest | ||||
| from swh.model.hashutil import hash_to_bytes | |||||
| from swh.scheduler.exc import StaleData, UnknownPolicy | from swh.scheduler.exc import StaleData, UnknownPolicy | ||||
| from swh.scheduler.interface import SchedulerInterface | from swh.scheduler.interface import SchedulerInterface | ||||
| from swh.scheduler.model import ListedOrigin, ListedOriginPageToken, OriginVisitStats | from swh.scheduler.model import ListedOrigin, ListedOriginPageToken, OriginVisitStats | ||||
| 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) | ||||
| ▲ Show 20 Lines • Show All 793 Lines • ▼ Show 20 Lines | def test_origin_visit_stats_upsert(self, swh_scheduler) -> None: | ||||
| url=url, | url=url, | ||||
| visit_type="git", | visit_type="git", | ||||
| last_eventful=eventful_date, | last_eventful=eventful_date, | ||||
| last_uneventful=uneventful_date, | last_uneventful=uneventful_date, | ||||
| last_failed=failed_date, | last_failed=failed_date, | ||||
| ) | ) | ||||
| assert failed_visit == expected_visit_stats | assert failed_visit == expected_visit_stats | ||||
| def test_origin_visit_stats_upsert_with_snapshot(self, swh_scheduler) -> None: | |||||
| eventful_date = utcnow() | |||||
| url = "https://github.com/666/test" | |||||
| visit_stats = OriginVisitStats( | |||||
| url=url, | |||||
| visit_type="git", | |||||
| last_eventful=eventful_date, | |||||
| last_uneventful=None, | |||||
| last_failed=None, | |||||
| last_snapshot=hash_to_bytes("d81cc0710eb6cf9efd5b920a8453e1e07157b6cd"), | |||||
| ) | |||||
| swh_scheduler.origin_visit_stats_upsert(visit_stats) | |||||
| assert swh_scheduler.origin_visit_stats_get(url, "git") == visit_stats | |||||
| assert swh_scheduler.origin_visit_stats_get(url, "svn") is None | |||||