<commit_before># -*- encoding: utf-8 -*-
#
# Copyright © 2014-2015 eNovance
#
# Authors: <NAME> <<EMAIL>>
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import calendar
import contextlib
import datetime
from email import utils as email_utils
import hashlib
import json
import uuid

import mock
from oslo_utils import timeutils
import pecan
import six
from stevedore import extension
import testscenarios
from testtools import testcase
import webtest

from gnocchi import archive_policy
from gnocchi.rest import app
from gnocchi import storage
from gnocchi.tests import base as tests_base
from gnocchi import utils


load_tests = testscenarios.load_tests_apply_scenarios


class FakeMemcache(object):
    VALID_TOKEN_ADMIN = '4<PASSWORD>'
    USER_ID_ADMIN = str(uuid.uuid4())
    PROJECT_ID_ADMIN = str(uuid.uuid4())

    VALID_TOKEN = '45<PASSWORD>'
    USER_ID = str(uuid.uuid4())
    PROJECT_ID = str(uuid.uuid4())

    VALID_TOKEN_2 = '4562<PASSWORD>'
    # We replace "-" to simulate a middleware that would send UUID in a non
    # normalized format.
    USER_ID_2 = str(uuid.uuid4()).replace("-", "")
    PROJECT_ID_2 = str(uuid.uuid4()).replace("-", "")

    def get(self, key):
        dt = "2100-01-01T23:59:59"
        if key == "tokens/%s" % self.VALID_TOKEN_ADMIN:
            return json.dumps(({'access': {
                'token': {'id': self.VALID_TOKEN_ADMIN,
                          'expires': dt},
                'user': {
                    'id': self.USER_ID_ADMIN,
                    'name': 'adminusername',
                    'tenantId': self.PROJECT_ID_ADMIN,
                    'tenantName': 'myadmintenant',
                    'roles': [
                        {'name': 'admin'},
                    ]},
            }}, dt))
        elif key == "tokens/%s" % self.VALID_TOKEN:
            return json.dumps(({'access': {
                'token': {'id': self.VALID_TOKEN,
                          'expires': dt},
                'user': {
                    'id': self.USER_ID,
                    'name': 'myusername',
                    'tenantId': self.PROJECT_ID,
                    'tenantName': 'mytenant',
                    'roles': [
                        {'name': 'member'},
                    ]},
            }}, dt))
        elif key == "tokens/%s" % self.VALID_TOKEN_2:
            return json.dumps(({'access': {
                'token': {'id': self.VALID_TOKEN_2,
                          'expires': dt},
                'user': {
                    'id': self.USER_ID_2,
                    'name': 'myusername2',
                    'tenantId': self.PROJECT_ID_2,
                    'tenantName': 'mytenant2',
                    'roles': [
                        {'name': 'member'},
                    ]},
            }}, dt))

    @staticmethod
    def set(key, value, **kwargs):
        pass


class TestingApp(webtest.TestApp):
    CACHE_NAME = 'fake.cache'

    def __init__(self, *args, **kwargs):
        self.auth = kwargs.pop('auth')
        self.storage = kwargs.pop('storage')
        self.indexer = kwargs.pop('indexer')
        super(TestingApp, self).__init__(*args, **kwargs)
        # Setup Keystone auth_token fake cache
        self.extra_environ.update({self.CACHE_NAME: FakeMemcache()})
        self.token = FakeMemcache.VALID_TOKEN

    @contextlib.contextmanager
    def use_admin_user(self):
        if not self.auth:
            raise testcase.TestSkipped("No auth enabled")
        old_token = self.token
        self.token = FakeMemcache.VALID_TOKEN_ADMIN
        try:
            yield
        finally:
            self.token = old_token

    @contextlib.contextmanager
    def use_another_user(self):
        if not self.auth:
            raise testcase.TestSkipped("No auth enabled")
        old_token = self.token
        self.token = FakeMemcache.VALID_TOKEN_2
        try:
            yield
        finally:
            self.token = old_token

    def do_request(self, req, *args, **kwargs):
        req.headers['X-Auth-Token'] = self.token
        self.storage.process_measures(self.indexer)
        return super(TestingApp, self).do_request(req, *args, **kwargs)


