<commit_before># vim: tabstop=4 shiftwidth=4 softtabstop=4

# Copyright 2011 OpenStack, LLC
# All Rights Reserved.
#
#    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.

"""
Tests a Glance API server which uses an S3 backend by default

This test requires that a real S3 account is available. It looks
in a file /tests/functional/test_s3.conf for the credentials to
use.

Note that this test clears the entire bucket from the S3 account
for use by the test case, so make sure you supply credentials for
test accounts only.

If a connection cannot be established, all the test cases are
skipped.
"""

import ConfigParser
import json
import os
import tempfile
import unittest

from tests import functional
from tests.utils import execute

FIVE_KB = 5 * 1024
FIVE_GB = 5 * 1024 * 1024 * 1024


class TestS3(functional.FunctionalTest):

    """Functional tests for the S3 backend"""

    # Test machines can set the GLANCE_TEST_MIGRATIONS_CONF variable
    # to override the location of the config file for migration testing
    CONFIG_FILE_PATH = os.environ.get('GLANCE_TEST_S3_CONF',
                                      os.path.join('tests', 'functional',
                                                   'test_s3.conf'))

    def setUp(self):
        """
        Test a connection to an S3 store using the credentials
        found in the environs or /tests/functional/test_s3.conf, if found.
        If the connection fails, mark all tests to skip.
        """
        self.inited = False
        self.skip_tests = True

        if self.inited:
            return

        if os.path.exists(TestS3.CONFIG_FILE_PATH):
            cp = ConfigParser.RawConfigParser()
            try:
                cp.read(TestS3.CONFIG_FILE_PATH)
                defaults = cp.defaults()
                for key, value in defaults.items():
                    self.__dict__[key] = value
            except ConfigParser.ParsingError, e:
                print ("Failed to read test_s3.conf config file. "
                       "Got error: %s" % e)
                super(TestS3, self).setUp()
                self.inited = True
                return

        from boto.s3.connection import S3Connection
        from boto.exception import S3ResponseError

        try:
            s3_host = self.s3_store_host
            access_key = self.s3_store_access_key
            secret_key = self.s3_store_secret_key
            bucket_name = self.s3_store_bucket
        except AttributeError, e:
            print ("Failed to find required configuration options for "
                   "S3 store. Got error: %s" % e)
            self.inited = True
            super(TestS3, self).setUp()
            return

        s3_conn = S3Connection(access_key, secret_key, host=s3_host)

        self.bucket = None
        try:
            buckets = s3_conn.get_all_buckets()
            for bucket in buckets:
                if bucket.name == bucket_name:
                    self.bucket = bucket
        except S3ResponseError, e:
            print ("Failed to connect to S3 with credentials,"
                   "to find bucket. Got error: %s" % e)
            self.inited = True
            super(TestS3, self).setUp()
            return

        self.s3_conn = s3_conn

        if not self.bucket:
            try:
                self.bucket = s3_conn.create_bucket(bucket_name)
            except boto.exception.S3ResponseError, e:
                print ("Failed to create bucket. Got error: %s" % e)
                self.inited = True
                super(TestS3, self).setUp()
                return
        else:
            self.clear_bucket()

        self.skip_tests = False
        self.inited = True
        self.default_store = 's3'
<commit_msg>Fix for boto1.9b issue 540 (http://code.google.com/p/boto/issues/detail?id=540)<commit_after># vim: tabstop=4 shiftwidth=4 softtabstop=4

# Copyright 2011 OpenStack, LLC
# All Rights Reserved.
#
#    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.

"""
Tests a Glance API server which uses an S3 backend by default

This test requires that a real S3 account is available. It looks
in a file /tests/functional/test_s3.conf for the credentials to
use.

Note that this test clears the entire bucket from the S3 account
for use by the test case, so make sure you supply credentials for
test accounts only.

If a connection cannot be established, all the test cases are
skipped.
"""

import ConfigParser
import json
import os
import tempfile
import unittest

from tests import functional
from tests.utils import execute

FIVE_KB = 5 * 1024
FIVE_GB = 5 * 1024 * 1024 * 1024


class TestS3(functional.FunctionalTest):

    """Functional tests for the S3 backend"""

    # Test machines can set the GLANCE_TEST_MIGRATIONS_CONF variable
    # to override the location of the config file for migration testing
    CONFIG_FILE_PATH = os.environ.get('GLANCE_TEST_S3_CONF',
                                      os.path.join('tests', 'functional',
                                                   'test_s3.conf'))

    def setUp(self):
        """
        Test a connection to an S3 store using the credentials
        found in the environs or /tests/functional/test_s3.conf, if found.
        If the connection fails, mark all tests to skip.
        """
        self.inited = False
        self.skip_tests = True

        if self.inited:
            return

        if os.path.exists(TestS3.CONFIG_FILE_PATH):
            cp = ConfigParser.RawConfigParser()
            try:
                cp.read(TestS3.CONFIG_FILE_PATH)
                defaults = cp.defaults()
                for key, value in defaults.items():
                    self.__dict__[key] = value
            except ConfigParser.ParsingError, e:
                print ("Failed to read test_s3.conf config file. "
                       "Got error: %s" % e)
                super(TestS3, self).setUp()
                self.inited = True
                return

        from boto.s3.connection import S3Connection
        from boto.exception import S3ResponseError

        try:
            s3_host = self.s3_store_host
            access_key = self.s3_store_access_key
            secret_key = self.s3_store_secret_key
            bucket_name = self.s3_store_bucket
        except AttributeError, e:
            print ("Failed to find required configuration options for "
                   "S3 store. Got error: %s" % e)
            self.inited = True
            super(TestS3, self).setUp()
            return

        s3_conn = S3Connection(access_key, secret_key, host=s3_host)

        self.bucket = None
        try:
            buckets = s3_conn.get_all_buckets()
            for bucket in buckets:
                if bucket.name == bucket_name:
                    self.bucket = bucket
        except S3ResponseError, e:
            print ("Failed to connect to S3 with credentials,"
                   "to find bucket. Got error: %s" % e)
            self.inited = True
            super(TestS3, self).setUp()
            return
        except TypeError, e:
            # This hack is necessary because of a bug in boto 1.9b:
            # http://code.google.com/p/boto/issues/detail?id=540
            print ("Failed to connect to S3 with credentials. "
                   "Got error: %s" % e)
            self.inited = True
            super(TestS3, self).setUp()
            return

        self.s3_conn = s3_conn

        if not self.bucket:
            try:
                self.bucket = s3_conn.create_bucket(bucket_name)
            except boto.exception.S3ResponseError, e:
                print ("Failed to create bucket. Got error: %s" % e)
                self.inited = True
                super(TestS3, self).setUp()
                return
        else:
            self.clear_bucket()

        self.skip_tests = False
        self.inited = True
        self.default_store = 's3'

{'fraction_non_alphanumeric': 0.07086878638602777, 'fraction_numerical': 0.016457680250783698, 'mean_word_length': 1.9202353710362863, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6617369', 'n_tokens_mistral': 2520, 'n_tokens_neox': 2388, 'n_words': 845}