Commit f8ad2ad5 authored by Pierre Dittgen's avatar Pierre Dittgen

Fix flake8 issues

parent de1d32ef
Pipeline #114643 passed with stage
in 3 minutes and 37 seconds
......@@ -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
......@@ -36,15 +36,22 @@ class FakeArgs(argparse.Namespace):
"""Mimic command-line args."""
def __init__(self, **kwargs):
"""Build a new instance."""
self.attrs = kwargs
def __getattr__(self, attr):
"""Mimic attribute access."""
return self.attrs[attr] if attr in self.attrs else None
EMPTY_DICT: Dict[str, Any] = {}
async def process_resources_core(
resources: List[Resource], assert_func=None, additional_args={}
resources: List[Resource], assert_func=None, additional_args=EMPTY_DICT
):
"""Factorize code for test cases."""
def process_resource(resource: Resource):
pass
......@@ -77,8 +84,7 @@ 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]
resources = [Resource(id=letter) for letter in list("ABCDEF")]
def assert_func(resource_event_by_id=None):
assert resource_event_by_id is not None
......@@ -95,11 +101,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
......@@ -107,10 +112,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
)
......@@ -121,11 +126,10 @@ 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"]
resources = [Resource(id=letter) for letter in list("ABCDEF")]
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
......@@ -133,12 +137,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 (
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)
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