class RestTest(tests_base.TestCase, testscenarios.TestWithScenarios):

    scenarios = [
        ('noauth', dict(middlewares=[])),
        ('keystone', dict(
            middlewares=['keystonemiddleware.auth_token.AuthProtocol'])),
    ]

    def setUp(self):
        super(RestTest, self).setUp()
        c = {}
        c.update(app.PECAN_CONFIG)
        c['indexer'] = self.index
        c['storage'] = self.storage
        c['not_implemented_middleware'] = False
        self.conf.set_override("cache", TestingApp.CACHE_NAME,
                               group='keystone_authtoken')
        # TODO(jd) Override these options with values. They are not used, but
        # if they are None (their defaults), the keystone authtoken middleware
        # prints a warning… :( When the bug is fixed we can remove that!
        # See https://bugs.launchpad.net/keystonemiddleware/+bug/1429179
        self.conf.set_override("identity_uri", "foobar",
                               group="keystone_authtoken")
        self.conf.set_override("auth_uri", "foobar",
                               group="keystone_authtoken")

        if hasattr(self, "middlewares"):
            self.conf.set_override("middlewares",
                                   self.middlewares, group="api")

        self.app = TestingApp(pecan.load_app(c, cfg=self.conf),
                              storage=self.storage,
                              indexer=self.index,
                              auth=bool(self.conf.api.middlewares))

    def test_deserialize_force_json(self):
        with self.app.use_admin_user():
            self.app.post(
                "/v1/archive_policy",
                params="foo",
                status=415)

    def test_capabilities(self):
        custom_agg = extension.Extension('test_aggregation', None, None, None)<commit_msg>Make test_rest keystone cache work with keystonemiddleware 2.0<commit_after># -*- encoding: utf-8 -*-
#
# Copyright © 2014-2015 eNovance
#
# Authors: <NAME> <<EMAIL>>
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import calendar
import contextlib
import datetime
from email import utils as email_utils
import hashlib
import json
import uuid

import mock
from oslo_utils import timeutils
import pecan
import six
from stevedore import extension
import testscenarios
from testtools import testcase
import webtest

from gnocchi import archive_policy
from gnocchi.rest import app
from gnocchi import storage
from gnocchi.tests import base as tests_base
from gnocchi import utils


load_tests = testscenarios.load_tests_apply_scenarios


class FakeMemcache(object):
    VALID_TOKEN_ADMIN = '<PASSWORD>'
    ADMIN_TOKEN_HASH = hashlib.sha256(
        VALID_TOKEN_ADMIN.encode('utf-8')).hexdigest()
    USER_ID_ADMIN = str(uuid.uuid4())
    PROJECT_ID_ADMIN = str(uuid.uuid4())

    VALID_TOKEN = '<PASSWORD>'
    TOKEN_HASH = hashlib.sha256(VALID_TOKEN.encode('utf-8')).hexdigest()
    USER_ID = str(uuid.uuid4())
    PROJECT_ID = str(uuid.uuid4())

    VALID_TOKEN_2 = '<PASSWORD>'
    TOKEN_2_HASH = hashlib.sha256(VALID_TOKEN_2.encode('utf-8')).hexdigest()
    # We replace "-" to simulate a middleware that would send UUID in a non
    # normalized format.
    USER_ID_2 = str(uuid.uuid4()).replace("-", "")
    PROJECT_ID_2 = str(uuid.uuid4()).replace("-", "")

    def get(self, key):
        dt = "2100-01-01T23:59:59"
        if (key == "tokens/%s" % self.ADMIN_TOKEN_HASH or
                key == "tokens/%s" % self.VALID_TOKEN_ADMIN):
            return json.dumps(({'access': {
                'token': {'id': self.VALID_TOKEN_ADMIN,
                          'expires': dt},
                'user': {
                    'id': self.USER_ID_ADMIN,
                    'name': 'adminusername',
                    'tenantId': self.PROJECT_ID_ADMIN,
                    'tenantName': 'myadmintenant',
                    'roles': [
                        {'name': 'admin'},
                    ]},
            }}, dt))
        elif (key == "tokens/%s" % self.TOKEN_HASH or
                key == "tokens/%s" % self.VALID_TOKEN):
            return json.dumps(({'access': {
                'token': {'id': self.VALID_TOKEN,
                          'expires': dt},
                'user': {
                    'id': self.USER_ID,
                    'name': 'myusername',
                    'tenantId': self.PROJECT_ID,
                    'tenantName': 'mytenant',
                    'roles': [
                        {'name': 'member'},
                    ]},
            }}, dt))
        elif (key == "tokens/%s" % self.TOKEN_2_HASH or
              key == "tokens/%s" % self.VALID_TOKEN_2):
            return json.dumps(({'access': {
                'token': {'id': self.VALID_TOKEN_2,
                          'expires': dt},
                'user': {
                    'id': self.USER_ID_2,
                    'name': 'myusername2',
                    'tenantId': self.PROJECT_ID_2,
                    'tenantName': 'mytenant2',
                    'roles': [
                        {'name': 'member'},
                    ]},
            }}, dt))

    @staticmethod
    def set(key, value, **kwargs):
        pass


