test_integration.py

# -*- coding: utf-8 -*-

import datetime
import locale
import os
import unittest

from pyramid.wsgi import wsgiapp
from pyramid.view import view_config
from pyramid.static import static_view
from pyramid.compat import (
    text_,
    url_quote,
    )

from zope.interface import Interface

# 5 years from now (more or less)
fiveyrsfuture = datetime.datetime.utcnow() + datetime.timedelta(5*365)

defaultlocale = locale.getdefaultlocale()[1]

class INothing(Interface):
    pass

@view_config(for_=INothing)
@wsgiapp
def wsgiapptest(environ, start_response):
    """ """
    return '123'

class WGSIAppPlusViewConfigTests(unittest.TestCase):
    def test_it(self):
        from venusian import ATTACH_ATTR
        import types
        self.assertTrue(getattr(wsgiapptest, ATTACH_ATTR))
        self.assertTrue(type(wsgiapptest) is types.FunctionType)
        context = DummyContext()
        request = DummyRequest()
        result = wsgiapptest(context, request)
        self.assertEqual(result, '123')

    def test_scanned(self):
        from pyramid.interfaces import IRequest
        from pyramid.interfaces import IView
        from pyramid.interfaces import IViewClassifier
        from pyramid.config import Configurator
        from pyramid.tests import test_integration
        config = Configurator()
        config.scan(test_integration)
        config.commit()
        reg = config.registry
        view = reg.adapters.lookup(
            (IViewClassifier, IRequest, INothing), IView, name='')
        self.assertEqual(view.__original_view__, wsgiapptest)

class IntegrationBase(object):
    root_factory = None
    package = None
    def setUp(self):
        from pyramid.config import Configurator
        config = Configurator(root_factory=self.root_factory,
                              package=self.package)
        config.include(self.package)
        app = config.make_wsgi_app()
        from webtest import TestApp
        self.testapp = TestApp(app)
        self.config = config

    def tearDown(self):
        self.config.end()

here = os.path.dirname(__file__)

class TestStaticAppBase(IntegrationBase):
    def test_basic(self):
        res = self.testapp.get('/minimal.pt', status=200)
        _assertBody(res.body, os.path.join(here, 'fixtures/minimal.pt'))

    def test_hidden(self):
        res = self.testapp.get('/static/.hiddenfile', status=200)
        _assertBody(res.body, os.path.join(here, 'fixtures/static/.hiddenfile'))

    if defaultlocale is not None:
        # These tests are expected to fail on LANG=C systems
        def test_highchars_in_pathelement(self):
            url = url_quote('/static/héhé/index.html')
            res = self.testapp.get(url, status=200)
            _assertBody(
                res.body,
                os.path.join(here,
                             text_('fixtures/static/héhé/index.html', 'utf-8'))
                )

        def test_highchars_in_filename(self):
            url = url_quote('/static/héhé.html')
            res = self.testapp.get(url, status=200)
            _assertBody(
                res.body,
                os.path.join(here,
                             text_('fixtures/static/héhé.html', 'utf-8'))
                )

    def test_not_modified(self):
        self.testapp.extra_environ = {
            'HTTP_IF_MODIFIED_SINCE':httpdate(fiveyrsfuture)}
        res = self.testapp.get('/minimal.pt', status=304)
        self.assertEqual(res.body, b'')

    def test_file_in_subdir(self):
        fn = os.path.join(here, 'fixtures/static/index.html')
        res = self.testapp.get('/static/index.html', status=200)
        _assertBody(res.body, fn)

    def test_directory_noslash_redir(self):
        res = self.testapp.get('/static', status=301)
        self.assertEqual(res.headers['Location'], 'http://localhost/static/')

    def test_directory_noslash_redir_preserves_qs(self):
        res = self.testapp.get('/static?a=1&b=2', status=301)
        self.assertEqual(res.headers['Location'],
                         'http://localhost/static/?a=1&b=2')

    def test_directory_noslash_redir_with_scriptname(self):
        self.testapp.extra_environ = {'SCRIPT_NAME':'/script_name'}
        res = self.testapp.get('/static', status=301)
        self.assertEqual(res.headers['Location'],
                         'http://localhost/script_name/static/')

    def test_directory_withslash(self):
        fn = os.path.join(here, 'fixtures/static/index.html')
        res = self.testapp.get('/static/', status=200)
        _assertBody(res.body, fn)

    def test_range_inclusive(self):
        self.testapp.extra_environ = {'HTTP_RANGE':'bytes=1-2'}
        res = self.testapp.get('/static/index.html', status=206)
        self.assertEqual(res.body, b'ht')

    def test_range_tilend(self):
        self.testapp.extra_environ = {'HTTP_RANGE':'bytes=-5'}
        res = self.testapp.get('/static/index.html', status=206)
        self.assertEqual(res.body, b'tml>\n')

    def test_range_notbytes(self):
        self.testapp.extra_environ = {'HTTP_RANGE':'kHz=-5'}
        res = self.testapp.get('/static/index.html', status=200)
        _assertBody(res.body,
                    os.path.join(here, 'fixtures/static/index.html'))

    def test_range_multiple(self):
        res = self.testapp.get('/static/index.html',
                               [('HTTP_RANGE', 'bytes=10-11,11-12')],
                               status=200)
        _assertBody(res.body,
                    os.path.join(here, 'fixtures/static/index.html'))

    def test_range_oob(self):
        self.testapp.extra_environ = {'HTTP_RANGE':'bytes=1000-1002'}
        self.testapp.get('/static/index.html', status=416)

    def test_notfound(self):
        self.testapp.get('/static/wontbefound.html', status=404)

    def test_oob_dotdotslash(self):
        self.testapp.get('/static/../../test_integration.py', status=404)

    def test_oob_dotdotslash_encoded(self):
        self.testapp.get('/static/%2E%2E%2F/test_integration.py', status=404)

    def test_oob_slash(self):
        self.testapp.get('/%2F/test_integration.py', status=404)

