Differential D6011 Diff 21723 swh/search/tests/test_journal_client.py

Changeset 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		from datetime import datetime, timezone
import functools		import functools
from unittest.mock import MagicMock		from unittest.mock import MagicMock

		import pytest

from swh.model.model import (		from swh.model.model import (
ObjectType,		ObjectType,
Person,		Person,
Release,		Release,
Revision,		Revision,
RevisionType,		RevisionType,
Snapshot,		Snapshot,
SnapshotBranch,		SnapshotBranch,
TargetType,		TargetType,
Timestamp,		Timestamp,
TimestampWithTimezone,		TimestampWithTimezone,
hash_to_bytes,		hash_to_bytes,
)		)
from swh.search.journal_client import process_journal_objects		from swh.search.journal_client import (
		fetch_last_revision_release_date,
		process_journal_objects,
		)
from swh.storage import get_storage		from swh.storage import get_storage

DATES = [		DATES = [
TimestampWithTimezone(		TimestampWithTimezone(
timestamp=Timestamp(seconds=1234567891, microseconds=0,),		timestamp=Timestamp(seconds=1234567891, microseconds=0,),
offset=120,		offset=120,
negative_utc=False,		negative_utc=False,
),		),
▲ Show 20 Lines • Show All 113 Lines • ▼ Show 20 Lines	Snapshot(
b"target/release3": SnapshotBranch(		b"target/release3": SnapshotBranch(
target_type=TargetType.RELEASE, target=RELEASES[2].id		target_type=TargetType.RELEASE, target=RELEASES[2].id
),		),
b"target/alias": SnapshotBranch(		b"target/alias": SnapshotBranch(
target_type=TargetType.ALIAS, target=b"target/revision1"		target_type=TargetType.ALIAS, target=b"target/revision1"
),		),
},		},
),		),
		Snapshot(
		branches={
		b"target/revision1": SnapshotBranch(
		target_type=TargetType.REVISION, target=REVISIONS[0].id,
		)
		},
		),
		Snapshot(
		branches={
		b"target/release1": SnapshotBranch(
		target_type=TargetType.RELEASE, target=RELEASES[0].id
		)
		},
		),
		Snapshot(branches={}),
]		]


		@pytest.fixture
		def storage():
		storage = get_storage("memory")

		storage.revision_add(REVISIONS)
		storage.release_add(RELEASES)
		storage.snapshot_add(SNAPSHOTS)
		return storage


def test_journal_client_origin_from_journal():		def test_journal_client_origin_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,)

worker_fn({"origin": [{"url": "http://foobar.baz"},]})		worker_fn({"origin": [{"url": "http://foobar.baz"},]})
search_mock.origin_update.assert_called_once_with(		search_mock.origin_update.assert_called_once_with(
[{"url": "http://foobar.baz"},]		[{"url": "http://foobar.baz"},]
Show All 13 Lines	def test_journal_client_origin_visit_from_journal():
worker_fn = functools.partial(process_journal_objects, search=search_mock,)		worker_fn = functools.partial(process_journal_objects, search=search_mock,)

worker_fn({"origin_visit": [{"origin": "http://foobar.baz", "type": "git"},]})		worker_fn({"origin_visit": [{"origin": "http://foobar.baz", "type": "git"},]})
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(storage):
search_mock = MagicMock()		search_mock = MagicMock()
storage = get_storage("memory")

storage.revision_add(REVISIONS)
storage.release_add(RELEASES)
storage.snapshot_add(SNAPSHOTS)

worker_fn = functools.partial(		worker_fn = functools.partial(
process_journal_objects, search=search_mock, storage=storage		process_journal_objects, search=search_mock, storage=storage
)		)
current_datetime = datetime.now(tz=timezone.utc)		current_datetime = datetime.now(tz=timezone.utc)

worker_fn(		worker_fn(
{		{
▲ Show 20 Lines • Show All 69 Lines • ▼ Show 20 Lines	search_mock.origin_update.assert_called_once_with(
"@context": "https://doi.org/10.5063/schema/codemeta-2.0",		"@context": "https://doi.org/10.5063/schema/codemeta-2.0",
"description": "foo bar",		"description": "foo bar",
"programmingLanguage": "python",		"programmingLanguage": "python",
"license": "MIT",		"license": "MIT",
},		},
},		},
]		]
)		)


		def test_fetch_last_revision_release_date(storage):
		for snapshot in SNAPSHOTS:
		assert fetch_last_revision_release_date(snapshot.id, storage) is not None