def views(self, map_config, redis, session):
     request = DummyRequest()
     with testing.testConfig(request=request) as config:
         config.include('pyramid_chameleon')
         setattr(request, 'db_session', session)
         setattr(request.registry, 'redis_client', redis)
         yield ContentViews(request)

def test_subscriber_predicate(settings):
    """Test that the ``asset_request`` subscriber predicate.

    It should correctly match asset requests when its value is ``True``,
    and other requests when ``False``.
    """
    mock1 = Mock()
    mock2 = Mock()

    with testConfig(settings=settings) as config:
        config.include(assets)
        config.add_subscriber(mock1, DummyEvent, asset_request=False)
        config.add_subscriber(mock2, DummyEvent, asset_request=True)

        request1 = DummyRequest('/')
        request1.matched_route = None

        pattern = config.get_webassets_env().url + '*subpath'
        request2 = DummyRequest(config.get_webassets_env().url + '/t.png')
        request2.matched_route = Route('__' + pattern, pattern)

        event1 = DummyEvent(request1)
        event2 = DummyEvent(request2)

        config.registry.notify(event1)
        config.registry.notify(event2)

        mock1.assert_called_onceventwith(event1)
        mock2.assert_called_onceventwith(event2)

 def test_functional_using_assetspec(self):
     from ..._compat import u
     renderer = self._makeOne(())
     with testing.testConfig() as config:
         config.include('pyramid_chameleon')
         result = renderer('substanced.widget.tests:fixtures/test.pt')
     self.assertEqual(result.strip(), u('<div>Test</div>'))

    def test_policy_selected_event(self):
        from pyramid.testing import testConfig
        from pyramid_multiauth import MultiAuthPolicySelected

        policies = [TestAuthnPolicy2(), TestAuthnPolicy3()]
        policy = MultiAuthenticationPolicy(policies)
        # Simulate loading from config:
        policies[0]._pyramid_multiauth_name = "name"

        with testConfig() as config:
            request = DummyRequest()

            selected_policy = []

            def track_policy(event):
                selected_policy.append(event)

            config.add_subscriber(track_policy, MultiAuthPolicySelected)

            self.assertEquals(policy.authenticated_userid(request), "test2")

            self.assertEquals(selected_policy[0].policy, policies[0])
            self.assertEquals(selected_policy[0].policy_name, "name")
            self.assertEquals(selected_policy[0].userid, "test2")
            self.assertEquals(selected_policy[0].request, request)
            self.assertEquals(len(selected_policy), 1)

            # Effective principals also triggers an event when groupfinder
            # is provided.
            policy_with_group = MultiAuthenticationPolicy(policies,
                                                          lambda u, r: ['foo'])
            policy_with_group.effective_principals(request)
            self.assertEquals(len(selected_policy), 2)

    def test_it_another_request_type(self):
        """ on /foo/craete, success redirect is /foo/list """
        from pyramid.testing import testConfig
        from zope.interface import Interface
        class IAPIRequest(Interface):
            pass

        with testConfig(autocommit=False) as config:
            config.include("mokehehe.workflow")
            config.add_route("foo.list", "/foo/list")
            config.add_route("foo.create", "/foo/create")
            config.add_route("foo.api.list", "/api/foo/list")

            with config.get_workflow().sub(IAfterModelCreate) as goto:
                goto("foo.create", "foo.list")
                goto("foo.create", "foo.api.list", request_type=IAPIRequest)
            config.commit()

            ### on request time. api request
            request2 = self._makeRequest(config)
            class route:
                name = "foo.create"
            request2.matched_route = route
            from zope.interface import directlyProvides
            directlyProvides(request2, IAPIRequest) ##on tweens

            result = request2.workflow[IAfterModelCreate].route_path()
            self.assertEqual(result, "/api/foo/list")

    def test_it(self):
        from pyramid.testing import testConfig
        from mokehehe.viewobject import Mapper, parse_get, viewobject_method

        def todict(v):
            return {"value": v}

        with testConfig() as config:
            class Ob(object):
                def __call__(self, message):
                    return message * 2

                @viewobject_method
                def nihongo(self):
                    return "hai"

            config.include("mokehehe.viewobject")
            config.add_route("hello", "/")
            MOb = config.add_mapped_viewobject(Mapper(parse_get, todict), Ob)
            self.assertIsNotNone(MOb)
            config.add_view(MOb, route_name="hello", renderer="json")
            config.add_view(MOb, route_name="hello", request_type=INihongoRequest, attr="nihongo", renderer="json")
            config.commit()

            result = self.callView(config, "/hello", GET={"message": "hello"})
            self.assertEqual(result.body.decode("utf-8"), u'{"value": "hellohello"}')

            result = self.callView(config, "/hello", iface=INihongoRequest, GET={})
            self.assertEqual(result.body.decode("utf-8"), u'{"value": "hai"}')

