Merge 1763776 into 052c556

Author: fredrikt

src/pyff/api.py

@@ -1,6 +1,7 @@
 import importlib
 import threading
 from datetime import datetime, timedelta
+from enum import Enum
 from json import dumps
 from typing import Any, Dict, Generator, Iterable, List, Mapping, Optional, Tuple
 
@@ -21,7 +22,7 @@
 from pyff.constants import config
 from pyff.exceptions import ResourceException
 from pyff.logs import get_log
-from pyff.pipes import plumbing
+from pyff.pipes import PipeState, plumbing
 from pyff.repo import MDRepository
 from pyff.resource import Resource
 from pyff.samlmd import entity_display_name
@@ -153,16 +154,36 @@ def request_handler(request: Request) -> Response:
     return r
 
 
-def process_handler(request: Request) -> Response:
+class ContentNegPolicy(Enum):
+    extension = 'extension'  # current default
+    adaptive = 'adaptive'
+    header = 'header'  # future default
+
+
+def _process_content_negotiate(
+    policy: ContentNegPolicy, alias: str, path: Optional[str], pfx, request: Request
+) -> Tuple[MediaAccept, Optional[str], Optional[str]]:
     """
-    The main request handler for pyFF. Implements API call hooks and content negotiation.
+    Determine requested content type, based on policy, Accept request header and path extension.
 
-    :param request: the HTTP request object
-    :return: the data to send to the client
+    content_negotiation_policy is one of three values:
+
+        1. extension - current default, inspect the path and if it ends in
+           an extension, e.g. .xml or .json, always strip off the extension to
+           get the entityID and if no accept header or a wildcard header, then
+           use the extension to determine the return Content-Type.
+
+        2. adaptive - only if no accept header or if a wildcard, then inspect
+           the path and if it ends in an extension strip off the extension to
+           get the entityID and use the extension to determine the return
+           Content-Type.
+
+        3. header - future default, do not inspect the path for an extension and
+           use only the Accept header to determine the return Content-Type.
     """
     _ctypes = {'xml': 'application/samlmetadata+xml;application/xml;text/xml', 'json': 'application/json'}
 
-    def _d(x: Optional[str], do_split: bool = True) -> Tuple[Optional[str], Optional[str]]:
+    def _split_path(x: Optional[str], do_split: bool = True) -> Tuple[Optional[str], Optional[str]]:
         """ Split a path into a base component and an extension. """
         if x is not None:
             x = x.strip()
