151 lines
3.4 KiB
Python
Raw Normal View History

2010-07-12 17:03:45 -05:00
""" Swift tests """
import os
from contextlib import contextmanager
from tempfile import NamedTemporaryFile
2010-07-12 17:03:45 -05:00
from eventlet.green import socket
from tempfile import mkdtemp
from shutil import rmtree
2010-07-12 17:03:45 -05:00
def readuntil2crlfs(fd):
rv = ''
lc = ''
crlfs = 0
while crlfs < 2:
c = fd.read(1)
rv = rv + c
if c == '\r' and lc != '\n':
crlfs = 0
if lc == '\r' and c == '\n':
crlfs += 1
lc = c
return rv
def connect_tcp(hostport):
rv = socket.socket()
rv.connect(hostport)
return rv
@contextmanager
def tmpfile(content):
with NamedTemporaryFile('w', delete=False) as f:
file_name = f.name
f.write(str(content))
try:
yield file_name
finally:
os.unlink(file_name)
2011-01-19 14:18:37 -06:00
xattr_data = {}
2011-01-19 16:19:43 -06:00
2011-01-19 14:18:37 -06:00
def _get_inode(fd):
if not isinstance(fd, int):
try:
fd = fd.fileno()
except AttributeError:
return os.stat(fd).st_ino
return os.fstat(fd).st_ino
2011-01-19 16:19:43 -06:00
2011-01-19 14:18:37 -06:00
def _setxattr(fd, k, v):
inode = _get_inode(fd)
data = xattr_data.get(inode, {})
data[k] = v
xattr_data[inode] = data
2011-01-19 16:19:43 -06:00
2011-01-19 14:18:37 -06:00
def _getxattr(fd, k):
inode = _get_inode(fd)
data = xattr_data.get(inode, {}).get(k)
if not data:
raise IOError
return data
import xattr
xattr.setxattr = _setxattr
xattr.getxattr = _getxattr
@contextmanager
def temptree(files, contents=''):
# generate enough contents to fill the files
c = len(files)
contents = (list(contents) + [''] * c)[:c]
tempdir = mkdtemp()
for path, content in zip(files, contents):
if os.path.isabs(path):
path = '.' + path
new_path = os.path.join(tempdir, path)
subdir = os.path.dirname(new_path)
if not os.path.exists(subdir):
os.makedirs(subdir)
with open(new_path, 'w') as f:
f.write(str(content))
try:
yield tempdir
finally:
rmtree(tempdir)
2011-03-15 22:12:03 -07:00
class FakeLogger(object):
# a thread safe logger
2011-05-10 15:36:01 -07:00
def __init__(self, *args, **kwargs):
self.log_dict = dict(error=[], info=[], warning=[], debug=[])
2011-03-15 22:12:03 -07:00
def error(self, *args, **kwargs):
self.log_dict['error'].append((args, kwargs))
def info(self, *args, **kwargs):
self.log_dict['info'].append((args, kwargs))
def warning(self, *args, **kwargs):
self.log_dict['warning'].append((args, kwargs))
2011-05-10 15:36:01 -07:00
def debug(self, *args, **kwargs):
self.log_dict['debug'].append((args, kwargs))
2011-03-15 22:12:03 -07:00
class MockTrue(object):
"""
2010-07-29 13:31:27 -05:00
Instances of MockTrue evaluate like True
Any attr accessed on an instance of MockTrue will return a MockTrue instance
Any method called on an instance of MockTrue will return a MockTrue instance
>>> thing = MockTrue()
>>> thing
True
>>> thing == True # True == True
True
>>> thing == False # True == False
False
>>> thing != True # True != True
False
>>> thing != False # True != False
True
>>> thing.attribute
True
>>> thing.method()
True
>>> thing.attribute.method()
True
>>> thing.method().attribute
True
"""
def __getattribute__(self, *args, **kwargs):
return self
def __call__(self, *args, **kwargs):
return self
def __repr__(*args, **kwargs):
return repr(True)
def __eq__(self, other):
return other is True
def __ne__(self, other):
return other is not True