Differential D6114 Diff 22144 swh/scanner/tests/test_scanner.py

Changeset View

Standalone View

swh/scanner/tests/test_scanner.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

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

from swh.scanner.data import MerkleNodeInfo		from swh.scanner.data import MerkleNodeInfo, init_merkle_node_info
from swh.scanner.policy import DirectoryPriority, FilePriority, LazyBFS, QueryAll		from swh.scanner.policy import DirectoryPriority, FilePriority, LazyBFS, QueryAll
from swh.scanner.scanner import get_policy_obj, run		from swh.scanner.scanner import get_policy_obj, run

from .data import unknown_swhids		from .data import unknown_swhids


@pytest.mark.options(debug=False)		@pytest.mark.options(debug=False)
def test_app(app):		def test_app(app):
Show All 10 Lines	def test_get_policy_obj_bfs(big_source_tree, nodes_data):
assert isinstance(get_policy_obj(big_source_tree, nodes_data, "auto"), LazyBFS)		assert isinstance(get_policy_obj(big_source_tree, nodes_data, "auto"), LazyBFS)


def test_scanner_result_bfs(live_server, event_loop, source_tree):		def test_scanner_result_bfs(live_server, event_loop, source_tree):
api_url = url_for("index", _external=True)		api_url = url_for("index", _external=True)
config = {"web-api": {"url": api_url, "auth-token": None}}		config = {"web-api": {"url": api_url, "auth-token": None}}

nodes_data = MerkleNodeInfo()		nodes_data = MerkleNodeInfo()
		init_merkle_node_info(source_tree, nodes_data, {"known"})
policy = LazyBFS(source_tree, nodes_data)		policy = LazyBFS(source_tree, nodes_data)
event_loop.run_until_complete(run(config, policy))		event_loop.run_until_complete(
		run(config, policy, source_tree, nodes_data, {"known"})
		)
for node in source_tree.iter_tree():		for node in source_tree.iter_tree():
if str(node.swhid()) in unknown_swhids:		if str(node.swhid()) in unknown_swhids:
assert nodes_data[node.swhid()]["known"] is False		assert nodes_data[node.swhid()]["known"] is False
else:		else:
assert nodes_data[node.swhid()]["known"] is True		assert nodes_data[node.swhid()]["known"] is True


def test_scanner_result_file_priority(live_server, event_loop, source_tree):		def test_scanner_result_file_priority(live_server, event_loop, source_tree):
api_url = url_for("index", _external=True)		api_url = url_for("index", _external=True)
config = {"web-api": {"url": api_url, "auth-token": None}}		config = {"web-api": {"url": api_url, "auth-token": None}}

nodes_data = MerkleNodeInfo()		nodes_data = MerkleNodeInfo()
		init_merkle_node_info(source_tree, nodes_data, {"known"})
policy = FilePriority(source_tree, nodes_data)		policy = FilePriority(source_tree, nodes_data)
event_loop.run_until_complete(run(config, policy))		event_loop.run_until_complete(
		run(config, policy, source_tree, nodes_data, {"known"})
		)
for node in source_tree.iter_tree():		for node in source_tree.iter_tree():
if str(node.swhid()) in unknown_swhids:		if str(node.swhid()) in unknown_swhids:
assert nodes_data[node.swhid()]["known"] is False		assert nodes_data[node.swhid()]["known"] is False
else:		else:
assert nodes_data[node.swhid()]["known"] is True		assert nodes_data[node.swhid()]["known"] is True


def test_scanner_result_directory_priority(live_server, event_loop, source_tree):		def test_scanner_result_directory_priority(live_server, event_loop, source_tree):
api_url = url_for("index", _external=True)		api_url = url_for("index", _external=True)
config = {"web-api": {"url": api_url, "auth-token": None}}		config = {"web-api": {"url": api_url, "auth-token": None}}

nodes_data = MerkleNodeInfo()		nodes_data = MerkleNodeInfo()
		init_merkle_node_info(source_tree, nodes_data, {"known"})
policy = DirectoryPriority(source_tree, nodes_data)		policy = DirectoryPriority(source_tree, nodes_data)
event_loop.run_until_complete(run(config, policy))		event_loop.run_until_complete(
		run(config, policy, source_tree, nodes_data, {"known"})
		)
for node in source_tree.iter_tree():		for node in source_tree.iter_tree():
if str(node.swhid()) in unknown_swhids:		if str(node.swhid()) in unknown_swhids:
assert nodes_data[node.swhid()]["known"] is False		assert nodes_data[node.swhid()]["known"] is False
else:		else:
assert nodes_data[node.swhid()]["known"] is True		assert nodes_data[node.swhid()]["known"] is True


def test_scanner_result_query_all(live_server, event_loop, source_tree):		def test_scanner_result_query_all(live_server, event_loop, source_tree):
api_url = url_for("index", _external=True)		api_url = url_for("index", _external=True)
config = {"web-api": {"url": api_url, "auth-token": None}}		config = {"web-api": {"url": api_url, "auth-token": None}}

nodes_data = MerkleNodeInfo()		nodes_data = MerkleNodeInfo()
		init_merkle_node_info(source_tree, nodes_data, {"known"})
policy = QueryAll(source_tree, nodes_data)		policy = QueryAll(source_tree, nodes_data)
event_loop.run_until_complete(run(config, policy))		event_loop.run_until_complete(
		run(config, policy, source_tree, nodes_data, {"known"})
		)
for node in source_tree.iter_tree():		for node in source_tree.iter_tree():
if str(node.swhid()) in unknown_swhids:		if str(node.swhid()) in unknown_swhids:
assert nodes_data[node.swhid()]["known"] is False		assert nodes_data[node.swhid()]["known"] is False
else:		else:
assert nodes_data[node.swhid()]["known"] is True		assert nodes_data[node.swhid()]["known"] is True