class TestStaticAppUsingAbsPath(TestStaticAppBase, unittest.TestCase):
    package = 'pyramid.tests.pkgs.static_abspath'

class TestStaticAppUsingAssetSpec(TestStaticAppBase, unittest.TestCase):
    package = 'pyramid.tests.pkgs.static_assetspec'

class TestStaticAppNoSubpath(unittest.TestCase):
    staticapp = static_view(os.path.join(here, 'fixtures'), use_subpath=False)
    def _makeRequest(self, extra):
        from pyramid.request import Request
        from io import BytesIO
        kw = {'PATH_INFO':'',
              'SCRIPT_NAME':'',
              'SERVER_NAME':'localhost',
              'SERVER_PORT':'80',
              'REQUEST_METHOD':'GET',
              'wsgi.version':(1,0),
              'wsgi.url_scheme':'http',
              'wsgi.input':BytesIO()}
        kw.update(extra)
        request = Request(kw)
        return request

    def test_basic(self):
        request = self._makeRequest({'PATH_INFO':'/minimal.pt'})
        context = DummyContext()
        result = self.staticapp(context, request)
        self.assertEqual(result.status, '200 OK')
        _assertBody(result.body, os.path.join(here, 'fixtures/minimal.pt'))

class TestStaticAppWithRoutePrefix(IntegrationBase, unittest.TestCase):
    package = 'pyramid.tests.pkgs.static_routeprefix'

    def test_includelevel1(self):
        res = self.testapp.get('/static/minimal.pt', status=200)
        _assertBody(res.body,
                    os.path.join(here, 'fixtures/minimal.pt'))

    def test_includelevel2(self):
        res = self.testapp.get('/prefix/static/index.html', status=200)
        _assertBody(res.body,
                    os.path.join(here, 'fixtures/static/index.html'))


class TestFixtureApp(IntegrationBase, unittest.TestCase):
    package = 'pyramid.tests.pkgs.fixtureapp'
    def test_another(self):
        res = self.testapp.get('/another.html', status=200)
        self.assertEqual(res.body, b'fixture')

    def test_root(self):
        res = self.testapp.get('/', status=200)
        self.assertEqual(res.body, b'fixture')

    def test_dummyskin(self):
        self.testapp.get('/dummyskin.html', status=404)

    def test_error(self):
        res = self.testapp.get('/error.html', status=200)
        self.assertEqual(res.body, b'supressed')

    def test_protected(self):
        self.testapp.get('/protected.html', status=403)

class TestStaticPermApp(IntegrationBase, unittest.TestCase):
    package = 'pyramid.tests.pkgs.staticpermapp'
    root_factory = 'pyramid.tests.pkgs.staticpermapp:RootFactory'
    def test_allowed(self):
        result = self.testapp.get('/allowed/index.html', status=200)
        _assertBody(result.body,
                    os.path.join(here, 'fixtures/static/index.html'))

    def test_denied_via_acl_global_root_factory(self):
        self.testapp.extra_environ = {'REMOTE_USER':'bob'}
        self.testapp.get('/protected/index.html', status=403)

    def test_allowed_via_acl_global_root_factory(self):
        self.testapp.extra_environ = {'REMOTE_USER':'fred'}
        result = self.testapp.get('/protected/index.html', status=200)
        _assertBody(result.body, 
                    os.path.join(here, 'fixtures/static/index.html'))

    def test_denied_via_acl_local_root_factory(self):
        self.testapp.extra_environ = {'REMOTE_USER':'fred'}
        self.testapp.get('/factory_protected/index.html', status=403)

    def test_allowed_via_acl_local_root_factory(self):
        self.testapp.extra_environ = {'REMOTE_USER':'bob'}
        result = self.testapp.get('/factory_protected/index.html', status=200)
        _assertBody(result.body,
                    os.path.join(here, 'fixtures/static/index.html'))

