dkirillov/s3-tests
1
0
Fork 0
forked from TrueCloudLab/s3-tests
Code Pull requests Activity

S3 Fuzzer: Change how random data works

Browse source 
Remove SpecialVariables dict subclass in favor of RepeatExpandingFormatter
string.Formatter subclass.
This commit is contained in:
Kyle Marsh 2011-08-18 12:34:56 -07:00
parent bfca00ac4c
commit e12f124686
2 changed files with 69 additions and 83 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

81
s3tests/functional/test_fuzzer.py
View file

@ -140,60 +140,50 @@ def test_descend_nonexistant_child():
assert_raises(KeyError, descend_graph, graph, 'nonexistant_child_node', prng)
def test_SpecialVariables_dict():
def test_expand_random_printable():
prng = random.Random(1)
testdict = {'foo': 'bar'}
tester = SpecialVariables(testdict, prng)
eq(tester['foo'], 'bar')
eq(tester['random 10-15 printable'], '[/pNI$;92@')
got = expand({}, '{random 10-15 printable}', prng)
eq(got, '[/pNI$;92@')
def test_SpecialVariables_binary():
def test_expand_random_binary():
prng = random.Random(1)
tester = SpecialVariables({}, prng)
eq(tester['random 10-15 binary'], '\xdfj\xf1\xd80>a\xcd\xc4\xbb')
got = expand({}, '{random 10-15 binary}', prng)
eq(got, '\xdfj\xf1\xd80>a\xcd\xc4\xbb')
def test_SpeicalVariables_random_no_args():
def test_expand_random_no_args():
prng = random.Random(1)
tester = SpecialVariables({}, prng)
for _ in xrange(1000):
val = tester['random']
val = val.replace('{{', '{').replace('}}','}')
assert_true(0 <= len(val) <= 1000)
assert_true(reduce(lambda x, y: x and y, [x in string.printable for x in val]))
got = expand({}, '{random}', prng)
assert_true(0 <= len(got) <= 1000)
assert_true(reduce(lambda x, y: x and y, [x in string.printable for x in got]))
def test_SpeicalVariables_random_no_charset():
def test_expand_random_no_charset():
prng = random.Random(1)
tester = SpecialVariables({}, prng)
for _ in xrange(1000):
val = tester['random 10-30']
val = val.replace('{{', '{').replace('}}','}')
assert_true(10 <= len(val) <= 30)
assert_true(reduce(lambda x, y: x and y, [x in string.printable for x in val]))
got = expand({}, '{random 10-30}', prng)
assert_true(10 <= len(got) <= 30)
assert_true(reduce(lambda x, y: x and y, [x in string.printable for x in got]))
def test_SpeicalVariables_random_exact_length():
def test_expand_random_exact_length():
prng = random.Random(1)
tester = SpecialVariables({}, prng)
for _ in xrange(1000):
val = tester['random 10 digits']
assert_true(len(val) == 10)
assert_true(reduce(lambda x, y: x and y, [x in string.digits for x in val]))
got = expand({}, '{random 10 digits}', prng)
assert_true(len(got) == 10)
assert_true(reduce(lambda x, y: x and y, [x in string.digits for x in got]))
def test_SpecialVariables_random_errors():
def test_expand_random_bad_charset():
prng = random.Random(1)
tester = SpecialVariables({}, prng)
assert_raises(KeyError, expand, {}, '{random 10-30 foo}', prng)
assert_raises(KeyError, lambda x: tester[x], 'random 10-30 foo')
assert_raises(ValueError, lambda x: tester[x], 'random printable')
def test_expand_random_missing_length():
prng = random.Random(1)
assert_raises(ValueError, expand, {}, '{random printable}', prng)
def test_assemble_decision():
@ -210,54 +200,60 @@ def test_assemble_decision():
def test_expand_escape():
prng = random.Random(1)
decision = dict(
foo='{{bar}}',
)
got = expand(decision, '{foo}')
got = expand(decision, '{foo}', prng)
eq(got, '{bar}')
def test_expand_indirect():
prng = random.Random(1)
decision = dict(
foo='{bar}',
bar='quux',
)
got = expand(decision, '{foo}')
got = expand(decision, '{foo}', prng)
eq(got, 'quux')
def test_expand_indirect_double():
prng = random.Random(1)
decision = dict(
foo='{bar}',
bar='{quux}',
quux='thud',
)
got = expand(decision, '{foo}')
got = expand(decision, '{foo}', prng)
eq(got, 'thud')
def test_expand_recursive():
prng = random.Random(1)
decision = dict(
foo='{foo}',
)
e = assert_raises(RecursionError, expand, decision, '{foo}')
e = assert_raises(RecursionError, expand, decision, '{foo}', prng)
eq(str(e), "Runaway recursion in string formatting: 'foo'")
def test_expand_recursive_mutual():
prng = random.Random(1)
decision = dict(
foo='{bar}',
bar='{foo}',
)
e = assert_raises(RecursionError, expand, decision, '{foo}')
e = assert_raises(RecursionError, expand, decision, '{foo}', prng)
eq(str(e), "Runaway recursion in string formatting: 'foo'")
def test_expand_recursive_not_too_eager():
prng = random.Random(1)
decision = dict(
foo='bar',
)
got = expand(decision, 100*'{foo}')
got = expand(decision, 100*'{foo}', prng)
eq(got, 100*'bar')
@ -356,15 +352,14 @@ def test_expand_headers():
graph = build_graph()
prng = random.Random(1)
decision = descend_graph(graph, 'node1', prng)
special_decision = SpecialVariables(decision, prng)
expanded_headers = expand_headers(special_decision)
expanded_headers = expand_headers(decision, prng)
for header, value in expanded_headers:
if header == 'my-header':
assert_true(value in ['h1', 'h2', 'h3'])
elif header.startswith('random-header-'):
assert_true(20 <= len(value) <= 30)
assert_true(string.strip(value, SpecialVariables.charsets['punctuation']) is '')
assert_true(string.strip(value, RepeatExpandingFormatter.charsets['punctuation']) is '')
else:
raise DecisionGraphError('unexpected header found: "%s"' % header)

