import threading
import traceback
import os
-import re
import shlex
from io import BytesIO, StringIO
conf_fp = BytesIO()
ctx.ceph[cluster].conf.write(conf_fp)
conf_fp.seek(0)
- lines = conf_fp.readlines()
- m = None
- for l in lines:
- m = re.search("rgw.crypt.sse.s3.backend *= *(.*)", l.decode())
- if m:
- break
- ctx.ceph[cluster].rgw_crypt_sse_s3_backend = m.expand("\\1") if m else None
- conf_fp.seek(0)
writes = ctx.cluster.run(
args=[
'sudo', 'mkdir', '-p', '/etc/ceph', run.Raw('&&'),
testdir = teuthology.get_testdir(ctx)
for client, client_config in config.items():
client_config = client_config or {}
- (cluster_name,_,_) = teuthology.split_role(client)
(remote,) = ctx.cluster.only(client).remotes.keys()
args = [
'S3TEST_CONF={tdir}/archive/s3-tests.{client}.conf'.format(tdir=testdir, client=client),
attrs += ['!fails_with_subdomain']
if client_config.get('without-sse-s3'):
attrs += ['!sse-s3']
- elif client_config.get('with-sse-s3'):
- pass
- elif ctx.ceph[cluster_name].rgw_crypt_sse_s3_backend is None:
- attrs += ['!sse-s3']
if 'extra_attrs' in client_config:
attrs = client_config.get('extra_attrs')