@@ -178,6 +199,45 @@ def _d(x: Optional[str], do_split: bool = True) -> Tuple[Optional[str], Optional
 
         return x, None
 
+    # TODO - sometimes the client sends > 1 accept header value with ','.
+    accept = str(request.accept).split(',')[0]
+    valid_accept = accept and not ('application/*' in accept or 'text/*' in accept or '*/*' in accept)
+
+    path_no_extension, extension = _split_path(path, True)
+    accept_from_extension = accept
+    if extension:
+        accept_from_extension = _ctypes.get(extension, accept)
+
+    if policy == ContentNegPolicy.extension:
+        path = path_no_extension
+        if not valid_accept:
+            accept = accept_from_extension
+    elif policy == ContentNegPolicy.adaptive:
+        if not valid_accept:
+            path = path_no_extension
+            accept = accept_from_extension
+
+    if not accept:
+        log.warning('Could not determine accepted response type')
+        raise exc.exception_response(400)
+
+    q: Optional[str]
+    if pfx and path:
+        q = f'{{{pfx}}}{path}'
+        path = f'/{alias}/{path}'
+    else:
+        q = path
+
+    return MediaAccept(accept), path, q
+
+
+def process_handler(request: Request) -> Response:
+    """
+    The main request handler for pyFF. Implements API call hooks and content negotiation.
+
+    :param request: the HTTP request object
+    :return: the data to send to the client
+    """
     log.debug(f'Processing request: {request}')
 
     if request.matchdict is None:
@@ -215,83 +275,45 @@ def _d(x: Optional[str], do_split: bool = True) -> Tuple[Optional[str], Optional
         if pfx is None:
             raise exc.exception_response(404)
 
-    # content_negotiation_policy is one of three values:
-    # 1. extension - current default, inspect the path and if it ends in
-    #    an extension, e.g. .xml or .json, always strip off the extension to
-    #    get the entityID and if no accept header or a wildcard header, then
-    #    use the extension to determine the return Content-Type.
-    #
-    # 2. adaptive - only if no accept header or if a wildcard, then inspect
-    #    the path and if it ends in an extension strip off the extension to
-    #    get the entityID and use the extension to determine the return
-    #    Content-Type.
-    #
-    # 3. header - future default, do not inspect the path for an extension and
-    #    use only the Accept header to determine the return Content-Type.
-    policy = config.content_negotiation_policy
-
-    # TODO - sometimes the client sends > 1 accept header value with ','.
-    accept = str(request.accept).split(',')[0]
-    valid_accept = accept and not ('application/*' in accept or 'text/*' in accept or '*/*' in accept)
-
-    new_path: Optional[str] = path
-    path_no_extension, extension = _d(new_path, True)
-    accept_from_extension = accept
-    if extension:
-        accept_from_extension = _ctypes.get(extension, accept)
-
-    if policy == 'extension':
-        new_path = path_no_extension
-        if not valid_accept:
-            accept = accept_from_extension
-    elif policy == 'adaptive':
-        if not valid_accept:
-            new_path = path_no_extension
-            accept = accept_from_extension
-
-    if not accept:
-        log.warning('Could not determine accepted response type')
-        raise exc.exception_response(400)
+    try:
+        policy = ContentNegPolicy(config.content_negotiation_policy)
+    except ValueError:
+        log.debug(
+            f'Invalid value for config.content_negotiation_policy: {config.content_negotiation_policy}, '
+            f'defaulting to "extension"'
+        )
+        policy = ContentNegPolicy.extension
 
-    q: Optional[str]
-    if pfx and new_path:
-        q = f'{{{pfx}}}{new_path}'
-        new_path = f'/{alias}/{new_path}'
-    else:
-        q = new_path
+    accept, new_path, q = _process_content_negotiate(policy, alias, path, pfx, request)
 
     try:
-        accepter = MediaAccept(accept)
         for p in request.registry.plumbings:
-            state = {
-                entry: True,
-                'headers': {'Content-Type': None},
-                'accept': accepter,
-                'url': request.current_route_url(),
-                'select': q,
-                'match': match.lower() if match else match,
-                'path': new_path,
-                'stats': {},
-            }
+            state = PipeState(
+                conditions={entry},
+                headers={'Content-Type': None},
+                accept=accept,
+                url=request.current_route_url(),
+                select=q,
+                match=match.lower() if match else match,
+                path=new_path,
+                stats={},
+            )
 
             r = p.process(request.registry.md, state=state, raise_exceptions=True, scheduler=request.registry.scheduler)
-            log.debug(f'Plumbing process result: {r}')
             if r is None:
                 r = []
 
             response = Response()
-            _headers = state.get('headers', {})
-            response.headers.update(_headers)
-            ctype = _headers.get('Content-Type', None)
+            response.headers.update(state.headers)
+            ctype = state.headers.get('Content-Type', None)
             if not ctype:
-                r, t = _fmt(r, accepter)
+                r, t = _fmt(r, accept)
                 ctype = t
 
             response.text = b2u(r)
             response.size = len(r)
             response.content_type = ctype
-            cache_ttl = int(state.get('cache', 0))
-            response.expires = datetime.now() + timedelta(seconds=cache_ttl)
+            response.expires = datetime.now() + timedelta(seconds=state.cache)
             return response
     except ResourceException as ex:
         import traceback

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, Iterable, List, Optional
 
 import ipaddr
 import six
@@ -274,10 +274,10 @@ def fork(req: Plumbing.Request, *opts):
 
 @deprecated(reason="any pipeline has been replace by other behaviour")
 @pipe(name='any')
-def _any(lst, d):
+def _any(lst: Iterable[Any], d: Any) -> Any:
     for x in lst:
         if x in d:
-            if type(d) == dict:
+            if isinstance(d, dict):
                 return d[x]
             else:
                 return True
@@ -383,14 +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.
     """
-    c = req.state.get(condition, None)
-    if c is None:
-        log.debug(f'Condition {repr(condition)} not present in state {req.state}')
+    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="%s.when" % req.plumbing.id).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
 
 
@@ -506,7 +512,7 @@ def publish(req: Plumbing.Request, *opts):
         try:
             validate_document(req.t)
         except DocumentInvalid as ex:
-            log.error(ex.error_log)
+            log.error(f'Got an error from validate_document {ex.error_log}')
             raise PipeException("XML schema validation failed")
 
     def _nop(x):
@@ -679,18 +685,35 @@ def load(req: Plumbing.Request, *opts):
     req.md.rm.reload(fail_on_error=bool(_opts['fail_on_error']))
 
 
-def _select_args(req):
-    args = req.args
-    if args is None and 'select' in req.state:
-        args = [req.state.get('select')]
-    if args is None:
+def _select_args(req: Plumbing.Request) -> List[str]:
+    log.debug(f'Select args: {req.args}, state: {req.state}')
+    args: List[str] = []
+
+    if req.args:
+        for this in req.args:
+            if not isinstance(this, str):
+                raise ValueError(f'Selection not possible with arg that is not a string: {this}')
+            args += [this]
+
+    if not args and req.state.select:
+        args = [req.state.select]
+        log.debug(f'Using req.state.select: {args}')
+    if not args:
         args = req.store.collections()
-    if args is None or not args:
+        log.debug(f'Using req.store.collections: {args}')
+    if not args:
         args = req.store.lookup('entities')
-    if args is None or not args:
+        if len(args) < 5:
+            log.debug(f'Using req.store.entities: {args}')
+        else:
+            log.debug(f'Using req.store.entities: {args[:4]} (truncated)')
+    if not args:
         args = []
 
-    log.info("selecting using args: %s" % args)
+    if len(args) < 5:
+        log.info(f'selecting using args: {args}')
+    else:
+        log.info(f'selecting using args: {args[:4]} (truncated)')
 
     return args
 
@@ -768,9 +791,9 @@ def select(req: Plumbing.Request, *opts):
 
     entities = resolve_entities(args, lookup_fn=req.md.store.select)
 
-    if req.state.get('match', None):  # TODO - allow this to be passed in via normal arguments
+    if req.state.match:  # TODO - allow this to be passed in via normal arguments
 
-        match = req.state['match']
+        match = req.state.match
 
         if isinstance(match, six.string_types):
             query = [match.lower()]
@@ -1435,11 +1458,11 @@ def emit(req: Plumbing.Request, ctype="application/xml", *opts):
         if not isinstance(d, six.binary_type):
             d = d.encode("utf-8")
         m.update(d)
-        req.state['headers']['ETag'] = m.hexdigest()
+        req.state.headers['ETag'] = m.hexdigest()
     else:
         raise PipeException("Empty")
 
-    req.state['headers']['Content-Type'] = ctype
+    req.state.headers['Content-Type'] = ctype
     if six.PY2:
         d = six.u(d)
     return d
@@ -1517,7 +1540,7 @@ def finalize(req: Plumbing.Request, *opts):
         if name is None or 0 == len(name):
             name = req.args.get('Name', None)
         if name is None or 0 == len(name):
-            name = req.state.get('url', None)
+            name = req.state.url
             if name and 'baseURL' in req.args:
 
                 try:
@@ -1569,7 +1592,7 @@ def finalize(req: Plumbing.Request, *opts):
         # TODO: offset can be None here, if validUntil is not a valid duration or ISO date
         #       What is the right action to take then?
         if offset:
-            req.state['cache'] = int(total_seconds(offset) / 50)
+            req.state.cache = int(total_seconds(offset) / 50)
 
     cache_duration = req.args.get('cacheDuration', e.get('cacheDuration', None))
     if cache_duration is not None and len(cache_duration) > 0:
@@ -1578,7 +1601,7 @@ def finalize(req: Plumbing.Request, *opts):
             raise PipeException("Unable to parse %s as xs:duration" % cache_duration)
 
         e.set('cacheDuration', cache_duration)
-        req.state['cache'] = int(total_seconds(offset))
+        req.state.cache = int(total_seconds(offset))
 
     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())