class TestCCBug(IntegrationBase, unittest.TestCase):
    # "unordered" as reported in IRC by author of
    # http://labs.creativecommons.org/2010/01/13/cc-engine-and-web-non-frameworks/
    package = 'pyramid.tests.pkgs.ccbugapp'
    def test_rdf(self):
        res = self.testapp.get('/licenses/1/v1/rdf', status=200)
        self.assertEqual(res.body, b'rdf')

    def test_juri(self):
        res = self.testapp.get('/licenses/1/v1/juri', status=200)
        self.assertEqual(res.body, b'juri')

class TestHybridApp(IntegrationBase, unittest.TestCase):
    # make sure views registered for a route "win" over views registered
    # without one, even though the context of the non-route view may
    # be more specific than the route view.
    package = 'pyramid.tests.pkgs.hybridapp'
    def test_root(self):
        res = self.testapp.get('/', status=200)
        self.assertEqual(res.body, b'global')

    def test_abc(self):
        res = self.testapp.get('/abc', status=200)
        self.assertEqual(res.body, b'route')

    def test_def(self):
        res = self.testapp.get('/def', status=200)
        self.assertEqual(res.body, b'route2')

    def test_ghi(self):
        res = self.testapp.get('/ghi', status=200)
        self.assertEqual(res.body, b'global')

    def test_jkl(self):
        self.testapp.get('/jkl', status=404)

    def test_mno(self):
        self.testapp.get('/mno', status=404)

    def test_pqr_global2(self):
        res = self.testapp.get('/pqr/global2', status=200)
        self.assertEqual(res.body, b'global2')

    def test_error(self):
        res = self.testapp.get('/error', status=200)
        self.assertEqual(res.body, b'supressed')

    def test_error2(self):
        res = self.testapp.get('/error2', status=200)
        self.assertEqual(res.body, b'supressed2')

    def test_error_sub(self):
        res = self.testapp.get('/error_sub', status=200)
        self.assertEqual(res.body, b'supressed2')

class TestRestBugApp(IntegrationBase, unittest.TestCase):
    # test bug reported by delijati 2010/2/3 (http://pastebin.com/d4cc15515)
    package = 'pyramid.tests.pkgs.restbugapp'
    def test_it(self):
        res = self.testapp.get('/pet', status=200)
        self.assertEqual(res.body, b'gotten')

class TestForbiddenAppHasResult(IntegrationBase, unittest.TestCase):
    # test that forbidden exception has ACLDenied result attached
    package = 'pyramid.tests.pkgs.forbiddenapp'
    def test_it(self):
        res = self.testapp.get('/x', status=403)
        message, result = [x.strip() for x in res.body.split(b'\n')]
        self.assertTrue(message.endswith(b'failed permission check'))
        self.assertTrue(
            result.startswith(b"ACLDenied permission 'private' via ACE "
                              b"'<default deny>' in ACL "
                              b"'<No ACL found on any object in resource "
                              b"lineage>' on context"))
        self.assertTrue(
            result.endswith(b"for principals ['system.Everyone']"))

class TestViewDecoratorApp(IntegrationBase, unittest.TestCase):
    package = 'pyramid.tests.pkgs.viewdecoratorapp'
    def _configure_mako(self):
        tmpldir = os.path.join(os.path.dirname(__file__),
                               'pkgs',
                               'viewdecoratorapp',
                               'views')
        self.config.registry.settings['mako.directories'] = tmpldir

    def test_first(self):
        # we use mako here instead of chameleon because it works on Jython
        self._configure_mako()
        res = self.testapp.get('/first', status=200)
        self.assertTrue(b'OK' in res.body)

    def test_second(self):
        # we use mako here instead of chameleon because it works on Jython
        self._configure_mako()
        res = self.testapp.get('/second', status=200)
        self.assertTrue(b'OK2' in res.body)

