Marmelator/pyinfra
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 36 Wiki Activity

moved parameter combination based test skipping into operation factory

Browse Source
This commit is contained in:
Matthias Bisping 2022-05-23 11:28:35 +02:00
parent 13888524fb
commit 54ca81d577
2 changed files with 7 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
test/fixtures/server.py vendored
View File

@ -1,4 +1,5 @@
import io import io
import logging
import socket import socket
from itertools import repeat from itertools import repeat
from multiprocessing import Process from multiprocessing import Process
@ -19,6 +20,8 @@ from pyinfra.server.utils import make_streamable_and_wrap_in_packing_logic
from pyinfra.utils.func import starlift from pyinfra.utils.func import starlift
from test.utils.image import image_to_bytes from test.utils.image import image_to_bytes
logger = logging.getLogger(__name__)
@pytest.fixture @pytest.fixture
def host(): def host():
@ -105,6 +108,8 @@ def core_operation(item_type, one_to_many, analysis_task):
} }
return d[one_to_many][item_type][analysis_task] return d[one_to_many][item_type][analysis_task]
except KeyError: except KeyError:
pytest.skip(f"No operation defined for parameter combination.")
logger.debug(f"No operation defined for [{one_to_many=}, {item_type=}, {analysis_task=}].")
return Nothing return Nothing

7
test/unit_tests/server/pipeline_test.py
View File

@ -29,11 +29,8 @@ def test_mock_pipeline():
@pytest.mark.parametrize("client_pipeline_type", ["rest", "basic"]) @pytest.mark.parametrize("client_pipeline_type", ["rest", "basic"])
def test_pipeline( def test_pipeline(core_operation, client_pipeline, input_data_items, metadata, targets, n_items):
core_operation, client_pipeline, input_data_items, metadata, targets, item_type, one_to_many, n_items assert core_operation is not Nothing
):
if core_operation is Nothing:
pytest.skip(f"No operation defined for parameter combination: {item_type=}, {one_to_many=}")
output = compose(llift(unpack), client_pipeline)(input_data_items, metadata) output = compose(llift(unpack), client_pipeline)(input_data_items, metadata)
assert n_items == 0 or len(output) > 0 assert n_items == 0 or len(output) > 0
assert output == targets assert output == targets