def test_includeme_with_usage(db_settings, db_wipe):
    # Initialize a table to ensure table reflection is working.
    conn_str = db_settings['db.common.url']
    with psycopg2.connect(conn_str) as conn:
        with conn.cursor() as cur:
            cur.execute("CREATE TABLE smurfs ("
                        "  name TEXT PRIMARY KEY,"
                        "  role TEXT,"
                        "  tastiness INTEGER);")
        conn.commit()

    # Call the target function
    request = testing.DummyRequest()
    with testing.testConfig(request=request, settings=db_settings) as config:
        includeme(config)

        # The request doesn't yet have the added methods and I'm not sure
        # how to make the test request have those, so we'll call them
        # directly instead. We aren't testing the framework's
        # add_request_method anyhow.

        # Check the engines definitions
        # ... looking for the unnamed util lookup
        assert str(get_db_engine(request).url) == db_settings['db.common.url']
        # ... looking for the named util lookup
        expected_url = db_settings['db.readonly.url']
        assert str(get_db_engine(request, 'readonly').url) == expected_url

        with pytest.raises(ComponentLookupError):
            get_db_engine(request, 'foobar')

        # Check the tables definition
        assert hasattr(db_tables(request), 'smurfs')

 def test_form_submitted_success_w_locator_adapter(self):
     from pyramid.testing import testConfig
     from zope.interface import Interface
     from ....interfaces import IUserLocator
     from ....testing import make_site
     request = testing.DummyRequest()
     request.params['form.submitted'] = True
     request.params['login'] = 'login'
     request.params['password'] = 'password'
     request.sdiapi = DummySDIAPI()
     request.params['csrf_token'] = request.session.get_csrf_token()
     context = make_site()
     user = DummyUser(1)
     user.__oid__ = 2
     locator = DummyLocator(user)
     def _locator(context, request):
         return locator
     with testConfig() as config:
         config.testing_add_renderer(
                 'substanced.sdi.views:templates/login.pt')
         config.registry.registerAdapter(_locator, (Interface, Interface),
                                         IUserLocator)
         result = self._callFUT(context, request)
     self.assertEqual(result.location, 'http://example.com')
     self.assertTrue(result.headers)
     self.assertTrue('sdi.came_from' not in request.session)

 def test_alternate_homepage(self, config):
     request = DummyRequest()
     with testing.testConfig(request=request) as config:
         assert not configure_content(config, enable=False)
         res = empty_homepage_view(request)
         assert res.content_type == 'text/html'
         assert b'It works' in res.body

