def test_PUT_using_header(self):
     app = self._make_app()
     request = DummyRequest(method='POST')
     request.headers['X-HTTP-Method-Override'] = 'PUT'
     self._assert_before(request)
     app.registry.notify(NewRequest(request))
     self._assert_after(request, 'PUT')

    def test_forbidden(self):
        from ptahcrowd.forbidden import Forbidden

        class Context(object):
            """ """

        request = DummyRequest()
        request.url = 'http://example.com'
        request.application_url = 'http://example.com'
        request.root = Context()

        excview = Forbidden(HTTPForbidden(), request)
        excview.update()

        res = request.response

        self.assertIs(excview.__parent__, request.root)
        self.assertEqual(res.status, '302 Found')
        self.assertEqual(
            text_(res.headers['location']),
            'http://example.com/login.html?came_from=http%3A%2F%2Fexample.com')

        excview = Forbidden(HTTPForbidden(), request)
        res = excview()
        self.assertEqual(res.status, '302 Found')

    def test__get_child_layers_info_without_scalehint(self):
        from pyramid.testing import DummyRequest
        from c2cgeoportal.views.entry import Entry

        request = DummyRequest()
        request.user = None
        entry = Entry(request)

        class Layer(object):
            pass

        child_layer_1 = Layer()
        child_layer_1.name = 'layer_1'
        child_layer_1.scaleHint = None
        child_layer_1.layers = []

        child_layer_2 = Layer()
        child_layer_2.name = 'layer_2'
        child_layer_2.scaleHint = None
        child_layer_2.layers = []

        layer = Layer()
        layer.layers = [child_layer_1, child_layer_2]

        child_layers_info = entry._get_child_layers_info(layer)

        expected = [{
            'name': 'layer_1'
        }, {
            'name': 'layer_2',
        }]
        self.assertEqual(child_layers_info, expected)

    def test_profile_csv(self):
        from pyramid.testing import DummyRequest
        from c2cgeoportal.views.profile import Profile

        request = DummyRequest()
        request.registry.settings = {
            "raster": {
                "dem": {"file": "c2cgeoportal/tests/data/dem.shp", "round": 1},
                "dem2": {"file": "c2cgeoportal/tests/data/dem.shp", "round": 1}
            }
        }
        profile = Profile(request)

        request.params["nbPoints"] = "3"
        request.params["geom"] = '{"type":"LineString",' \
            '"coordinates":[[548009.5,215990],[547990,216009.5]]}'
        response = profile.csv()
        self.assertEqual(response.body, """distance,dem2,dem,x,y
0.0,1166,1166,548009,215990
9.2,1181,1181,548003,215996
18.4,1181,1181,547996,216003""")

        request.params["layers"] = "dem"
        response = profile.csv()
        self.assertEqual(response.body, """distance,dem,x,y
0.0,1166,548009,215990
9.2,1181,548003,215996
18.4,1181,547996,216003""")

 def dummy_request(self):
     request = DummyRequest()
     request.context = DummyResource()
     request.userdb = self.userdb
     request.db = self.db
     request.registry.settings = self.settings
     return request

    def _makeRequest(self, db, fs):
        from pyramid.testing import DummyRequest

        request = DummyRequest()
        request.db = db
        request.fs = fs
        return request

