Changeset View
Changeset View
Standalone View
Standalone View
swh/scanner/tests/test_scanner.py
# Copyright (C) 2020 The Software Heritage developers | # Copyright (C) 2020 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 pytest | import pytest | ||||
import json | import json | ||||
from pathlib import PosixPath | from pathlib import PosixPath | ||||
from .data import correct_api_response | from .data import correct_api_response | ||||
from swh.scanner.scanner import pids_discovery, get_subpaths, run | from swh.scanner.scanner import pids_discovery, get_subpaths, run | ||||
from swh.scanner.model import Tree | from swh.scanner.model import Tree | ||||
from swh.scanner.cli import extract_regex_objs | |||||
from swh.scanner.exceptions import APIError | from swh.scanner.exceptions import APIError | ||||
aio_url = "http://example.org/api/known/" | aio_url = "http://example.org/api/known/" | ||||
def test_scanner_correct_api_request(mock_aioresponse, event_loop, aiosession): | def test_scanner_correct_api_request(mock_aioresponse, event_loop, aiosession): | ||||
mock_aioresponse.post( | mock_aioresponse.post( | ||||
aio_url, | aio_url, | ||||
Show All 24 Lines | def test_scanner_raise_apierror_input_size_limit(event_loop, aiosession, live_server): | ||||
request = [ | request = [ | ||||
"swh:1:cnt:7c4c57ba9ff496ad179b8f65b1d286edbda34c9a" for i in range(901) | "swh:1:cnt:7c4c57ba9ff496ad179b8f65b1d286edbda34c9a" for i in range(901) | ||||
] # /known/ is limited at 900 | ] # /known/ is limited at 900 | ||||
with pytest.raises(APIError): | with pytest.raises(APIError): | ||||
event_loop.run_until_complete(pids_discovery(request, aiosession, api_url)) | event_loop.run_until_complete(pids_discovery(request, aiosession, api_url)) | ||||
def test_scanner_get_subpaths(temp_folder, tmp_path): | def test_scanner_get_subpaths(temp_folder): | ||||
paths = temp_folder["paths"].keys() | root = temp_folder["root"] | ||||
pids = temp_folder["paths"].values() | |||||
actual_result = [] | |||||
for subpath, pid in get_subpaths(tmp_path): | for subpath, pid in get_subpaths(root, tuple()): | ||||
assert subpath in paths | # also check if it's a symlink since pytest tmp_dir fixture create | ||||
assert pid in pids | # also a symlink to each directory inside the tmp_dir path | ||||
if subpath.is_dir() and not subpath.is_symlink(): | |||||
actual_result.append((subpath, pid)) | |||||
assert len(actual_result) == 2 | |||||
vlorentz: this test doesn't fail if `get_subpaths` returns nothing, which is what happens here.
You… | |||||
@pytest.mark.options(debug=False) | @pytest.mark.options(debug=False) | ||||
def test_app(app): | def test_app(app): | ||||
assert not app.debug | assert not app.debug | ||||
def test_scanner_result(live_server, event_loop, test_folder): | def test_scanner_result(live_server, event_loop, test_folder): | ||||
api_url = live_server.url() + "/" | api_url = live_server.url() + "/" | ||||
result_path = test_folder.joinpath(PosixPath("sample-folder-result.json")) | result_path = test_folder.joinpath(PosixPath("sample-folder-result.json")) | ||||
with open(result_path, "r") as json_file: | with open(result_path, "r") as json_file: | ||||
expected_result = json.loads(json_file.read()) | expected_result = json.loads(json_file.read()) | ||||
sample_folder = test_folder.joinpath(PosixPath("sample-folder")) | sample_folder = test_folder.joinpath(PosixPath("sample-folder")) | ||||
source_tree = Tree(sample_folder) | source_tree = Tree(sample_folder) | ||||
event_loop.run_until_complete(run(sample_folder, api_url, source_tree)) | event_loop.run_until_complete(run(sample_folder, api_url, source_tree, tuple())) | ||||
actual_result = source_tree.getTree() | |||||
assert actual_result == expected_result | |||||
def test_scanner_result_with_exclude_patterns(live_server, event_loop, test_folder): | |||||
api_url = live_server.url() + "/" | |||||
result_path = test_folder.joinpath( | |||||
PosixPath("sample-folder-result-no-toexclude.json") | |||||
) | |||||
with open(result_path, "r") as json_file: | |||||
expected_result = json.loads(json_file.read()) | |||||
sample_folder = test_folder.joinpath(PosixPath("sample-folder")) | |||||
patterns = (str(sample_folder) + "/toexclude",) | |||||
exclude_pattern = { | |||||
reg_obj for reg_obj in extract_regex_objs(sample_folder, patterns) | |||||
} | |||||
source_tree = Tree(sample_folder) | |||||
event_loop.run_until_complete( | |||||
run(sample_folder, api_url, source_tree, exclude_pattern) | |||||
) | |||||
actual_result = source_tree.getTree() | actual_result = source_tree.getTree() | ||||
assert actual_result == expected_result | assert actual_result == expected_result |
this test doesn't fail if get_subpaths returns nothing, which is what happens here.
You should assign the return value of get_subpaths, check the length, then iterate on it.