Changeset View
Changeset View
Standalone View
Standalone View
swh/scheduler/tests/test_model.py
# Copyright (C) 2020-2021 The Software Heritage developers | # Copyright (C) 2020-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 | ||||
import uuid | |||||
import attr | import attr | ||||
from swh.scheduler import model | from swh.scheduler import model | ||||
def test_select_columns(): | def test_select_columns(): | ||||
@attr.s | @attr.s | ||||
▲ Show 20 Lines • Show All 72 Lines • ▼ Show 20 Lines | def test_insert_primary_key(): | ||||
@attr.s | @attr.s | ||||
class TestModel2(model.BaseSchedulerModel): | class TestModel2(model.BaseSchedulerModel): | ||||
col1 = attr.ib(type=str, metadata={"primary_key": True}) | col1 = attr.ib(type=str, metadata={"primary_key": True}) | ||||
col2 = attr.ib(type=str, metadata={"primary_key": True}) | col2 = attr.ib(type=str, metadata={"primary_key": True}) | ||||
test1 = attr.ib(type=str) | test1 = attr.ib(type=str) | ||||
assert TestModel2.primary_key_columns() == ("col1", "col2") | assert TestModel2.primary_key_columns() == ("col1", "col2") | ||||
def test_listed_origin_as_task_dict(): | |||||
origin = model.ListedOrigin( | |||||
lister_id=uuid.uuid4(), | |||||
url="http://example.com/", | |||||
visit_type="git", | |||||
) | |||||
task = origin.as_task_dict() | |||||
assert task == { | |||||
"type": "load-git", | |||||
"arguments": {"args": [], "kwargs": {"url": "http://example.com/"}}, | |||||
} | |||||
loader_args = {"foo": "bar", "baz": {"foo": "bar"}} | |||||
origin_w_args = model.ListedOrigin( | |||||
lister_id=uuid.uuid4(), | |||||
url="http://example.com/svn/", | |||||
visit_type="svn", | |||||
extra_loader_arguments=loader_args, | |||||
) | |||||
task_w_args = origin_w_args.as_task_dict() | |||||
assert task_w_args == { | |||||
"type": "load-svn", | |||||
"arguments": { | |||||
"args": [], | |||||
"kwargs": {"url": "http://example.com/svn/", **loader_args}, | |||||
}, | |||||
} |