Changeset View
Changeset View
Standalone View
Standalone View
swh/search/tests/test_journal_client.py
# Copyright (C) 2019-2021 The Software Heritage developers | # Copyright (C) 2019-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 datetime import datetime, timezone | |||||
import functools | import functools | ||||
from unittest.mock import MagicMock | from unittest.mock import MagicMock | ||||
from swh.search.journal_client import process_journal_objects | from swh.search.journal_client import process_journal_objects | ||||
def test_journal_client_origin_from_journal(): | def test_journal_client_origin_from_journal(): | ||||
search_mock = MagicMock() | search_mock = MagicMock() | ||||
Show All 22 Lines | def test_journal_client_origin_visit_from_journal(): | ||||
search_mock.origin_update.assert_called_once_with( | search_mock.origin_update.assert_called_once_with( | ||||
[{"url": "http://foobar.baz", "visit_types": ["git"]},] | [{"url": "http://foobar.baz", "visit_types": ["git"]},] | ||||
) | ) | ||||
def test_journal_client_origin_visit_status_from_journal(): | def test_journal_client_origin_visit_status_from_journal(): | ||||
search_mock = MagicMock() | search_mock = MagicMock() | ||||
worker_fn = functools.partial(process_journal_objects, search=search_mock,) | worker_fn = functools.partial(process_journal_objects, search=search_mock,) | ||||
ardumont: same remark. | |||||
current_datetime = datetime.now(tz=timezone.utc) | |||||
worker_fn( | worker_fn( | ||||
{ | { | ||||
"origin_visit_status": [ | "origin_visit_status": [ | ||||
{"origin": "http://foobar.baz", "status": "full"} # full visits ok | { | ||||
"origin": "http://foobar.baz", | |||||
"status": "full", | |||||
"visit": 5, | |||||
"date": current_datetime, | |||||
} # full visits ok | |||||
] | ] | ||||
} | } | ||||
) | ) | ||||
search_mock.origin_update.assert_called_once_with( | search_mock.origin_update.assert_called_once_with( | ||||
[{"url": "http://foobar.baz", "has_visits": True},] | [ | ||||
{ | |||||
"url": "http://foobar.baz", | |||||
"has_visits": True, | |||||
"nb_visit": 5, | |||||
"last_visit_date": current_datetime.isoformat(), | |||||
}, | |||||
] | |||||
) | ) | ||||
search_mock.reset_mock() | search_mock.reset_mock() | ||||
# non-full visits are filtered out | # non-full visits are filtered out | ||||
worker_fn( | worker_fn( | ||||
{"origin_visit_status": [{"origin": "http://foobar.baz", "status": "partial"}]} | { | ||||
"origin_visit_status": [ | |||||
{ | |||||
"origin": "http://foobar.baz", | |||||
"status": "partial", | |||||
"visit": 5, | |||||
"date": current_datetime, | |||||
} | |||||
] | |||||
} | |||||
) | ) | ||||
search_mock.origin_update.assert_not_called() | search_mock.origin_update.assert_not_called() | ||||
def test_journal_client_origin_metadata_from_journal(): | def test_journal_client_origin_metadata_from_journal(): | ||||
search_mock = MagicMock() | search_mock = MagicMock() | ||||
worker_fn = functools.partial(process_journal_objects, search=search_mock,) | worker_fn = functools.partial(process_journal_objects, search=search_mock,) | ||||
Show All 25 Lines |
same remark.