class TestNotFoundView(IntegrationBase, unittest.TestCase):
    package = 'pyramid.tests.pkgs.notfoundview'

    def test_it(self):
        res = self.testapp.get('/wontbefound', status=200)
        self.assertTrue(b'generic_notfound' in res.body)
        res = self.testapp.get('/bar', status=302)
        self.assertEqual(res.location, 'http://localhost/bar/')
        res = self.testapp.get('/bar/', status=200)
        self.assertTrue(b'OK bar' in res.body)
        res = self.testapp.get('/foo', status=302)
        self.assertEqual(res.location, 'http://localhost/foo/')
        res = self.testapp.get('/foo/', status=200)
        self.assertTrue(b'OK foo2' in res.body)
        res = self.testapp.get('/baz', status=200)
        self.assertTrue(b'baz_notfound' in res.body)

class TestForbiddenView(IntegrationBase, unittest.TestCase):
    package = 'pyramid.tests.pkgs.forbiddenview'

    def test_it(self):
        res = self.testapp.get('/foo', status=200)
        self.assertTrue(b'foo_forbidden' in res.body)
        res = self.testapp.get('/bar', status=200)
        self.assertTrue(b'generic_forbidden' in res.body)
        
class TestViewPermissionBug(IntegrationBase, unittest.TestCase):
    # view_execution_permitted bug as reported by Shane at http://lists.repoze.org/pipermail/repoze-dev/2010-October/003603.html
    package = 'pyramid.tests.pkgs.permbugapp'
    def test_test(self):
        res = self.testapp.get('/test', status=200)
        self.assertTrue(b'ACLDenied' in res.body)

    def test_x(self):
        self.testapp.get('/x', status=403)

class TestDefaultViewPermissionBug(IntegrationBase, unittest.TestCase):
    # default_view_permission bug as reported by Wiggy at http://lists.repoze.org/pipermail/repoze-dev/2010-October/003602.html
    package = 'pyramid.tests.pkgs.defpermbugapp'
    def test_x(self):
        res = self.testapp.get('/x', status=403)
        self.assertTrue(b'failed permission check' in res.body)

    def test_y(self):
        res = self.testapp.get('/y', status=403)
        self.assertTrue(b'failed permission check' in res.body)

    def test_z(self):
        res = self.testapp.get('/z', status=200)
        self.assertTrue(b'public' in res.body)

from pyramid.tests.pkgs.exceptionviewapp.models import \
     AnException, NotAnException
excroot = {'anexception':AnException(),
           'notanexception':NotAnException()}

class TestExceptionViewsApp(IntegrationBase, unittest.TestCase):
    package = 'pyramid.tests.pkgs.exceptionviewapp'
    root_factory = lambda *arg: excroot
    def test_root(self):
        res = self.testapp.get('/', status=200)
        self.assertTrue(b'maybe' in res.body)

    def test_notanexception(self):
        res = self.testapp.get('/notanexception', status=200)
        self.assertTrue(b'no' in res.body)

    def test_anexception(self):
        res = self.testapp.get('/anexception', status=200)
        self.assertTrue(b'yes' in res.body)

    def test_route_raise_exception(self):
        res = self.testapp.get('/route_raise_exception', status=200)
        self.assertTrue(b'yes' in res.body)

    def test_route_raise_exception2(self):
        res = self.testapp.get('/route_raise_exception2', status=200)
        self.assertTrue(b'yes' in res.body)

    def test_route_raise_exception3(self):
        res = self.testapp.get('/route_raise_exception3', status=200)
        self.assertTrue(b'whoa' in res.body)

    def test_route_raise_exception4(self):
        res = self.testapp.get('/route_raise_exception4', status=200)
        self.assertTrue(b'whoa' in res.body)

