Differential D6065 Diff 21993 swh/scanner/tests/test_policy.py

Changeset View

Standalone View

swh/scanner/tests/test_policy.py

# Copyright (C) 2021 The Software Heritage developers		# Copyright (C) 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 json		import json

from flask import url_for		from flask import url_for
import pytest		import pytest

from swh.model.identifiers import CoreSWHID, ObjectType		from swh.model.identifiers import CONTENT, CoreSWHID, ObjectType
from swh.scanner.data import MerkleNodeInfo		from swh.scanner.data import MerkleNodeInfo
from swh.scanner.exceptions import APIError		from swh.scanner.exceptions import APIError
from swh.scanner.policy import (		from swh.scanner.policy import (
DirectoryPriority,		DirectoryPriority,
FilePriority,		FilePriority,
		GreedyBFS,
LazyBFS,		LazyBFS,
		source_size,
swhids_discovery,		swhids_discovery,
)		)

from .data import correct_api_response		from .data import correct_api_response

aio_url = "http://example.org/api/known/"		aio_url = "http://example.org/api/known/"


Show All 16 Lines	def test_scanner_raise_apierror(mock_aioresponse, event_loop, aiosession):
mock_aioresponse.post(aio_url, content_type="application/json", status=413)		mock_aioresponse.post(aio_url, content_type="application/json", status=413)

with pytest.raises(APIError):		with pytest.raises(APIError):
event_loop.run_until_complete(		event_loop.run_until_complete(
swhids_discovery([], aiosession, "http://example.org/api/")		swhids_discovery([], aiosession, "http://example.org/api/")
)		)


def test_scanner_raise_apierror_input_size_limit(event_loop, aiosession, live_server):

api_url = url_for("index", _external=True)
request = [
"swh:1:cnt:7c4c57ba9ff496ad179b8f65b1d286edbda34c9a" for i in range(901)
] # /known/ is limited at 900

with pytest.raises(APIError):
event_loop.run_until_complete(swhids_discovery(request, aiosession, api_url))


def test_scanner_directory_priority_has_contents(source_tree):		def test_scanner_directory_priority_has_contents(source_tree):
nodes_data = MerkleNodeInfo()		nodes_data = MerkleNodeInfo()
policy = DirectoryPriority(source_tree, nodes_data)		policy = DirectoryPriority(source_tree, nodes_data)
assert policy.has_contents(source_tree[b"/bar/barfoo"])		assert policy.has_contents(source_tree[b"/bar/barfoo"])


def get_backend_swhids_order(tmp_requests):		def get_backend_swhids_order(tmp_requests):
with open(tmp_requests, "r") as f:		with open(tmp_requests, "r") as f:
▲ Show 20 Lines • Show All 69 Lines • ▼ Show 20 Lines	):

backend_swhids_requests = get_backend_swhids_order(tmp_requests)		backend_swhids_requests = get_backend_swhids_order(tmp_requests)

for swhid in backend_swhids_requests[0:4]:		for swhid in backend_swhids_requests[0:4]:
assert CoreSWHID.from_string(swhid).object_type == ObjectType.CONTENT		assert CoreSWHID.from_string(swhid).object_type == ObjectType.CONTENT

for swhid in backend_swhids_requests[5:]:		for swhid in backend_swhids_requests[5:]:
assert CoreSWHID.from_string(swhid).object_type == ObjectType.DIRECTORY		assert CoreSWHID.from_string(swhid).object_type == ObjectType.DIRECTORY


		def test_greedy_bfs_policy(
		live_server, event_loop, aiosession, big_source_tree, tmp_requests
		):
		open(tmp_requests, "w").close()
		api_url = url_for("index", _external=True)

		nodes_data = MerkleNodeInfo()
		policy = GreedyBFS(big_source_tree, nodes_data)
		event_loop.run_until_complete(policy.run(aiosession, api_url))

		backend_swhids_requests = get_backend_swhids_order(tmp_requests)

		last_swhid = backend_swhids_requests[-1]
		assert CoreSWHID.from_string(last_swhid).object_type == ObjectType.CONTENT


		@pytest.mark.asyncio
		async def test_greedy_bfs_get_nodes_chunks(live_server, aiosession, big_source_tree):
		api_url = url_for("index", _external=True)

		nodes_data = MerkleNodeInfo()
		policy = GreedyBFS(big_source_tree, nodes_data)
		chunks = [
		n_chunk
		async for n_chunk in policy.get_nodes_chunks(
		aiosession, api_url, source_size(big_source_tree)
		)
		]
		assert len(chunks) == 2
		assert chunks[1][-1].object_type == CONTENT