71
s3tests/fuzz_headers.py
View file

@ -117,42 +117,23 @@ def make_choice(choices, prng):
return prng.choice(weighted_choices)
def expand_headers(decision):
def expand_headers(decision, prng):
expanded_headers = []
for header in decision['headers']:
h = expand(decision, header[0])
v = expand(decision, header[1])
h = expand(decision, header[0], prng)
v = expand(decision, header[1], prng)
expanded_headers.append([h, v])
return expanded_headers
def expand(decision, value):
def expand(decision, value, prng):
c = itertools.count()
fmt = RepeatExpandingFormatter()
fmt = RepeatExpandingFormatter(prng)
new = fmt.vformat(value, [], decision)
return new
class RepeatExpandingFormatter(string.Formatter):
def __init__(self, _recursion=0):
super(RepeatExpandingFormatter, self).__init__()
# this class assumes it is always instantiated once per
# formatting; use that to detect runaway recursion
self._recursion = _recursion
def get_value(self, key, args, kwargs):
val = super(RepeatExpandingFormatter, self).get_value(key, args, kwargs)
if self._recursion > 5:
raise RecursionError(key)
fmt = self.__class__(_recursion=self._recursion+1)
# must use vformat not **kwargs so our SpecialVariables is not
# downgraded to just a dict
n = fmt.vformat(val, args, kwargs)
return n
class SpecialVariables(dict):
charsets = {
'printable': string.printable,
'punctuation': string.punctuation,
@ -160,21 +141,29 @@ class SpecialVariables(dict):
'digits': string.digits
}
def __init__(self, orig_dict, prng):
super(SpecialVariables, self).__init__(orig_dict)
def __init__(self, prng, _recursion=0):
super(RepeatExpandingFormatter, self).__init__()
# this class assumes it is always instantiated once per
# formatting; use that to detect runaway recursion
self.prng = prng
self._recursion = _recursion
def __getitem__(self, key):
def get_value(self, key, args, kwargs):
fields = key.split(None, 1)
fn = getattr(self, 'special_{name}'.format(name=fields[0]), None)
if fn is None:
return super(SpecialVariables, self).__getitem__(key)
if len(fields) == 1:
fields.append('')
return fn(fields[1])
if fn is not None:
if len(fields) == 1:
fields.append('')
return fn(fields[1])
val = super(RepeatExpandingFormatter, self).get_value(key, args, kwargs)
if self._recursion > 5:
raise RecursionError(key)
fmt = self.__class__(self.prng, _recursion=self._recursion+1)
# must use vformat not **kwargs so our SpecialVariables is not
# downgraded to just a dict
n = fmt.vformat(val, args, kwargs)
return n
def special_random(self, args):
arg_list = args.split()
@ -199,12 +188,10 @@ class SpecialVariables(dict):
num_bytes = length + 8
tmplist = [self.prng.getrandbits(64) for _ in xrange(num_bytes / 8)]
tmpstring = struct.pack((num_bytes / 8) * 'Q', *tmplist)
tmpstring = tmpstring[0:length]
return tmpstring[0:length]
else:
charset = self.charsets[charset_arg]
tmpstring = ''.join([self.prng.choice(charset) for _ in xrange(length)]) # Won't scale nicely
return tmpstring.replace('{', '{{').replace('}', '}}')
return ''.join([self.prng.choice(charset) for _ in xrange(length)]) # Won't scale nicely
def parse_options():
@ -262,9 +249,13 @@ def _main():
prng = random.Random(request_seed)
decision = assemble_decision(decision_graph, prng)
decision.update(constants)
request = expand_decision(decision, prng)
response = s3_connection.make_request(request['method'], request['path'], data=request['body'], headers=request['headers'], override_num_retries=0)
method = expand(decision, decision['method'], prng)
path = expand(decision, decision['path'], prng)
body = expand(decision, decision['body'], prng)
headers = expand_headers(decision, prng)
response = s3_connection.make_request(method, path, data=body, headers=headers, override_num_retries=0)
if response.status == 500 or response.status == 503:
print 'Request generated with seed %d failed:\n%s' % (request_seed, request)