class TestConflictApp(unittest.TestCase):
    package = 'pyramid.tests.pkgs.conflictapp'
    def _makeConfig(self):
        from pyramid.config import Configurator
        config = Configurator()
        return config

    def test_autoresolved_view(self):
        config = self._makeConfig()
        config.include(self.package)
        app = config.make_wsgi_app()
        from webtest import TestApp
        self.testapp = TestApp(app)
        res = self.testapp.get('/')
        self.assertTrue(b'a view' in res.body)
        res = self.testapp.get('/route')
        self.assertTrue(b'route view' in res.body)

    def test_overridden_autoresolved_view(self):
        from pyramid.response import Response
        config = self._makeConfig()
        config.include(self.package)
        def thisview(request):
            return Response('this view')
        config.add_view(thisview)
        app = config.make_wsgi_app()
        from webtest import TestApp
        self.testapp = TestApp(app)
        res = self.testapp.get('/')
        self.assertTrue(b'this view' in res.body)

    def test_overridden_route_view(self):
        from pyramid.response import Response
        config = self._makeConfig()
        config.include(self.package)
        def thisview(request):
            return Response('this view')
        config.add_view(thisview, route_name='aroute')
        app = config.make_wsgi_app()
        from webtest import TestApp
        self.testapp = TestApp(app)
        res = self.testapp.get('/route')
        self.assertTrue(b'this view' in res.body)

    def test_nonoverridden_authorization_policy(self):
        config = self._makeConfig()
        config.include(self.package)
        app = config.make_wsgi_app()
        from webtest import TestApp
        self.testapp = TestApp(app)
        res = self.testapp.get('/protected', status=403)
        self.assertTrue(b'403 Forbidden' in res.body)

    def test_overridden_authorization_policy(self):
        config = self._makeConfig()
        config.include(self.package)
        from pyramid.testing import DummySecurityPolicy
        config.set_authorization_policy(DummySecurityPolicy('fred'))
        config.set_authentication_policy(DummySecurityPolicy(permissive=True))
        app = config.make_wsgi_app()
        from webtest import TestApp
        self.testapp = TestApp(app)
        res = self.testapp.get('/protected', status=200)
        self.assertTrue('protected view' in res)

class ImperativeIncludeConfigurationTest(unittest.TestCase):
    def setUp(self):
        from pyramid.config import Configurator
        config = Configurator()
        from pyramid.tests.pkgs.includeapp1.root import configure
        configure(config)
        app = config.make_wsgi_app()
        from webtest import TestApp
        self.testapp = TestApp(app)
        self.config = config

    def tearDown(self):
        self.config.end()

    def test_root(self):
        res = self.testapp.get('/', status=200)
        self.assertTrue(b'root' in res.body)

    def test_two(self):
        res = self.testapp.get('/two', status=200)
        self.assertTrue(b'two' in res.body)

    def test_three(self):
        res = self.testapp.get('/three', status=200)
        self.assertTrue(b'three' in res.body)

class SelfScanAppTest(unittest.TestCase):
    def setUp(self):
        from pyramid.tests.test_config.pkgs.selfscan import main
        config = main()
        app = config.make_wsgi_app()
        from webtest import TestApp
        self.testapp = TestApp(app)
        self.config = config

    def tearDown(self):
        self.config.end()

    def test_root(self):
        res = self.testapp.get('/', status=200)
        self.assertTrue(b'root' in res.body)

    def test_two(self):
        res = self.testapp.get('/two', status=200)
        self.assertTrue(b'two' in res.body)

class WSGIApp2AppTest(unittest.TestCase):
    def setUp(self):
        from pyramid.tests.pkgs.wsgiapp2app import main
        config = main()
        app = config.make_wsgi_app()
        from webtest import TestApp
        self.testapp = TestApp(app)
        self.config = config

    def tearDown(self):
        self.config.end()

    def test_hello(self):
        res = self.testapp.get('/hello', status=200)
        self.assertTrue(b'Hello' in res.body)

if os.name != 'java': # uses chameleon
    class RendererScanAppTest(IntegrationBase, unittest.TestCase):
        package = 'pyramid.tests.pkgs.rendererscanapp'
        def test_root(self):
            res = self.testapp.get('/one', status=200)
            self.assertTrue(b'One!' in res.body)

        def test_two(self):
            res = self.testapp.get('/two', status=200)
            self.assertTrue(b'Two!' in res.body)

        def test_rescan(self):
            self.config.scan('pyramid.tests.pkgs.rendererscanapp')
            app = self.config.make_wsgi_app()
            from webtest import TestApp
            testapp = TestApp(app)
            res = testapp.get('/one', status=200)
            self.assertTrue(b'One!' in res.body)
            res = testapp.get('/two', status=200)
            self.assertTrue(b'Two!' in res.body)

class DummyContext(object):
    pass

class DummyRequest:
    subpath = ('__init__.py',)
    traversed = None
    environ = {'REQUEST_METHOD':'GET', 'wsgi.version':(1,0)}
    def get_response(self, application):
        return application(None, None)

def httpdate(ts):
    return ts.strftime("%a, %d %b %Y %H:%M:%S GMT")

def read_(filename):
    with open(filename, 'rb') as fp:
        val = fp.read()
        return val
    
def _assertBody(body, filename):
    if defaultlocale is None: # pragma: no cover
        # If system locale does not have an encoding then default to utf-8
        filename = filename.encode('utf-8')
    # strip both \n and \r for windows
    body = body.replace(b'\r', b'')
    body = body.replace(b'\n', b'')
    data = read_(filename)
    data = data.replace(b'\r', b'')
    data = data.replace(b'\n', b'')
    assert(body == data)