diff --git a/src/pyff/api.py b/src/pyff/api.py index 4eb751cd..2f61821b 100644 --- a/src/pyff/api.py +++ b/src/pyff/api.py @@ -467,7 +467,7 @@ def _response() -> Generator[bytes, bytes, None]: for e in entities: if in_loop: yield b',' - yield dumps(e) + yield dumps(e).encode('utf-8') in_loop = True yield b']' diff --git a/src/pyff/samlmd.py b/src/pyff/samlmd.py index 1cc4b8a8..4de92b4e 100644 --- a/src/pyff/samlmd.py +++ b/src/pyff/samlmd.py @@ -897,7 +897,7 @@ def entity_extended_display(entity, langs=None): if info == entity.get('entityID'): info = '' - return display.strip(), info.strip() + return (display or '').strip(), (info or '').strip() def entity_display_name(entity: Element, langs=None) -> str: diff --git a/src/pyff/test/test_md_api.py b/src/pyff/test/test_md_api.py index 75c26551..72b26772 100644 --- a/src/pyff/test/test_md_api.py +++ b/src/pyff/test/test_md_api.py @@ -1,17 +1,20 @@ +import json import os import shutil import tempfile import unittest from datetime import datetime, timezone +from types import SimpleNamespace from urllib.parse import quote as urlescape import pytest import requests from lxml import etree from mako.lookup import TemplateLookup +from pyramid.testing import DummyRequest from wsgi_intercept.interceptor import RequestsInterceptor, UrllibInterceptor -from pyff.api import mkapp +from pyff.api import mkapp, search_handler from pyff.constants import config from pyff.samlmd import iter_entities from pyff.test import SignerTestCase @@ -149,6 +152,51 @@ def test_load_and_query(self): ) +class SearchHandlerTest(unittest.TestCase): + """ + The search endpoint streams its result, and WSGI servers such as gunicorn + only accept bytes from the response iterator. + """ + + class _Store: + def __init__(self, results): + self.results = results + + def search(self, query=None, entity_filter=None, **kwargs): + return iter(self.results) + + def _search(self, results, **params): + request = DummyRequest(params=params) + request.registry = SimpleNamespace(md=SimpleNamespace(store=self._Store(results))) + response = search_handler(request) + return response, list(response.app_iter) + + def test_chunks_are_bytes(self): + results = [ + {'entityID': 'https://idp1.example.org/idp', 'title': 'Example University'}, + {'entityID': 'https://idp2.example.org/idp', 'title': 'Östergötland'}, + ] + _response, chunks = self._search(results, q='example') + assert chunks + assert all(isinstance(chunk, bytes) for chunk in chunks) + assert json.loads(b''.join(chunks)) == results + + def test_single_result(self): + results = [{'entityID': 'https://idp.example.org/idp', 'title': 'Example University'}] + _response, chunks = self._search(results, q='example') + assert all(isinstance(chunk, bytes) for chunk in chunks) + assert json.loads(b''.join(chunks)) == results + + def test_no_results(self): + _response, chunks = self._search([], q='nothing') + assert all(isinstance(chunk, bytes) for chunk in chunks) + assert b''.join(chunks) == b'[]' + + def test_content_type(self): + response, _chunks = self._search([], q='example') + assert response.content_type == 'application/json' + + class PyFFAPITestResources(PipeLineTest): """ Runs twill tests using the wsgi-intercept diff --git a/src/pyff/test/test_repo.py b/src/pyff/test/test_repo.py index db8bfe2c..a731d06e 100644 --- a/src/pyff/test/test_repo.py +++ b/src/pyff/test/test_repo.py @@ -74,6 +74,48 @@ def test_lang_sv(self): assert name == 'Example University' assert desc == 'Identity Provider for Example University' + def test_extended_display_empty_description(self): + e = deepcopy(root(self.t)) + for elt in e.iter("{{{}}}Description".format(NS['mdui'])): + elt.text = None + + name, desc = entity_extended_display(e) + assert name == 'Example University' + assert desc == '' + + def test_extended_display_empty_url_without_description(self): + e = deepcopy(root(self.t)) + for elt in list(e.iter("{{{}}}Description".format(NS['mdui']))): + elt.getparent().remove(elt) + for elt in e.iter("{{{}}}OrganizationURL".format(NS['md'])): + elt.text = None + + name, desc = entity_extended_display(e) + assert name == 'Example University' + assert desc == '' + + def test_extended_display_url_without_description(self): + e = root(parse_xml(os.path.join(self.datadir, 'test04-trailing-slash-sp.xml'))) + + name, desc = entity_extended_display(e) + assert name == 'Example Service' + assert desc == 'https://www.example.com/' + + def test_extended_display_empty_names(self): + e = deepcopy(root(self.t)) + for tag in ( + "{{{}}}DisplayName".format(NS['mdui']), + "{{{}}}OrganizationName".format(NS['md']), + "{{{}}}OrganizationDisplayName".format(NS['md']), + "{{{}}}ServiceName".format(NS['md']), + ): + for elt in e.iter(tag): + elt.text = None + + name, desc = entity_extended_display(e) + assert isinstance(name, str) + assert isinstance(desc, str) + def test_utils(self): entity_id = root(self.t).get('entityID') self.md.store.update(root(self.t), entity_id)