def test_websocket_good_origin(config):
    config.registry.settings.update({'origins': 'http://good'})
    config.include('h.streamer')
    req = DummyRequest(headers={'Origin': 'http://good'})
    req.get_response = MagicMock()
    res = websocket(req)
    assert res.code != 403

    def test_remember_me_with_bad_endpoint(self):
        """Test the post-login hook with a bad openid endpoint"""
        req = DummyRequest(params={
            'openid.op_endpoint': 'bad_endpoint',
        })
        req.db = self.db
        def flash(msg):
            pass
        req.session.flash = flash
        info = {
            'identity_url': 'http://lmacken.id.fedoraproject.org',
            'groups': [u'releng'],
        }
        req.registry.settings = self.app_settings

        try:
            resp = remember_me(None, req, info)
            assert False, 'remember_me should have thrown an exception'
        except Exception:
            # A ComponentLookupError is thrown because we're doing this outside
            # of the webapp
            pass

        # The user should not exist
        self.assertIsNone(User.get(u'lmacken', self.db))

 def test_get_current_page(self):
     request = DummyRequest()
     page = get_current_page(request)
     self.assertEqual(page, 1)
     request.GET['page'] = 5
     page = get_current_page(request)
     self.assertEqual(page, 5)

    def test_call(self):
        config = testing.setUp()
        config.add_route('some_route', '/')

        pagination_request_writer = PaginationRequestWriterMock()
        param_property_naming = PropertyNamingMock(
            'page-number',
            'page-size',
            'sort-property',
            'sort-direction')
        url_creator_factory = UrlCreatorFactoryMock('http://example.com')
        request = DummyRequest()
        request.registry.pagination = {
            'some_route': {
                'pagination_request_writer': pagination_request_writer,
                'param_property_naming': param_property_naming,
                'url_creator_factory': url_creator_factory,
            }
        }
        request.pagination = Pagination(
            Paging(
                100,
                PagingRequest(3, 12)),
            Sorting('some_property', 'desc'))
        request.matched_route = RouteMock('some_route')
        event = BeforeRenderEventMock()
        event['request'] = request
        subscriber = PaginationBeforeRenderSubscriber()
        subscriber(event)
        self.assertEqual(pagination_request_writer.call_count, 1)

        pagination = event.rendering_val['pagination']
        self.assertEqual('http://example.com', str(pagination.url))

 def test_build_url_forward_headers(self):
     request = DummyRequest()
     request.environ = {
         "SERVER_NAME": "example.com"
     }
     request.registry.settings = {
         "checker": {
             "forward_headers": ["Cookie"]
         }
     }
     request.headers["Cookie"] = "test"
     self.assertEqual(
         build_url(
             "Test",
             "https://camptocamp.com/toto?titi#tutu",
             request
         ),
         (
             "https://camptocamp.com/toto?titi#tutu",
             {
                 "Cache-Control": "no-cache",
                 "Cookie": "test",
             }
         )
     )

    def test_call_exception(self):

        class ExceptionPaginationReader(object):
            # pylint: disable=too-few-public-methods

            def __init__(self, exception):
                self._exception = exception

            def __call__(self, request, content_size_provider):
                raise self._exception


        content_size_provider_mock = ContentSizeProviderMock(100)

        request = DummyRequest()
        request.registry.pagination = {
            'some_route': {
                'content_size_provider': content_size_provider_mock,
                'pagination_reader': ExceptionPaginationReader(
                    PageNotFoundException()),
            }
        }
        request.matched_route = RouteMock('some_route')

        event = ContextFoundEventMock(request)
        subscriber = PaginationContextFoundSubscriber()
        with self.assertRaises(ParameterValidationException):
            subscriber(event)

    def _undo_delete_column(self, base, old_name, path, results):
        for doc in results:
            if doc.get(old_name, None):
                id_doc=doc['_metadata']['id_doc']

                url='/%s/doc/%d/%s' % (
                    base.metadata.name, 
                    id_doc, "/".join(path)
                )
                params={
                    'value': doc[old_name], 
                    'alter_files': False
                }
                request=DummyRequest(path=url, params=params)
                request.method='PUT'
                request.matchdict={
                    'base': base.metadata.name,
                    'id': str(id_doc),
                    'path': "/".join(path)
                }
                doc_view=DocumentCustomView(
                    DocumentContextFactory(request), 
                    request
                )
                response=doc_view.put_path()

