Adding fake_http_connect to test.utils.

- Copy fake_http_connect function from swift repository.

tests/test_swiftclient.py

@@ -19,7 +19,7 @@ import unittest
 from urlparse import urlparse
 
 # TODO: mock http connection class with more control over headers
-from test.unit.proxy.test_server import fake_http_connect
+from utils import fake_http_connect
 
 from swiftclient import client as c
 

tests/utils.py

@@ -0,0 +1,115 @@
+# Copyright (c) 2010-2012 OpenStack, LLC.
+#
+# 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.
+from httplib import HTTPException
+
+from eventlet import Timeout, sleep
+
+
+def fake_http_connect(*code_iter, **kwargs):
+
+    class FakeConn(object):
+
+        def __init__(self, status, etag=None, body='', timestamp='1'):
+            self.status = status
+            self.reason = 'Fake'
+            self.host = '1.2.3.4'
+            self.port = '1234'
+            self.sent = 0
+            self.received = 0
+            self.etag = etag
+            self.body = body
+            self.timestamp = timestamp
+
+        def getresponse(self):
+            if kwargs.get('raise_exc'):
+                raise Exception('test')
+            if kwargs.get('raise_timeout_exc'):
+                raise Timeout()
+            return self
+
+        def getexpect(self):
+            if self.status == -2:
+                raise HTTPException()
+            if self.status == -3:
+                return FakeConn(507)
+            return FakeConn(100)
+
+        def getheaders(self):
+            headers = {'content-length': len(self.body),
+                       'content-type': 'x-application/test',
+                       'x-timestamp': self.timestamp,
+                       'last-modified': self.timestamp,
+                       'x-object-meta-test': 'testing',
+                       'etag':
+                            self.etag or '"68b329da9893e34099c7d8ad5cb9c940"',
+                       'x-works': 'yes',
+                       'x-account-container-count': 12345}
+            if not self.timestamp:
+                del headers['x-timestamp']
+            try:
+                if container_ts_iter.next() is False:
+                    headers['x-container-timestamp'] = '1'
+            except StopIteration:
+                pass
+            if 'slow' in kwargs:
+                headers['content-length'] = '4'
+            if 'headers' in kwargs:
+                headers.update(kwargs['headers'])
+            return headers.items()
+
+        def read(self, amt=None):
+            if 'slow' in kwargs:
+                if self.sent < 4:
+                    self.sent += 1
+                    sleep(0.1)
+                    return ' '
+            rv = self.body[:amt]
+            self.body = self.body[amt:]
+            return rv
+
+        def send(self, amt=None):
+            if 'slow' in kwargs:
+                if self.received < 4:
+                    self.received += 1
+                    sleep(0.1)
+
+        def getheader(self, name, default=None):
+            return dict(self.getheaders()).get(name.lower(), default)
+
+    timestamps_iter = iter(kwargs.get('timestamps') or ['1'] * len(code_iter))
+    etag_iter = iter(kwargs.get('etags') or [None] * len(code_iter))
+    x = kwargs.get('missing_container', [False] * len(code_iter))
+    if not isinstance(x, (tuple, list)):
+        x = [x] * len(code_iter)
+    container_ts_iter = iter(x)
+    code_iter = iter(code_iter)
+
+    def connect(*args, **ckwargs):
+        if 'give_content_type' in kwargs:
+            if len(args) >= 7 and 'Content-Type' in args[6]:
+                kwargs['give_content_type'](args[6]['Content-Type'])
+            else:
+                kwargs['give_content_type']('')
+        if 'give_connect' in kwargs:
+            kwargs['give_connect'](*args, **ckwargs)
+        status = code_iter.next()
+        etag = etag_iter.next()
+        timestamp = timestamps_iter.next()
+        if status <= 0:
+            raise HTTPException()
+        return FakeConn(status, etag, body=kwargs.get('body', ''),
+                        timestamp=timestamp)
+
+    return connect

tools/test-requires

@@ -6,3 +6,4 @@ nosexcover
 openstack.nose_plugin
 pep8>=1.0
 sphinx>=1.1.2
+eventlet