def test_reply_notification_content():
    """
    The reply notification should have a subject, and both plain and
    html bodies.
    """
    with testConfig() as config:
        config.include('pyramid_chameleon')

        annotation = create_annotation()
        request = DummyRequest()

        with patch('h.auth.local.models.User') as mock_user:
            user = Mock(email='acct:[email protected]')
            mock_user.get_by_username.return_value = user

            notification = notifier.ReplyTemplate.generate_notification(
                request, annotation, {})

            assert notification['status']
            assert notification['recipients'] == ['acct:[email protected]']
            assert 'testuser has just left a reply to your annotation on' in \
                notification['text']
            assert '<a href="http://example.com/u/testuser">testuser</a> '\
                'has just left a reply to your annotation on' \
                in notification['html']
            assert notification['subject'] == \
                'testuser has replied to your annotation\n'

    def test_logging(self):
        """Test that the includeme function loads the logging config."""
        from .main import includeme
        with testing.testConfig(settings=self.settings) as config:
            includeme(config)

        # Set up the loggers
        root = logging.getLogger()
        local = logging.getLogger('pyramid_sawing.tests')

        # Try logging...
        root_log_msg = 'O⋮o'
        root.error(root_log_msg)
        local_log_msg = '>,<'
        local.info(local_log_msg)

        global logio
        logio.seek(self.logio_position)
        log_lines = logio.readlines()
        self.assertEqual(len(log_lines), 2)
        log_lines = [l.rstrip('\n') for l in log_lines]

        # Check the root message...
        parsed_root_msg = log_lines[0].split(';.;')
        self.assertEqual(parsed_root_msg, ['ERROR', root_log_msg])
        parsed_local_msg = log_lines[1].split(';.;')
        self.assertEqual(parsed_local_msg,
                         ['INFO ', socket.gethostname(), local_log_msg])

    def test_multiple(self):
        from pyramid.testing import testConfig

        class model:
            name = "model"
        class model2:
            name = "model2"

        def resource_factory(target):
            return target

        def create_route_name(target, fullname):
            return "{}.{}".format(fullname, target.name)

        def create_path(target, fullname):
            return "/{}/{}".format(fullname.replace(".", "/"), target.name)

        def myview(context, request):
            return "ok"

        with testConfig() as config:
            config.include("mokehehe.registering")
            builder = config.registering_builder("foo", resource_factory=resource_factory)
            with builder.sub("bar", route_name_craete=create_route_name, path_create=create_path) as view_builder:
                view_builder.build(view=myview).update(renderer="json")

            ## call
            builder(model)
            builder(model2)

            self.assertEqual(len(config.get_routes_mapper().get_routes()), 2)

    def test_logging_a_request(self):
        request = Request.blank('/foo')
        request.environ.update({
            'HTTP_VERSION': '1.1',
            'REMOTE_ADDR': '127.0.0.1',
            })
        response = mock.Mock()
        response.status = '404 Not Found'
        response.headers = {'content-length': 0}
        handler = mock.Mock(return_value=response)
        config_kwargs = {'request': request, 'settings': self.settings}
        with testing.testConfig(**config_kwargs) as config:
            request.registry = config.registry
            tween = self.make_one(handler=handler, registry=config.registry)
            tween(request)

        global logio
        logio.seek(self.logio_position)
        log_lines = logio.readlines()
        self.assertEqual(len(log_lines), 1)
        log_lines = [l.rstrip('\n') for l in log_lines]

        log_line = log_lines[0]
        self.assertEqual(
            log_line,
            '127.0.0.1 - - [asctime] "GET http://localhost:80/foo 1.1" 404 Not Found 0 "-" "-"')

 def test_main_template(self):
     self.config.testing_securitypolicy(permissive=False)
     request = testing.DummyRequest()
     inst = self._makeOne(request)
     with testing.testConfig() as config:
         config.include("pyramid_chameleon")
         self.assertTrue(inst.main_template)

 def test_get_macro_with_name(self):
     request = testing.DummyRequest()
     inst = self._makeOne(request)
     with testing.testConfig() as config:
         config.include("pyramid_chameleon")
         macro = inst.get_macro("substanced.sdi.views:templates/master.pt", "main")
     self.assertTrue(macro.include)

 def test_view_notfound(self):
     with testing.testConfig():
         request = testing.DummyRequest()
         request.matchdict['prefix'] = 'aaaa'
         request.matchdict['name'] = 'nonexistent-image.jpg'
         with self.assertRaises(HTTPNotFound):
             ImageView(request)()

def test_it():
    from pyramid.testing import testConfig

    with testConfig() as config:
        config.include("stringexchange")
        config.add_route("hello", "/")

        def hello_view(context, request):
            from pyramid.response import Response
            js = request.string_exchange.publisher("js")
            response = Response("""
            <html><head>{}</head><body></body></html>
            """.format(request.string_exchange.subscribe("js")))

            js.publish('<script src="my.js></script>"')
            assert "my.js" not in response.text
            return response

        config.add_view(hello_view, route_name="hello")

        # request time
        router = _makeRouter(config)

        request = _makeRequest(config, path="/")
        response = router.handle_request(request)

        assert "my.js" in response.text

 def test_non_ascii_bytes_in_userid(self):
     from pyramid.request import Request
     byte_str = b'\xe6\xbc\xa2'
     with testing.testConfig() as config:
         config.testing_securitypolicy(userid=byte_str)
         request = Request.blank('/')
         msg = self._callFUT(request)
     self.assertTrue(repr(byte_str) in msg, msg)

def test_csa_founder_signup(customer_mock, charge_mock, post):
    with testing.testConfig() as config:
        config.add_route('signup_success', '/success')
        request = testing.DummyRequest(post=post)
        request.registry.settings['stripe.api_key'] = 'api_key'

        with pytest.raises(HTTPFound):
            csa_founder_signup(request)

 def test_integer_user_id(self):
     # userids can apparently be integers as well
     from pyramid.request import Request
     with testing.testConfig() as config:
         config.testing_securitypolicy(userid=42)
         request = Request.blank('/')
         msg = self._callFUT(request)
     self.assertTrue('42' in msg)