Changeset View
Changeset View
Standalone 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 |