¡@
OpenStack Study: store_utils.py**** CubicPower OpenStack Study ****
# Copyright 2011 OpenStack Foundation # Copyright 2012 Red Hat, Inc # 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. """ Utility methods to set testcases up for Swift and/or S3 tests. """ from __future__ import print_function import BaseHTTPServer import thread from glance.openstack.common import units FIVE_KB = 5 * units.Ki
**** CubicPower OpenStack Study **** class RemoteImageHandler(BaseHTTPServer.BaseHTTPRequestHandler):
**** CubicPower OpenStack Study **** def do_HEAD(self): """ Respond to an image HEAD request fake metadata """ if 'images' in self.path: self.send_response(200) self.send_header('Content-Type', 'application/octet-stream') self.send_header('Content-Length', FIVE_KB) self.end_headers() return else: self.send_error(404, 'File Not Found: %s' % self.path) return
**** CubicPower OpenStack Study **** def do_GET(self): """ Respond to an image GET request with fake image content. """ if 'images' in self.path: self.send_response(200) self.send_header('Content-Type', 'application/octet-stream') self.send_header('Content-Length', FIVE_KB) self.end_headers() image_data = '*' * FIVE_KB self.wfile.write(image_data) self.wfile.close() return else: self.send_error(404, 'File Not Found: %s' % self.path) return
**** CubicPower OpenStack Study **** def log_message(self, format, *args): """ Simple override to prevent writing crap to stderr... """ pass def setup_http(test): server_class = BaseHTTPServer.HTTPServer remote_server = server_class(('127.0.0.1', 0), RemoteImageHandler) remote_ip, remote_port = remote_server.server_address
**** CubicPower OpenStack Study **** def setup_http(test): server_class = BaseHTTPServer.HTTPServer remote_server = server_class(('127.0.0.1', 0), RemoteImageHandler) remote_ip, remote_port = remote_server.server_address
**** CubicPower OpenStack Study **** def serve_requests(httpd): httpd.serve_forever() thread.start_new_thread(serve_requests, (remote_server,)) test.http_server = remote_server test.http_ip = remote_ip test.http_port = remote_port test.addCleanup(test.http_server.shutdown) def get_http_uri(test, image_id): uri = 'http://%(http_ip)s:%(http_port)d/images/' % test.__dict__ uri += image_id return uri
**** CubicPower OpenStack Study **** def get_http_uri(test, image_id): uri = 'http://%(http_ip)s:%(http_port)d/images/' % test.__dict__ uri += image_id return uri
|