Skip to content
Snippets Groups Projects
Commit eb9d214d authored by Timothe Jost's avatar Timothe Jost
Browse files

adding test

parent 8b705389
No related branches found
No related tags found
No related merge requests found
.vscode*
.pdm*
pypelines/__pycache__*
tests/__pycache__*
*.egg-info*
......@@ -9,3 +10,6 @@ dist/*
notebooks/pipelines_tests/*
notebooks/*.log
build*
.coverage
coverage*
pytest_results*
This diff is collapsed.
......@@ -25,14 +25,59 @@ requires-python = ">=3.10"
license = { text = "MIT" }
dynamic = ["version"]
[project.optional-dependencies]
celery = ["celery>=5.3.5", "alyx_connector>=2.1.5"]
[project.urls]
homepage = "https://gitlab.pasteur.fr/haisslab/data-management/pypelines"
repository = "https://gitlab.pasteur.fr/haisslab/data-management/pypelines"
documentation = "https://gitlab.pasteur.fr/haisslab/data-management/pypelines"
[project.optional-dependencies]
celery = ["celery>=5.3.5", "alyx_connector>=2.1.5"]
[tool.pdm.dev-dependencies]
dev = ["pytest>=8.3.2", "pytest-cov>=5.0.0"]
[tool.pytest.ini_options]
addopts = "--pdbcls=IPython.terminal.debugger:TerminalPdb --cov-report xml --cov --junitxml=pytest_results.xml"
testpaths = ["tests"]
[tool.pdm.version]
source = "file"
path = "src/pypelines/__init__.py"
[tool.black]
preview = true
line-length = 120
[tool.flake8]
max-line-length = 120
ignore = [
"F401",
"F403",
"F841",
"E401",
"E265",
"E704",
"E266",
"E203",
"E712",
"W503",
"W605",
]
[tool.pyright]
reportGeneralTypeIssues = "information"
reportOptionalMemberAccess = "information"
reportMissingImports = "information"
reportMissingModuleSource = "information"
reportInvalidStringEscapeSequence = "information"
reportOptionalOperand = "none"
reportOptionalSubscript = "none"
reportOptionalIterable = "none"
reportAttributeAccessIssue = "information"
typeCheckingMode = "basic"
include = ["src", "tests"]
exclude = ["**/__pycache__"]
__version__ = "0.0.68"
__version__ = "0.0.70"
from . import loggs
from .pipes import *
......
......@@ -10,11 +10,7 @@ if TYPE_CHECKING:
from .graphs import PipelineGraph
class BasePipelineType(Protocol):
def __getattr__(self, name: str) -> "BasePipe": ...
class Pipeline(BasePipelineType):
class Pipeline:
pipes: Dict[str, "BasePipe"]
runner_backend_class = BaseTaskBackend
......@@ -108,6 +104,11 @@ class Pipeline(BasePipelineType):
self.resolved = True
def __getattr__(self, name: str) -> "BasePipe":
if name in self.pipes:
return self.pipes[name]
raise AttributeError(f"'Pipeline' object has no attribute '{name}'")
def get_requirement_stack(
self, instance: "BaseStep", names: bool = False, max_recursion: int = 100
) -> List["BaseStep"]:
......
import pytest
# for testing on local version, instead of installed version,
# this may not be desired as testing uninstalled may not catch issues that occur after installation is performed
# comment the next three lines to test installed version
# import sys
# from pathlib import Path
# sys.path.append(str(Path(__file__).resolve().parent / "src"))
from pypelines import examples
from pypelines.sessions import Session
from pypelines import Pipeline, stepmethod, BaseStep
from pypelines.pickle_backend import PicklePipe
@pytest.fixture
def test_class_based_pypeline():
pipeline = Pipeline("test_class_based")
@pipeline.register_pipe
class MyPipe(PicklePipe):
class Steps:
def my_step(self, session, extra=""):
return 1
my_step.requires = []
return pipeline
@pytest.fixture
def test_method_based_pypeline():
pipeline = Pipeline("test_method_based")
@pipeline.register_pipe
class MyPipe(PicklePipe):
@stepmethod(requires=[])
def my_step(self, session, extra=""):
return 1
def test_pypeline_creation(test_class_based_pypeline):
assert isinstance(test_class_based_pypeline.MyPipe.my_step, BaseStep)
assert hasattr(test_class_based_pypeline.MyPipe.my_step, "generate")
assert hasattr(test_class_based_pypeline.MyPipe.my_step, "load")
assert hasattr(test_class_based_pypeline.MyPipe.my_step, "save")
import unittest
# for testing on local version, instead of installed version,
# this may not be desired as testing uninstalled may not catch issues that occur after installation is performed
# comment the next three lines to test installed version
# import sys
# from pathlib import Path
# sys.path.append(str(Path(__file__).resolve().parent / "src"))
from pypelines import examples
from pypelines.sessions import Session
class TestVersions(unittest.TestCase):
def setUp(self):
self.pipeline = examples.example_pipeline
self.session = Session(subject="test_subject", date="2023-10-10", number=0, path="C:/test", auto_path=True)
print(self.session.alias)
def test_pipeline_generate(self):
self.assertEqual(
self.pipeline.ExamplePipe.example_step1.generate(self.session, "bonjour"),
{"argument1": "bonjour", "optionnal_argument2": 23},
)
if __name__ == "__main__":
unittest.main()
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment