gistfile1.txt

#!/usr/bin/python -u
# Copyright (c) 2010-2012 OpenStack Foundation
#
# 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 unittest
import time

from swiftclient import client

from swift.common.manager import Manager
from swift.obj import diskfile
from swift.common import exceptions, utils

from test.unit import DebugLogger
from test.probe.common import reset_environment, get_server_number, \
    get_to_final_state


class TestPutIfNoneMatch(unittest.TestCase):

    def setUp(self):
        (self.pids, self.port2server, self.account_ring, self.container_ring,
         self.object_ring, self.url, self.token,
         self.account, self.configs) = reset_environment()

    def tearDown(self):
        Manager(['all']).stop()

    def get_config(self, node):
        server, number = get_server_number(node['port'], self.port2server)
        return utils.readconf(self.configs['object-server'][number],
                              section_name='object-replicator')

    def test_put_if_none_match_failure(self):
        # create a test container
        container = 'put-if-none-match-failures'
        client.put_container(self.url, self.token, container)
        # fail the first half of container servers
        object_server = Manager(['object'])
        first_half, second_half = (1, 3), (2, 4)
        for number in first_half:
            object_server.stop(number=number, wait=True)
        # create an object with first half down
        client.put_object(self.url, self.token, container, 'obj', 'VERIFY')
        # start the first half of servers back up
        for number in first_half:
            object_server.start(number=number, wait=True)
        # verify x-if-none-match
        try:
            client.put_object(self.url, self.token, container, 'obj', 'VERIFY',
                              headers={'if-none-match': '*'})
        except client.ClientException as e:
            if e.http_status != 412:
                raise
        else:
            self.fail('put with if-none-match to pre-existing object '
                      'did not return status 412!')
        # shut the second half of servers down
        for number in second_half:
            object_server.stop(number=number, wait=True)
        try:
            client.put_object(self.url, self.token, container, 'obj', 'VERIFY',
                              headers={'x-if-none-match': '*'})
        except client.ClientException as e:
            if e.http_status != 412:
                raise
        else:
            self.fail('put with if-none-match to pre-existing object '
                      'did not return status 412!')


if __name__ == "__main__":
    unittest.main()