class TestingApp(webtest.TestApp):
    CACHE_NAME = 'fake.cache'

    def __init__(self, *args, **kwargs):
        self.auth = kwargs.pop('auth')
        self.storage = kwargs.pop('storage')
        self.indexer = kwargs.pop('indexer')
        super(TestingApp, self).__init__(*args, **kwargs)
        # Setup Keystone auth_token fake cache
        self.extra_environ.update({self.CACHE_NAME: FakeMemcache()})
        self.token = FakeMemcache.VALID_TOKEN

    @contextlib.contextmanager
    def use_admin_user(self):
        if not self.auth:
            raise testcase.TestSkipped("No auth enabled")
        old_token = self.token
        self.token = FakeMemcache.VALID_TOKEN_ADMIN
        try:
            yield
        finally:
            self.token = old_token

    @contextlib.contextmanager
    def use_another_user(self):
        if not self.auth:
            raise testcase.TestSkipped("No auth enabled")
        old_token = self.token
        self.token = FakeMemcache.VALID_TOKEN_2
        try:
            yield
        finally:
            self.token = old_token

    def do_request(self, req, *args, **kwargs):
        req.headers['X-Auth-Token'] = self.token
        self.storage.process_measures(self.indexer)
        return super(TestingApp, self).do_request(req, *args, **kwargs)


class RestTest(tests_base.TestCase, testscenarios.TestWithScenarios):

    scenarios = [
        ('noauth', dict(middlewares=[])),
        ('keystone', dict(
            middlewares=['keystonemiddleware.auth_token.AuthProtocol'])),
    ]

    def setUp(self):
        super(RestTest, self).setUp()
        c = {}
        c.update(app.PECAN_CONFIG)
        c['indexer'] = self.index
        c['storage'] = self.storage
        c['not_implemented_middleware'] = False
        self.conf.set_override("cache", TestingApp.CACHE_NAME,
                               group='keystone_authtoken')
        # TODO(jd) Override these options with values. They are not used, but
        # if they are None (their defaults), the keystone authtoken middleware
        # prints a warning… :( When the bug is fixed we can remove that!
        # See https://bugs.launchpad.net/keystonemiddleware/+bug/1429179
        self.conf.set_override("identity_uri", "foobar",
                               group="keystone_authtoken")
        self.conf.set_override("auth_uri", "foobar",
                               group="keystone_authtoken")

        if hasattr(self, "middlewares"):
            self.conf.set_override("middlewares",
                                   self.middlewares, group="api")

        # TODO(chdent) Linting is turned off until a
        # keystonemiddleware bug is resolved.
        # See: https://bugs.launchpad.net/keystonemiddleware/+bug/1466499
        self.app = TestingApp(pecan.load_app(c, cfg=self.conf),
                              storage=self.storage,
                              indexer=self.index,
                              auth=bool(self.conf.api.middlewares),
                              lint=False)

    def test_deserialize_force_json(self):
        with self.app.use_admin_user():
            self.app.post(
                "/v1/archive_policy",
                params="foo",
                status=415)

    def test_capabilities(self):
        custom_agg = extension.Extension('test_aggregation', None, None, None)
{'fraction_non_alphanumeric': 0.11374753595677886, 'fraction_numerical': 0.010148207636708768, 'mean_word_length': 1.9921363040629096, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 15, 'https://': 3, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5186100', 'n_tokens_mistral': 4102, 'n_tokens_neox': 3934, 'n_words': 917}