本文整理汇总了Python中test.unit.temptree函数的典型用法代码示例。如果您正苦于以下问题:Python temptree函数的具体用法?Python temptree怎么用?Python temptree使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了temptree函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_search_tree
def test_search_tree(self):
# file match & ext miss
with temptree(["asdf.conf", "blarg.conf", "asdf.cfg"]) as t:
asdf = utils.search_tree(t, "a*", ".conf")
self.assertEquals(len(asdf), 1)
self.assertEquals(asdf[0], os.path.join(t, "asdf.conf"))
# multi-file match & glob miss & sort
with temptree(["application.bin", "apple.bin", "apropos.bin"]) as t:
app_bins = utils.search_tree(t, "app*", "bin")
self.assertEquals(len(app_bins), 2)
self.assertEquals(app_bins[0], os.path.join(t, "apple.bin"))
self.assertEquals(app_bins[1], os.path.join(t, "application.bin"))
# test file in folder & ext miss & glob miss
files = ("sub/file1.ini", "sub/file2.conf", "sub.bin", "bus.ini", "bus/file3.ini")
with temptree(files) as t:
sub_ini = utils.search_tree(t, "sub*", ".ini")
self.assertEquals(len(sub_ini), 1)
self.assertEquals(sub_ini[0], os.path.join(t, "sub/file1.ini"))
# test multi-file in folder & sub-folder & ext miss & glob miss
files = ("folder_file.txt", "folder/1.txt", "folder/sub/2.txt", "folder2/3.txt", "Folder3/4.txt" "folder.rc")
with temptree(files) as t:
folder_texts = utils.search_tree(t, "folder*", ".txt")
self.assertEquals(len(folder_texts), 4)
f1 = os.path.join(t, "folder_file.txt")
f2 = os.path.join(t, "folder/1.txt")
f3 = os.path.join(t, "folder/sub/2.txt")
f4 = os.path.join(t, "folder2/3.txt")
for f in [f1, f2, f3, f4]:
self.assert_(f in folder_texts)
开发者ID:colecrawford,项目名称:swift,代码行数:32,代码来源:test_utils.py
示例2: test_pattern_upload_all_logs
def test_pattern_upload_all_logs(self):
# test empty dir
with temptree([]) as t:
conf = {'log_dir': t}
uploader = MockLogUploader(conf)
self.assertRaises(SystemExit, uploader.run_once)
def get_random_length_str(max_len=10, chars=string.ascii_letters):
return ''.join(random.choice(chars) for x in
range(random.randint(1, max_len)))
template = 'prefix_%(random)s_%(digits)s.blah.' \
'%(datestr)s%(hour)0.2d00-%(next_hour)0.2d00-%(number)s.gz'
pattern = '''prefix_.*_[0-9]+\.blah\.
(?P<year>[0-9]{4})
(?P<month>[0-1][0-9])
(?P<day>[0-3][0-9])
(?P<hour>[0-2][0-9])00-[0-9]{2}00
-[0-9]?[0-9]\.gz'''
files_that_should_match = []
# add some files that match
for i in range(24):
fname = template % {
'random': get_random_length_str(),
'digits': get_random_length_str(16, string.digits),
'datestr': datetime.now().strftime('%Y%m%d'),
'hour': i,
'next_hour': i + 1,
'number': random.randint(0, 20),
}
files_that_should_match.append(fname)
# add some files that don't match
files = list(files_that_should_match)
for i in range(24):
fname = template % {
'random': get_random_length_str(),
'digits': get_random_length_str(16, string.digits),
'datestr': datetime.now().strftime('%Y%m'),
'hour': i,
'next_hour': i + 1,
'number': random.randint(0, 20),
}
files.append(fname)
for fname in files:
print fname
with temptree(files, contents=[COMPRESSED_DATA] * len(files)) as t:
self.assertEquals(len(os.listdir(t)), 48)
conf = {'source_filename_pattern': pattern, 'log_dir': t}
uploader = MockLogUploader(conf)
uploader.run_once()
self.assertEquals(len(os.listdir(t)), 24)
self.assertEquals(len(uploader.uploaded_files), 24)
files_that_were_uploaded = set(x[0] for x in
uploader.uploaded_files)
for f in files_that_should_match:
self.assert_(os.path.join(t, f) in files_that_were_uploaded)
开发者ID:btorch,项目名称:slogging,代码行数:60,代码来源:test_log_uploader.py
示例3: test_search_tree
def test_search_tree(self):
# file match & ext miss
with temptree(['asdf.conf', 'blarg.conf', 'asdf.cfg']) as t:
asdf = utils.search_tree(t, 'a*', '.conf')
self.assertEquals(len(asdf), 1)
self.assertEquals(asdf[0],
os.path.join(t, 'asdf.conf'))
# multi-file match & glob miss & sort
with temptree(['application.bin', 'apple.bin', 'apropos.bin']) as t:
app_bins = utils.search_tree(t, 'app*', 'bin')
self.assertEquals(len(app_bins), 2)
self.assertEquals(app_bins[0],
os.path.join(t, 'apple.bin'))
self.assertEquals(app_bins[1],
os.path.join(t, 'application.bin'))
# test file in folder & ext miss & glob miss
files = (
'sub/file1.ini',
'sub/file2.conf',
'sub.bin',
'bus.ini',
'bus/file3.ini',
)
with temptree(files) as t:
sub_ini = utils.search_tree(t, 'sub*', '.ini')
self.assertEquals(len(sub_ini), 1)
self.assertEquals(sub_ini[0],
os.path.join(t, 'sub/file1.ini'))
# test multi-file in folder & sub-folder & ext miss & glob miss
files = (
'folder_file.txt',
'folder/1.txt',
'folder/sub/2.txt',
'folder2/3.txt',
'Folder3/4.txt'
'folder.rc',
)
with temptree(files) as t:
folder_texts = utils.search_tree(t, 'folder*', '.txt')
self.assertEquals(len(folder_texts), 4)
f1 = os.path.join(t, 'folder_file.txt')
f2 = os.path.join(t, 'folder/1.txt')
f3 = os.path.join(t, 'folder/sub/2.txt')
f4 = os.path.join(t, 'folder2/3.txt')
for f in [f1, f2, f3, f4]:
self.assert_(f in folder_texts)
开发者ID:AsylumCorp,项目名称:swift,代码行数:49,代码来源:test_utils.py
示例4: test_create_container_fail
def test_create_container_fail(self):
files = [datetime.now().strftime('%Y%m%d%H')]
with temptree(files) as t:
conf = {'log_dir': t}
uploader = MockLogUploader(conf)
uploader.run_once()
self.assertEquals(len(uploader.uploaded_files), 1)
with temptree(files) as t:
conf = {'log_dir': t}
uploader = MockLogUploader(conf)
# mock create_container to fail
uploader.internal_proxy.create_container = lambda *args: False
uploader.run_once()
self.assertEquals(len(uploader.uploaded_files), 0)
开发者ID:colecrawford,项目名称:swift,代码行数:15,代码来源:test_log_uploader.py
示例5: test_init_request_processor
def test_init_request_processor(self):
config = """
[DEFAULT]
swift_dir = TEMPDIR
[pipeline:main]
pipeline = catch_errors proxy-server
[app:proxy-server]
use = egg:swift#proxy
conn_timeout = 0.2
[filter:catch_errors]
use = egg:swift#catch_errors
"""
contents = dedent(config)
with temptree(["proxy-server.conf"]) as t:
conf_file = os.path.join(t, "proxy-server.conf")
with open(conf_file, "w") as f:
f.write(contents.replace("TEMPDIR", t))
_fake_rings(t)
app, conf, logger, log_name = wsgi.init_request_processor(conf_file, "proxy-server")
# verify pipeline is catch_errors -> proxy-servery
expected = swift.common.middleware.catch_errors.CatchErrorMiddleware
self.assert_(isinstance(app, expected))
self.assert_(isinstance(app.app, swift.proxy.server.Application))
# config settings applied to app instance
self.assertEquals(0.2, app.app.conn_timeout)
# appconfig returns values from 'proxy-server' section
expected = {"__file__": conf_file, "here": os.path.dirname(conf_file), "conn_timeout": "0.2", "swift_dir": t}
self.assertEquals(expected, conf)
# logger works
logger.info("testing")
self.assertEquals("proxy-server", log_name)
开发者ID:Dieterbe,项目名称:swift,代码行数:34,代码来源:test_wsgi.py
示例6: _proxy_modify_wsgi_pipeline
def _proxy_modify_wsgi_pipeline(self, pipe):
config = """
[DEFAULT]
swift_dir = TEMPDIR
[pipeline:main]
pipeline = %s
[app:proxy-server]
use = egg:swift#proxy
conn_timeout = 0.2
[filter:healthcheck]
use = egg:swift#healthcheck
[filter:catch_errors]
use = egg:swift#catch_errors
[filter:gatekeeper]
use = egg:swift#gatekeeper
"""
config = config % (pipe,)
contents = dedent(config)
with temptree(['proxy-server.conf']) as t:
conf_file = os.path.join(t, 'proxy-server.conf')
with open(conf_file, 'w') as f:
f.write(contents.replace('TEMPDIR', t))
_fake_rings(t)
app = wsgi.loadapp(conf_file, global_conf={})
return app
开发者ID:Taejun,项目名称:swift,代码行数:30,代码来源:test_wsgi.py
示例7: test_two_realms_and_change_a_default
def test_two_realms_and_change_a_default(self):
fname = 'container-sync-realms.conf'
fcontents = '''
[DEFAULT]
mtime_check_interval = 60
[US]
key = 9ff3b71c849749dbaec4ccdd3cbab62b
cluster_dfw1 = http://dfw1.host/v1/
[UK]
key = e9569809dc8b4951accc1487aa788012
key2 = f6351bd1cc36413baa43f7ba1b45e51d
cluster_lon3 = http://lon3.host/v1/
'''
with temptree([fname], [fcontents]) as tempdir:
logger = FakeLogger()
fpath = os.path.join(tempdir, fname)
csr = ContainerSyncRealms(fpath, logger)
self.assertEqual(logger.all_log_lines(), {})
self.assertEqual(csr.mtime_check_interval, 60)
self.assertEqual(sorted(csr.realms()), ['UK', 'US'])
self.assertEqual(csr.key('US'), '9ff3b71c849749dbaec4ccdd3cbab62b')
self.assertEqual(csr.key2('US'), None)
self.assertEqual(csr.clusters('US'), ['DFW1'])
self.assertEqual(
csr.endpoint('US', 'DFW1'), 'http://dfw1.host/v1/')
self.assertEqual(csr.key('UK'), 'e9569809dc8b4951accc1487aa788012')
self.assertEqual(
csr.key2('UK'), 'f6351bd1cc36413baa43f7ba1b45e51d')
self.assertEqual(csr.clusters('UK'), ['LON3'])
self.assertEqual(
csr.endpoint('UK', 'LON3'), 'http://lon3.host/v1/')
开发者ID:Joyezhu,项目名称:swift,代码行数:33,代码来源:test_container_sync_realms.py
示例8: test_two_realms_and_change_a_default
def test_two_realms_and_change_a_default(self):
fname = "container-sync-realms.conf"
fcontents = """
[DEFAULT]
mtime_check_interval = 60
[US]
key = 9ff3b71c849749dbaec4ccdd3cbab62b
cluster_dfw1 = http://dfw1.host/v1/
[UK]
key = e9569809dc8b4951accc1487aa788012
key2 = f6351bd1cc36413baa43f7ba1b45e51d
cluster_lon3 = http://lon3.host/v1/
"""
with temptree([fname], [fcontents]) as tempdir:
logger = FakeLogger()
fpath = os.path.join(tempdir, fname)
csr = ContainerSyncRealms(fpath, logger)
self.assertEqual(logger.lines_dict, {})
self.assertEqual(csr.mtime_check_interval, 60)
self.assertEqual(sorted(csr.realms()), ["UK", "US"])
self.assertEqual(csr.key("US"), "9ff3b71c849749dbaec4ccdd3cbab62b")
self.assertEqual(csr.key2("US"), None)
self.assertEqual(csr.clusters("US"), ["DFW1"])
self.assertEqual(csr.endpoint("US", "DFW1"), "http://dfw1.host/v1/")
self.assertEqual(csr.key("UK"), "e9569809dc8b4951accc1487aa788012")
self.assertEqual(csr.key2("UK"), "f6351bd1cc36413baa43f7ba1b45e51d")
self.assertEqual(csr.clusters("UK"), ["LON3"])
self.assertEqual(csr.endpoint("UK", "LON3"), "http://lon3.host/v1/")
开发者ID:Edward1030,项目名称:swift,代码行数:30,代码来源:test_container_sync_realms.py
示例9: setUp
def setUp(self):
config = """
[DEFAULT]
swift_dir = TEMPDIR
[pipeline:main]
pipeline = healthcheck catch_errors tempurl proxy-server
[app:proxy-server]
use = egg:swift#proxy
conn_timeout = 0.2
[filter:catch_errors]
use = egg:swift#catch_errors
[filter:healthcheck]
use = egg:swift#healthcheck
[filter:tempurl]
paste.filter_factory = swift.common.middleware.tempurl:filter_factory
"""
contents = dedent(config)
with temptree(['proxy-server.conf']) as t:
conf_file = os.path.join(t, 'proxy-server.conf')
with open(conf_file, 'w') as f:
f.write(contents.replace('TEMPDIR', t))
ctx = wsgi.loadcontext(loadwsgi.APP, conf_file, global_conf={})
self.pipe = wsgi.PipelineWrapper(ctx)
开发者ID:Taejun,项目名称:swift,代码行数:29,代码来源:test_wsgi.py
示例10: test_run_server_with_latest_eventlet
def test_run_server_with_latest_eventlet(self):
config = """
[DEFAULT]
swift_dir = TEMPDIR
[pipeline:main]
pipeline = proxy-server
[app:proxy-server]
use = egg:swift#proxy
"""
def argspec_stub(server):
return mock.MagicMock(args=["capitalize_response_headers"])
contents = dedent(config)
with temptree(["proxy-server.conf"]) as t:
conf_file = os.path.join(t, "proxy-server.conf")
with open(conf_file, "w") as f:
f.write(contents.replace("TEMPDIR", t))
_fake_rings(t)
with nested(
mock.patch("swift.proxy.server.Application." "modify_wsgi_pipeline"),
mock.patch("swift.common.wsgi.wsgi"),
mock.patch("swift.common.wsgi.eventlet"),
mock.patch("swift.common.wsgi.inspect", getargspec=argspec_stub),
) as (_, _wsgi, _, _):
conf = wsgi.appconfig(conf_file)
logger = logging.getLogger("test")
sock = listen(("localhost", 0))
wsgi.run_server(conf, logger, sock)
_wsgi.server.assert_called()
args, kwargs = _wsgi.server.call_args
self.assertEquals(kwargs.get("capitalize_response_headers"), False)
开发者ID:benjkeller,项目名称:swift,代码行数:35,代码来源:test_wsgi.py
示例11: test_proxy_unmodified_wsgi_pipeline
def test_proxy_unmodified_wsgi_pipeline(self):
# Make sure things are sane even when we modify nothing
config = """
[DEFAULT]
swift_dir = TEMPDIR
[pipeline:main]
pipeline = catch_errors gatekeeper proxy-server
[app:proxy-server]
use = egg:swift#proxy
conn_timeout = 0.2
[filter:catch_errors]
use = egg:swift#catch_errors
[filter:gatekeeper]
use = egg:swift#gatekeeper
"""
contents = dedent(config)
with temptree(['proxy-server.conf']) as t:
conf_file = os.path.join(t, 'proxy-server.conf')
with open(conf_file, 'w') as f:
f.write(contents.replace('TEMPDIR', t))
_fake_rings(t)
app = wsgi.loadapp(conf_file, global_conf={})
self.assertEqual(self.pipeline_modules(app),
['swift.common.middleware.catch_errors',
'swift.common.middleware.gatekeeper',
'swift.proxy.server'])
开发者ID:Taejun,项目名称:swift,代码行数:32,代码来源:test_wsgi.py
示例12: test_proxy_modify_wsgi_pipeline
def test_proxy_modify_wsgi_pipeline(self):
config = """
[DEFAULT]
swift_dir = TEMPDIR
[pipeline:main]
pipeline = healthcheck proxy-server
[app:proxy-server]
use = egg:swift#proxy
conn_timeout = 0.2
[filter:healthcheck]
use = egg:swift#healthcheck
"""
contents = dedent(config)
with temptree(['proxy-server.conf']) as t:
conf_file = os.path.join(t, 'proxy-server.conf')
with open(conf_file, 'w') as f:
f.write(contents.replace('TEMPDIR', t))
_fake_rings(t)
app = wsgi.loadapp(conf_file, global_conf={})
self.assertEqual(self.pipeline_modules(app),
['swift.common.middleware.catch_errors',
'swift.common.middleware.gatekeeper',
'swift.common.middleware.healthcheck',
'swift.proxy.server'])
开发者ID:Taejun,项目名称:swift,代码行数:29,代码来源:test_wsgi.py
示例13: test_bad_pattern_in_config
def test_bad_pattern_in_config(self):
files = [datetime.now().strftime('%Y%m%d%H')]
with temptree(files, contents=[COMPRESSED_DATA] * len(files)) as t:
# invalid pattern
conf = {'log_dir': t, 'source_filename_pattern': '%Y%m%d%h'} # should be %H
uploader = MockLogUploader(conf)
self.assertFalse(uploader.validate_filename_pattern())
uploader.upload_all_logs()
self.assertEquals(uploader.uploaded_files, [])
conf = {'log_dir': t, 'source_filename_pattern': '%Y%m%d%H'}
uploader = MockLogUploader(conf)
self.assert_(uploader.validate_filename_pattern())
uploader.upload_all_logs()
self.assertEquals(len(uploader.uploaded_files), 1)
# deprecated warning on source_filename_format
class MockLogger():
def __init__(self):
self.msgs = defaultdict(list)
def log(self, level, msg):
self.msgs[level].append(msg)
def __getattr__(self, attr):
return partial(self.log, attr)
logger = MockLogger.logger = MockLogger()
def mock_get_logger(*args, **kwargs):
return MockLogger.logger
_orig_get_logger = log_uploader.utils.get_logger
try:
log_uploader.utils.get_logger = mock_get_logger
conf = {'source_filename_format': '%Y%m%d%H'}
uploader = MockLogUploader(conf, logger=logger)
self.assert_([m for m in logger.msgs['warning']
if 'deprecated' in m])
finally:
log_uploader.utils.get_logger = _orig_get_logger
# convert source_filename_format to regex
conf = {'source_filename_format': 'pattern-*.%Y%m%d%H.*.gz'}
uploader = MockLogUploader(conf)
expected = r'pattern-.*\.%Y%m%d%H\..*\.gz'
self.assertEquals(uploader.pattern, expected)
# use source_filename_pattern if we have the choice!
conf = {
'source_filename_format': 'bad',
'source_filename_pattern': 'good',
}
uploader = MockLogUploader(conf)
self.assertEquals(uploader.pattern, 'good')
开发者ID:colecrawford,项目名称:swift,代码行数:57,代码来源:test_log_uploader.py
示例14: test_empty
def test_empty(self):
fname = "container-sync-realms.conf"
fcontents = ""
with temptree([fname], [fcontents]) as tempdir:
logger = FakeLogger()
fpath = os.path.join(tempdir, fname)
csr = ContainerSyncRealms(fpath, logger)
self.assertEqual(logger.lines_dict, {})
self.assertEqual(csr.mtime_check_interval, 300)
self.assertEqual(csr.realms(), [])
开发者ID:Edward1030,项目名称:swift,代码行数:10,代码来源:test_container_sync_realms.py
示例15: test_proxy_modify_wsgi_pipeline_ordering
def test_proxy_modify_wsgi_pipeline_ordering(self):
config = """
[DEFAULT]
swift_dir = TEMPDIR
[pipeline:main]
pipeline = healthcheck proxy-logging bulk tempurl proxy-server
[app:proxy-server]
use = egg:swift#proxy
conn_timeout = 0.2
[filter:healthcheck]
use = egg:swift#healthcheck
[filter:proxy-logging]
use = egg:swift#proxy_logging
[filter:bulk]
use = egg:swift#bulk
[filter:tempurl]
use = egg:swift#tempurl
"""
new_req_filters = [
# not in pipeline, no afters
{"name": "catch_errors"},
# already in pipeline
{"name": "proxy_logging", "after_fn": lambda _: ["catch_errors"]},
# not in pipeline, comes after more than one thing
{"name": "container_quotas", "after_fn": lambda _: ["catch_errors", "bulk"]},
]
contents = dedent(config)
with temptree(["proxy-server.conf"]) as t:
conf_file = os.path.join(t, "proxy-server.conf")
with open(conf_file, "w") as f:
f.write(contents.replace("TEMPDIR", t))
_fake_rings(t)
with mock.patch.object(swift.proxy.server, "required_filters", new_req_filters):
app = wsgi.loadapp(conf_file, global_conf={})
self.assertEqual(
self.pipeline_modules(app),
[
"swift.common.middleware.catch_errors",
"swift.common.middleware.healthcheck",
"swift.common.middleware.proxy_logging",
"swift.common.middleware.bulk",
"swift.common.middleware.container_quotas",
"swift.common.middleware.tempurl",
"swift.proxy.server",
],
)
开发者ID:benjkeller,项目名称:swift,代码行数:55,代码来源:test_wsgi.py
示例16: test_load_app
def test_load_app(self):
config = """
[DEFAULT]
swift_dir = TEMPDIR
[pipeline:main]
pipeline = healthcheck proxy-server
[app:proxy-server]
use = egg:swift#proxy
conn_timeout = 0.2
[filter:catch_errors]
use = egg:swift#catch_errors
[filter:healthcheck]
use = egg:swift#healthcheck
"""
def modify_func(app, pipe):
new = pipe.create_filter('catch_errors')
pipe.insert_filter(new)
contents = dedent(config)
with temptree(['proxy-server.conf']) as t:
conf_file = os.path.join(t, 'proxy-server.conf')
with open(conf_file, 'w') as f:
f.write(contents.replace('TEMPDIR', t))
_fake_rings(t)
with mock.patch(
'swift.proxy.server.Application.modify_wsgi_pipeline',
modify_func):
app = wsgi.loadapp(conf_file, global_conf={})
exp = swift.common.middleware.catch_errors.CatchErrorMiddleware
self.assertTrue(isinstance(app, exp), app)
exp = swift.common.middleware.healthcheck.HealthCheckMiddleware
self.assertTrue(isinstance(app.app, exp), app.app)
exp = swift.proxy.server.Application
self.assertTrue(isinstance(app.app.app, exp), app.app.app)
# make sure you can turn off the pipeline modification if you want
def blow_up(*_, **__):
raise self.fail("needs more struts")
with mock.patch(
'swift.proxy.server.Application.modify_wsgi_pipeline',
blow_up):
app = wsgi.loadapp(conf_file, global_conf={},
allow_modify_pipeline=False)
# the pipeline was untouched
exp = swift.common.middleware.healthcheck.HealthCheckMiddleware
self.assertTrue(isinstance(app, exp), app)
exp = swift.proxy.server.Application
self.assertTrue(isinstance(app.app, exp), app.app)
开发者ID:heemanshu,项目名称:swift_juno,代码行数:55,代码来源:test_wsgi.py
示例17: test_log_cutoff
def test_log_cutoff(self):
files = [datetime.now().strftime('%Y%m%d%H')]
with temptree(files) as t:
conf = {'log_dir': t, 'new_log_cutoff': '7200'}
uploader = MockLogUploader(conf)
uploader.run_once()
self.assertEquals(len(uploader.uploaded_files), 0)
conf = {'log_dir': t, 'new_log_cutoff': '0'}
uploader = MockLogUploader(conf)
uploader.run_once()
self.assertEquals(len(uploader.uploaded_files), 1)
开发者ID:colecrawford,项目名称:swift,代码行数:11,代码来源:test_log_uploader.py
示例18: test_remove_file
def test_remove_file(self):
with temptree([]) as t:
file_name = os.path.join(t, 'blah.pid')
# assert no raise
self.assertEquals(os.path.exists(file_name), False)
self.assertEquals(utils.remove_file(file_name), None)
with open(file_name, 'w') as f:
f.write('1')
self.assert_(os.path.exists(file_name))
self.assertEquals(utils.remove_file(file_name), None)
self.assertFalse(os.path.exists(file_name))
开发者ID:AsylumCorp,项目名称:swift,代码行数:11,代码来源:test_utils.py
示例19: test_get_sig
def test_get_sig(self):
fname = "container-sync-realms.conf"
fcontents = ""
with temptree([fname], [fcontents]) as tempdir:
logger = FakeLogger()
fpath = os.path.join(tempdir, fname)
csr = ContainerSyncRealms(fpath, logger)
self.assertEqual(
csr.get_sig("GET", "/some/path", "1387212345.67890", "my_nonce", "realm_key", "user_key"),
"5a6eb486eb7b44ae1b1f014187a94529c3f9c8f9",
)
开发者ID:Edward1030,项目名称:swift,代码行数:11,代码来源:test_container_sync_realms.py
示例20: test_proxy_modify_wsgi_pipeline_ordering
def test_proxy_modify_wsgi_pipeline_ordering(self):
config = """
[DEFAULT]
swift_dir = TEMPDIR
[pipeline:main]
pipeline = healthcheck proxy-logging bulk tempurl proxy-server
[app:proxy-server]
use = egg:swift#proxy
conn_timeout = 0.2
[filter:healthcheck]
use = egg:swift#healthcheck
[filter:proxy-logging]
use = egg:swift#proxy_logging
[filter:bulk]
use = egg:swift#bulk
[filter:tempurl]
use = egg:swift#tempurl
"""
new_req_filters = [
# not in pipeline, no afters
{'name': 'catch_errors'},
# already in pipeline
{'name': 'proxy_logging',
'after': ['catch_errors']},
# not in pipeline, comes after more than one thing
{'name': 'container_quotas',
'after': ['catch_errors', 'bulk']}]
contents = dedent(config)
with temptree(['proxy-server.conf']) as t:
conf_file = os.path.join(t, 'proxy-server.conf')
with open(conf_file, 'w') as f:
f.write(contents.replace('TEMPDIR', t))
_fake_rings(t)
with mock.patch.object(swift.proxy.server, 'required_filters',
new_req_filters):
app = wsgi.loadapp(conf_file, global_conf={})
self.assertEqual(self.pipeline_modules(app), [
'swift.common.middleware.catch_errors',
'swift.common.middleware.healthcheck',
'swift.common.middleware.proxy_logging',
'swift.common.middleware.bulk',
'swift.common.middleware.container_quotas',
'swift.common.middleware.tempurl',
'swift.proxy.server'])
开发者ID:Taejun,项目名称:swift,代码行数:53,代码来源:test_wsgi.py
注:本文中的test.unit.temptree函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论