def test_websocket_view_same_origin(config):
    # example.com is the dummy request default host URL
    config.registry.settings.update({'origins': []})
    req = DummyRequest(headers={'Origin': 'http://example.com'})
    req.get_response = lambda _: mock.sentinel.good_response
    res = views.websocket_view(req)
    assert res == mock.sentinel.good_response

 def test_leaders_weekly(self):
     session = self.db_master_session
     for i in range(3):
         user = User(nickname=unicode(i))
         session.add(user)
         session.flush()
         score1 = Score(userid=user.id, value=i)
         score1.name = 'new_cell'
         session.add(score1)
         score2 = Score(userid=user.id, value=i)
         score2.name = 'new_wifi'
         session.add(score2)
     session.commit()
     request = DummyRequest()
     request.db_slave_session = self.db_master_session
     inst = self._make_view(request)
     result = inst.leaders_weekly_view()
     for score_name in ('new_cell', 'new_wifi'):
         self.assertEqual(
             result['scores'][score_name]['leaders1'],
             [{'nickname': u'2', 'num': 2, 'pos': 1},
              {'nickname': u'1', 'num': 1, 'pos': 2}])
         self.assertEqual(
             result['scores'][score_name]['leaders2'],
             [{'nickname': u'0', 'num': 0, 'pos': 3}])

    def test_leaders(self):
        session = self.db_master_session
        today = util.utcnow().date()
        yesterday = today - timedelta(days=1)
        for i in range(3):
            user = User(nickname=unicode(i))
            session.add(user)
            session.flush()
            score1 = Score(userid=user.id, time=today, value=i)
            score1.name = 'location'
            session.add(score1)
            score2 = Score(userid=user.id, time=yesterday, value=i + 1)
            score2.name = 'location'
            session.add(score2)
        session.commit()
        request = DummyRequest()
        request.db_slave_session = self.db_master_session
        request.registry.redis_client = self.redis_client
        inst = self._make_view(request)
        result = inst.leaders_view()
        self.assertEqual(
            result['leaders1'],
            [{'anchor': u'2', 'nickname': u'2', 'num': 5, 'pos': 1},
             {'anchor': u'1', 'nickname': u'1', 'num': 3, 'pos': 2}])
        self.assertEqual(
            result['leaders2'],
            [{'anchor': u'0', 'nickname': u'0', 'num': 1, 'pos': 3}])

        # call the view again, without a working db session, so
        # we can be sure to use the cached result
        inst = self._make_view(request)
        request.db_slave_session = None
        second_result = inst.leaders_view()
        self.assertEqual(second_result, result)

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_leaders_weekly(self):
        session = self.db_master_session
        for i in range(3):
            user = User(nickname=unicode(i))
            session.add(user)
            session.flush()
            score1 = Score(userid=user.id, value=i)
            score1.name = 'new_cell'
            session.add(score1)
            score2 = Score(userid=user.id, value=i)
            score2.name = 'new_wifi'
            session.add(score2)
        session.commit()
        request = DummyRequest()
        request.db_slave_session = self.db_master_session
        request.registry.redis_client = self.redis_client
        inst = self._make_view(request)
        result = inst.leaders_weekly_view()
        for score_name in ('new_cell', 'new_wifi'):
            self.assertEqual(
                result['scores'][score_name]['leaders1'],
                [{'nickname': u'2', 'num': 2, 'pos': 1},
                 {'nickname': u'1', 'num': 1, 'pos': 2}])
            self.assertEqual(
                result['scores'][score_name]['leaders2'],
                [{'nickname': u'0', 'num': 0, 'pos': 3}])

        # call the view again, without a working db session, so
        # we can be sure to use the cached result
        inst = self._make_view(request)
        request.db_slave_session = None
        second_result = inst.leaders_weekly_view()
        self.assertEqual(second_result, result)

def main(global_config, **settings):
    """ This function returns a Pyramid WSGI application.
    """
    config = Configurator(root_factory=getAppRoot, settings=settings)
    config.add_static_view('static', 'static', cache_max_age=3600)

#    authnPolicy = BasicAuthAuthenticationPolicy(check=checkAuthentication,
#                                                realm='Djosr Login') #, debug=True)
    authnPolicy = AuthTktAuthenticationPolicy(secret='not telling',
                                              callback=getGroups)#, debug=True)
    authzPolicy = ACLAuthorizationPolicy()
    config.set_authentication_policy(authnPolicy)
    config.set_authorization_policy(authzPolicy)

    sessFactory = UnencryptedCookieSessionFactoryConfig('not telling')
    config.set_session_factory(sessFactory)
    config.add_request_method(getUser, 'user', reify=True)
    config.scan()

    # this saves having to do a GET to catch simple bugs
    # TODO remove later when code is bug-free ;-)
    from pyramid.testing import DummyRequest
    dummy = DummyRequest()
    dummy.registry = config.registry
    getAppRoot(dummy)

    return config.make_wsgi_app()

    def test_stats(self):
        day = util.utcnow().date() - timedelta(1)
        session = self.db_master_session
        stats = [
            Stat(key=STAT_TYPE['cell'], time=day, value=2000000),
            Stat(key=STAT_TYPE['wifi'], time=day, value=2000000),
            Stat(key=STAT_TYPE['unique_cell'], time=day, value=1000000),
            Stat(key=STAT_TYPE['unique_wifi'], time=day, value=2000000),
        ]
        session.add_all(stats)
        session.commit()
        request = DummyRequest()
        request.db_slave_session = self.db_master_session
        request.registry.redis_client = self.redis_client
        inst = self._make_view(request)
        result = inst.stats_view()
        self.assertEqual(result['page_title'], 'Statistics')
        self.assertEqual(
            result['metrics1'], [
                {'name': 'Unique Cells', 'value': '1.00'},
                {'name': 'Cell Observations', 'value': '2.00'},
            ])
        self.assertEqual(
            result['metrics2'], [
                {'name': 'Unique Wifi Networks', 'value': '2.00'},
                {'name': 'Wifi Observations', 'value': '2.00'},
            ])

        # call the view again, without a working db session, so
        # we can be sure to use the cached result
        inst = self._make_view(request)
        request.db_slave_session = None
        second_result = inst.stats_view()
        self.assertEqual(second_result, result)