fix PipeState

It was incorrect to replace the various state['something'] = True with a
single-value thing. More than one needs to be able to be present at
once, so make it a set instead.

Author: fredrikt

src/pyff/api.py

@@ -289,7 +289,7 @@ def process_handler(request: Request) -> Response:
     try:
         for p in request.registry.plumbings:
             state = PipeState(
-                entry_name=entry,
+                conditions={entry},
                 headers={'Content-Type': None},
                 accept=accept,
                 url=request.current_route_url(),

src/pyff/builtins.py

@@ -13,7 +13,7 @@
 from copy import deepcopy
 from datetime import datetime
 from distutils.util import strtobool
-from typing import Dict, Optional
+from typing import Any, Dict, List, Optional
 
 import ipaddr
 import six
@@ -25,7 +25,7 @@
 from pyff.decorators import deprecated
 from pyff.exceptions import MetadataException
 from pyff.logs import get_log
-from pyff.pipes import PipeException, PipeState, PipelineCallback, Plumbing, pipe, registry
+from pyff.pipes import PipeException, PipelineCallback, Plumbing, pipe, registry
 from pyff.samlmd import (
     annotate_entity,
     discojson_t,
@@ -383,13 +383,20 @@ def when(req: Plumbing.Request, condition: str, *values):
     The condition operates on the state: if 'foo' is present in the state (with any value), then the something branch is
     followed. If 'bar' is present in the state with the value 'bill' then the other branch is followed.
     """
-    if req.state.entry_name is None:
-        log.debug(f'Condition {repr(condition)} not present in state {req.state}')
-    if req.state.entry_name is not None and (not values or _any(values, req.state.entry_name)):
+    log.debug(f'"when" called for condition "{condition}", values {values}, state {req.state}')
+    c: Any = None
+    if condition in req.state.conditions:
+        c = True
+    if condition == 'accept':
+        c = req.state.accept
+    if c is not None and (not values or _any(values, c)):
         if not isinstance(req.args, list):
             raise ValueError('Non-list arguments to "when" not allowed')
 
-        return Plumbing(pipeline=req.args, pid=f'{req.plumbing.id}.when').iprocess(req)
+        _pid = f'{req.plumbing.id}.when'
+        log.debug(f'Creating new plumbing: {_pid}, pipeline {req.args}')
+        return Plumbing(pipeline=req.args, pid=_pid).iprocess(req)
+    log.debug(f'Continuing on plumbing {req.id}')
     return req.t
 
 

src/pyff/md.py

@@ -8,7 +8,7 @@
 import traceback
 
 from pyff.constants import config, parse_options
-from pyff.pipes import plumbing
+from pyff.pipes import PipeState, plumbing
 from pyff.repo import MDRepository
 
 
@@ -30,7 +30,7 @@ def main():
     try:
         md = MDRepository()
         for p in args:
-            plumbing(p).process(md, state={'batch': True, 'stats': {}})
+            plumbing(p).process(md, state=PipeState(conditions={'batch'}))
         sys.exit(0)
     except Exception as ex:
         logging.debug(traceback.format_exc())

src/pyff/pipes.py

@@ -7,8 +7,7 @@
 import functools
 import os
 import traceback
-from typing import Any, Callable, Dict, Iterable, List, Optional, TYPE_CHECKING, Tuple, Type
-from typing import Union
+from typing import Any, Callable, Dict, Iterable, List, Optional, Set, TYPE_CHECKING, Tuple, Type, Union
 
 import yaml
 from apscheduler.schedulers.background import BackgroundScheduler
@@ -177,7 +176,7 @@ def __call__(self, t: ElementTree, state: Optional[PipeState] = None) -> Any:
         if not isinstance(state, PipeState):
             raise ValueError(f'PipelineCallback called with invalid state ({type(state)}')
         try:
-            state.entry_name = self.entry_point
+            state.conditions.add(self.entry_point)
             log.debug("state: {}".format(state))
             return self.plumbing.process(self.req.md, store=self.store, state=state, t=t)
         except Exception as ex:
@@ -187,14 +186,13 @@ def __call__(self, t: ElementTree, state: Optional[PipeState] = None) -> Any:
 
 
 class PipeState(BaseModel):
-    batch: bool = False
-    entry_name: Optional[str] = None
+    conditions: Set[str] = Field(set())
     headers: Dict[str, Any] = Field({})
     accept: Any = None  # TODO: Re-arrange classes so that type 'MediaAccept' works
     url: str = ''
-    select: str = ''
-    match: str = ''
-    path: str = ''
+    select: Optional[str] = None
+    match: Optional[str] = None
+    path: Optional[str] = None
     stats: Dict[str, Any] = Field({})
     cache: int = 0  # cache_ttl
 

src/pyff/test/test_pipeline.py

@@ -61,7 +61,7 @@ def run_pipeline(self, pl_name, ctx=None, md=None):
         template = templates.get_template(pl_name)
         with open(pipeline, "w") as fd:
             fd.write(template.render(ctx=ctx))
-        res = plumbing(pipeline).process(md, PipeState(entry_name='batch'))
+        res = plumbing(pipeline).process(md, PipeState(conditions={'batch'}))
         os.unlink(pipeline)
         return res, md, ctx
 
@@ -70,7 +70,7 @@ def exec_pipeline(self, pstr):
         p = yaml.safe_load(six.StringIO(pstr))
         print("\n{}".format(yaml.dump(p)))
         pl = Plumbing(p, pid="test")
-        res = pl.process(md, PipeState(entry_name='batch'))
+        res = pl.process(md, PipeState(conditions={'batch'}))
         return res, md
 
     @classmethod