Commit 874089b9 authored by Pierre Dittgen's avatar Pierre Dittgen

Fix tox issues

parent 2b105274
Pipeline #114641 passed with stage
in 3 minutes and 25 seconds
......@@ -41,7 +41,8 @@ ResourceId = str
class AbortIteration(Exception):
"""Raised by fetcher to abort iteration"""
"""Raised by fetcher to abort iteration."""
pass
......
......@@ -18,13 +18,13 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Tests about event loader"""
"""Tests about event loader."""
import argparse
import shutil
import tempfile
from pathlib import Path
from typing import List
from typing import Any, Dict, List
import pytest
......@@ -41,21 +41,26 @@ class FakeArgs(argparse.Namespace):
"""Mimic command-line args."""
def __init__(self, **kwargs):
"""Build new instance with given properties."""
self.attrs = kwargs
def __getattr__(self, attr):
"""Emulate attribute access."""
return self.attrs[attr] if attr in self.attrs else None
def do_nothing_with_given_resource(resource: Resource):
"""null function."""
"""Null function."""
pass
EMPTY_DICT: Dict[str, Any] = {}
async def process_resources_core(
resources: List[Resource],
assert_func=None,
additional_args={},
additional_args=EMPTY_DICT,
process_resource=do_nothing_with_given_resource,
on_exception=None,
):
......@@ -93,8 +98,8 @@ async def process_resources_core(
async def test_skip_no_resources():
"""Process all resources."""
RESOURCE_IDS = list("ABCDEF")
resources = [Resource(id=letter) for letter in RESOURCE_IDS]
resource_ids = list("ABCDEF")
resources = [Resource(id=letter) for letter in resource_ids]
def assert_func(resource_event_by_id=None):
assert resource_event_by_id is not None
......@@ -111,11 +116,10 @@ async def test_skip_no_resources():
async def test_process_one_resource():
"""Process ony one resource."""
RESOURCE_IDS = list("ABCDEF")
resources = [Resource(id=letter) for letter in RESOURCE_IDS]
SELECTED_RESOURCE_ID = "C"
resources = [Resource(id=letter) for letter in list("ABCDEF")]
selected_resource_id = "C"
additional_args = {"only": SELECTED_RESOURCE_ID}
additional_args = {"only": selected_resource_id}
def assert_func(resource_event_by_id=None):
assert resource_event_by_id is not None
......@@ -123,10 +127,10 @@ async def test_process_one_resource():
assert all(
evt.status == status.ResourceStatus.SKIPPED
for resource_id, evt in resource_event_by_id.items()
if resource_id != SELECTED_RESOURCE_ID
if resource_id != selected_resource_id
)
assert (
resource_event_by_id[SELECTED_RESOURCE_ID].status
resource_event_by_id[selected_resource_id].status
== status.ResourceStatus.SUCCESS
)
......@@ -137,11 +141,11 @@ async def test_process_one_resource():
async def test_exclude_some_resources():
"""Process all resources except some ones."""
RESOURCE_IDS = list("ABCDEF")
resources = [Resource(id=letter) for letter in RESOURCE_IDS]
EXCLUDED_RESOURCE_IDS = ["A", "E"]
resource_ids = list("ABCDEF")
resources = [Resource(id=letter) for letter in resource_ids]
excluded_resource_ids = ["A", "E"]
additional_args = {"exclude": EXCLUDED_RESOURCE_IDS}
additional_args = {"exclude": excluded_resource_ids}
def assert_func(resource_event_by_id=None):
assert resource_event_by_id is not None
......@@ -149,12 +153,12 @@ async def test_exclude_some_resources():
assert all(
evt.status == status.ResourceStatus.SKIPPED
for resource_id, evt in resource_event_by_id.items()
if resource_id in EXCLUDED_RESOURCE_IDS
if resource_id in excluded_resource_ids
)
assert all(
evt.status == status.ResourceStatus.SUCCESS
for resource_id, evt in resource_event_by_id.items()
if resource_id not in EXCLUDED_RESOURCE_IDS
if resource_id not in excluded_resource_ids
)
await process_resources_core(resources, assert_func, additional_args)
......@@ -164,8 +168,8 @@ async def test_exclude_some_resources():
async def test_abort_on_resource_list():
"""Abort iteration on the middle of the way."""
RESOURCE_IDS = list("ABCDEF")
resources = [Resource(id=letter) for letter in RESOURCE_IDS]
resource_ids = list("ABCDEF")
resources = [Resource(id=letter) for letter in resource_ids]
before_abort_resource_ids = {"A", "B"}
abort_resource_id = "C"
......@@ -174,7 +178,7 @@ async def test_abort_on_resource_list():
aborted_error_msg = "Resource processing aborted"
def process_resource(resource: Resource):
"""Do nothing but fails on one specific resource"""
"""Do nothing but fail on one specific resource."""
if resource.id == abort_resource_id:
raise AbortIteration()
......@@ -188,12 +192,12 @@ async def test_abort_on_resource_list():
)
event = resource_event_by_id[abort_resource_id]
assert (
event.status == status.ResourceStatus.FAILURE and
event.message.startswith(aborted_error_msg)
event.status == status.ResourceStatus.FAILURE
and event.message.startswith(aborted_error_msg)
)
assert all(
evt.status == status.ResourceStatus.SKIPPED and
evt.message.startswith(aborted_error_msg)
evt.status == status.ResourceStatus.SKIPPED
and evt.message.startswith(aborted_error_msg)
for resource_id, evt in resource_event_by_id.items()
if resource_id in after_abort_resource_ids
)
......@@ -202,8 +206,11 @@ async def test_abort_on_resource_list():
assert isinstance(exc, AbortIteration)
await process_resources_core(
resources, assert_func, {}, process_resource=process_resource,
on_exception=assert_abort_iteration
resources,
assert_func,
{},
process_resource=process_resource,
on_exception=assert_abort_iteration,
)
......@@ -211,15 +218,14 @@ async def test_abort_on_resource_list():
async def test_abort_on_only_one_resource():
"""Abort iteration on the only chosen resource."""
RESOURCE_IDS = list("ABCDEF")
resources = [Resource(id=letter) for letter in RESOURCE_IDS]
resources = [Resource(id=letter) for letter in list("ABCDEF")]
other_resource_ids = {"A", "B", "D", "E", "F"}
abort_resource_id = "C"
aborted_error_msg = "Resource processing aborted"
def process_resource(resource: Resource):
"""Do nothing but fails on one specific resource"""
"""Do nothing but fail on one specific resource."""
if resource.id == abort_resource_id:
raise AbortIteration()
......@@ -227,26 +233,26 @@ async def test_abort_on_only_one_resource():
assert resource_event_by_id is not None
assert len(resource_event_by_id) == len(resources)
assert all(
evt.status == status.ResourceStatus.SKIPPED and
not evt.message.startswith(aborted_error_msg)
evt.status == status.ResourceStatus.SKIPPED
and not evt.message.startswith(aborted_error_msg)
for resource_id, evt in resource_event_by_id.items()
if resource_id in other_resource_ids
)
aborted_evt = resource_event_by_id[abort_resource_id]
assert (
aborted_evt.status == status.ResourceStatus.FAILURE and
aborted_evt.message.startswith(aborted_error_msg)
aborted_evt.status == status.ResourceStatus.FAILURE
and aborted_evt.message.startswith(aborted_error_msg)
)
additional_args = {
'only': abort_resource_id
}
additional_args = {"only": abort_resource_id}
def assert_abort_iteration(exc: Exception):
assert isinstance(exc, AbortIteration)
await process_resources_core(
resources, assert_func, additional_args=additional_args,
resources,
assert_func,
additional_args=additional_args,
process_resource=process_resource,
on_exception=assert_abort_iteration,
)
......@@ -256,8 +262,8 @@ async def test_abort_on_only_one_resource():
async def test_abort_check_message():
"""Abort iteration and transmit message."""
RESOURCE_ID = 'R'
resources = [Resource(id=RESOURCE_ID)]
resource_id = "R"
resources = [Resource(id=resource_id)]
fetcher_error_msg = "Something is rotten in the state of Denmark"
aborted_error_msg = "Resource processing aborted"
......@@ -269,7 +275,7 @@ async def test_abort_check_message():
def assert_func(resource_event_by_id=None):
assert resource_event_by_id is not None
assert len(resource_event_by_id) == len(resources)
evt = resource_event_by_id[RESOURCE_ID]
evt = resource_event_by_id[resource_id]
assert evt.status == ResourceStatus.FAILURE
assert evt.message.startswith(aborted_error_msg)
assert evt.message.endswith(fetcher_error_msg)
......@@ -278,6 +284,9 @@ async def test_abort_check_message():
assert isinstance(exc, AbortIteration)
await process_resources_core(
resources, assert_func, {}, process_resource=process_resource,
on_exception=assert_abort_iteration
resources,
assert_func,
{},
process_resource=process_resource,
on_exception=assert_abort_iteration,
)
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment