Added tests.

Signed-off-by: Pavel Kirilin <[email protected]>

Author: s3rius

taskiq_pipelines/abc.py

@@ -9,16 +9,16 @@
 class AbstractStep(ABC):
     """Abstract pipeline step."""
 
-    step_name: str
-    known_steps: "Dict[str, Type[AbstractStep]]" = {}
+    _step_name: str
+    _known_steps: "Dict[str, Type[AbstractStep]]" = {}
 
     def __init_subclass__(cls, step_name: str, **kwargs: Any) -> None:
         super().__init_subclass__(**kwargs)
         # Sets step name to the step.
-        cls.step_name = step_name
+        cls._step_name = step_name
         # Registers new subclass in the dict of
         # known steps.
-        cls.known_steps[step_name] = cls
+        cls._known_steps[step_name] = cls
 
     @abstractmethod
     def dumps(self) -> str:

taskiq_pipelines/pipeliner.py

@@ -111,7 +111,7 @@ def call_next(
         """
         self.steps.append(
             DumpedStep(
-                step_type=SequentialStep.step_name,
+                step_type=SequentialStep._step_name,
                 step_data=SequentialStep.from_task(
                     task=task,
                     param_name=param_name,
@@ -167,7 +167,7 @@ def call_after(
         """
         self.steps.append(
             DumpedStep(
-                step_type=SequentialStep.step_name,
+                step_type=SequentialStep._step_name,
                 step_data=SequentialStep.from_task(
                     task=task,
                     param_name=EMPTY_PARAM_NAME,
@@ -236,7 +236,7 @@ def map(
         """
         self.steps.append(
             DumpedStep(
-                step_type=MapperStep.step_name,
+                step_type=MapperStep._step_name,
                 step_data=MapperStep.from_task(
                     task=task,
                     param_name=param_name,
@@ -308,7 +308,7 @@ def filter(
         """
         self.steps.append(
             DumpedStep(
-                step_type=FilterStep.step_name,
+                step_type=FilterStep._step_name,
                 step_data=FilterStep.from_task(
                     task=task,
                     param_name=param_name,

taskiq_pipelines/steps/__init__.py

@@ -10,7 +10,7 @@
 
 
 def parse_step(step_type: str, step_data: str) -> AbstractStep:
-    step_cls = AbstractStep.known_steps.get(step_type)
+    step_cls = AbstractStep._known_steps.get(step_type)
     if step_cls is None:
         logger.warning(f"Unknown step type: {step_type}")
         raise ValueError("Unknown step type.")

taskiq_pipelines/steps/sequential.py

@@ -24,22 +24,6 @@ class SequentialStep(pydantic.BaseModel, AbstractStep, step_name="sequential"):
     param_name: Union[Optional[int], str]
     additional_kwargs: Dict[str, Any]
 
-    @pydantic.validator("param_name")
-    def validate_param_name(
-        self,
-        value: Union[Optional[str], int],
-    ) -> Union[Optional[str], int]:
-        """
-        Validate param_name.
-
-        :param value: value to validate.
-        :raises ValueError: if value is not str, None or -1 (EMPTY_PARAM_NAME).
-        :return: param value.
-        """
-        if isinstance(value, int) and value != EMPTY_PARAM_NAME:
-            raise ValueError("must be str, None or -1 (EMPTY_PARAM_NAME)")
-        return value
-
     def dumps(self) -> str:
         """
         Dumps step as string.

taskiq_pipelines/tests/test_stub.py

@@ -0,0 +1,12 @@
+import pytest
+
+
+@pytest.fixture(scope="session")
+def anyio_backend() -> str:
+    """
+    Anyio backend.
+
+    Backend for anyio pytest plugin.
+    :return: backend name.
+    """
+    return "asyncio"

tests/conftest.py

@@ -0,0 +1,46 @@
+from typing import List
+
+import pytest
+from taskiq import InMemoryBroker
+
+from taskiq_pipelines import Pipeline, PipelineMiddleware
+
+
+@pytest.mark.anyio
+async def test_success() -> None:
+    """Test stub for CI."""
+
+    broker = InMemoryBroker().with_middlewares(PipelineMiddleware())
+
+    @broker.task
+    def add(i: int) -> int:
+        return i + 1
+
+    @broker.task
+    def double(i: int) -> int:
+        return i * 2
+
+    pipe = Pipeline(broker, add).call_next(double)
+    sent = await pipe.kiq(1)
+    res = await sent.wait_result()
+    assert res.return_value == 4
+
+
+@pytest.mark.anyio
+async def test_mapping_success() -> None:
+    """Test stub for CI."""
+
+    broker = InMemoryBroker().with_middlewares(PipelineMiddleware())
+
+    @broker.task
+    def ranger(i: int) -> List[int]:
+        return list(range(i))
+
+    @broker.task
+    def double(i: int) -> int:
+        return i * 2
+
+    pipe = Pipeline(broker, ranger).map(double)
+    sent = await pipe.kiq(4)
+    res = await sent.wait_result()
+    assert res.return_value == list(map(double, ranger(4)))