def setup_environment():
# source: http://stackoverflow.com/questions/17278650/python-3-script-using-libnotify-fails-as-cron-job # noqa
if 'TRAVIS_CI' in os.environ:
if 'DISPLAY' not in os.environ:
# TODO: Should this be on :99 ?
os.environ['DISPLAY'] = ':0'
if 'DBUS_SESSION_BUS_ADDRESS' not in os.environ:
print('NOTE: DBUS_SESSION_BUS_ADDRESS environment var not found!')
# Setup an environment for the fixtures to share so the bus address is the same for all # noqa
environment = environ.copy()
# Setup an environment for the fixtures to share so the bus address is the same for all
environment["DBUS_SESSION_BUS_ADDRESS"] = "unix:path=" + OUTSIDE_SOCKET
print("[DBUS_SESSION_BUS_ADDRESS]: {}".format(environment["DBUS_SESSION_BUS_ADDRESS"]))
return environment
python类copy()的实例源码
def test_openssl_failure(self):
"""Make sure we stop if the openssl signature check fails."""
with temp_paths() as (le_auto_path, venv_dir):
# Serve an unrelated hash signed with the good key (easier than
# making a bad key, and a mismatch is a mismatch):
resources = {'': '<a href="certbot/">certbot/</a>',
'certbot/json': dumps({'releases': {'99.9.9': None}}),
'v99.9.9/letsencrypt-auto': build_le_auto(version='99.9.9'),
'v99.9.9/letsencrypt-auto.sig': signed('something else')}
with serving(resources) as base_url:
copy(LE_AUTO_PATH, le_auto_path)
try:
out, err = run_le_auto(le_auto_path, venv_dir, base_url)
except CalledProcessError as exc:
eq_(exc.returncode, 1)
self.assertTrue("Couldn't verify signature of downloaded "
"certbot-auto." in exc.output)
else:
self.fail('Signature check on certbot-auto erroneously passed.')
def test_openssl_failure(self):
"""Make sure we stop if the openssl signature check fails."""
with ephemeral_dir() as venv_dir:
# Serve an unrelated hash signed with the good key (easier than
# making a bad key, and a mismatch is a mismatch):
resources = {'': '<a href="certbot/">certbot/</a>',
'certbot/json': dumps({'releases': {'99.9.9': None}}),
'v99.9.9/letsencrypt-auto': build_le_auto(version='99.9.9'),
'v99.9.9/letsencrypt-auto.sig': signed('something else')}
with serving(resources) as base_url:
copy(LE_AUTO_PATH, venv_dir)
try:
out, err = run_le_auto(venv_dir, base_url)
except CalledProcessError as exc:
eq_(exc.returncode, 1)
self.assertIn("Couldn't verify signature of downloaded "
"certbot-auto.",
exc.output)
else:
self.fail('Signature check on certbot-auto erroneously passed.')
def test_env(self):
"""test env passing"""
env = environ.copy()
def randstr():
return ''.join([chr(randint(ord('A'), ord('Z'))) for i in range(randint(1, 32))])
for i in range(128):
key = randstr()
if key not in env:
env[key] = randstr()
pty = Pty()
pty.spawn(SpawnConfig(SpawnConfig.flag.auto_shutdown, cmdline = r'python tests\env.py', env = env))
with open(pty.conout_name(), 'rb') as f:
f.read()
with open('env', 'rb') as f:
self.assertEqual(pickle.load(f), env)
def run_le_auto(le_auto_path, venv_dir, base_url, **kwargs):
"""Run the prebuilt version of letsencrypt-auto, returning stdout and
stderr strings.
If the command returns other than 0, raise CalledProcessError.
"""
env = environ.copy()
d = dict(VENV_PATH=venv_dir,
# URL to PyPI-style JSON that tell us the latest released version
# of LE:
LE_AUTO_JSON_URL=base_url + 'certbot/json',
# URL to dir containing letsencrypt-auto and letsencrypt-auto.sig:
LE_AUTO_DIR_TEMPLATE=base_url + '%s/',
# The public key corresponding to signing.key:
LE_AUTO_PUBLIC_KEY="""-----BEGIN PUBLIC KEY-----
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAsMoSzLYQ7E1sdSOkwelg
tzKIh2qi3bpXuYtcfFC0XrvWig071NwIj+dZiT0OLZ2hPispEH0B7ISuuWg1ll7G
hFW0VdbxL6JdGzS2ShNWkX9hE9z+j8VqwDPOBn3ZHm03qwpYkBDwQib3KqOdYbTT
uUtJmmGcuk3a9Aq/sCT6DdfmTSdP5asdQYwIcaQreDrOosaS84DTWI3IU+UYJVgl
LsIVPBuy9IcgHidUQ96hJnoPsDCWsHwX62495QKEarauyKQrJzFes0EY95orDM47
Z5o/NDiQB11m91yNB0MmPYY9QSbnOA9j7IaaC97AwRLuwXY+/R2ablTcxurWou68
iQIDAQAB
-----END PUBLIC KEY-----""",
**kwargs)
env.update(d)
return out_and_err(
le_auto_path + ' --version',
shell=True,
env=env)
def run_le_auto(venv_dir, base_url, **kwargs):
"""Run the prebuilt version of letsencrypt-auto, returning stdout and
stderr strings.
If the command returns other than 0, raise CalledProcessError.
"""
env = environ.copy()
d = dict(XDG_DATA_HOME=venv_dir,
# URL to PyPI-style JSON that tell us the latest released version
# of LE:
LE_AUTO_JSON_URL=base_url + 'certbot/json',
# URL to dir containing letsencrypt-auto and letsencrypt-auto.sig:
LE_AUTO_DIR_TEMPLATE=base_url + '%s/',
# The public key corresponding to signing.key:
LE_AUTO_PUBLIC_KEY="""-----BEGIN PUBLIC KEY-----
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAsMoSzLYQ7E1sdSOkwelg
tzKIh2qi3bpXuYtcfFC0XrvWig071NwIj+dZiT0OLZ2hPispEH0B7ISuuWg1ll7G
hFW0VdbxL6JdGzS2ShNWkX9hE9z+j8VqwDPOBn3ZHm03qwpYkBDwQib3KqOdYbTT
uUtJmmGcuk3a9Aq/sCT6DdfmTSdP5asdQYwIcaQreDrOosaS84DTWI3IU+UYJVgl
LsIVPBuy9IcgHidUQ96hJnoPsDCWsHwX62495QKEarauyKQrJzFes0EY95orDM47
Z5o/NDiQB11m91yNB0MmPYY9QSbnOA9j7IaaC97AwRLuwXY+/R2ablTcxurWou68
iQIDAQAB
-----END PUBLIC KEY-----""",
**kwargs)
env.update(d)
return out_and_err(
join(venv_dir, 'letsencrypt-auto') + ' --version',
shell=True,
env=env)
def __init__(self, project_name: str, env: dict = {}) -> None:
self.project_path = str(COMPOSE_PROJECTS_DIR / project_name)
self.env = environ.copy()
self.env.update(env)
self.command(['up', '--force-recreate', '-d'])
def command(self, args: list, call_kwargs: dict = {}):
_call_kwargs = {'env': self.env.copy()}
_call_kwargs.update(call_kwargs)
chdir(self.project_path)
return check_output(['docker-compose'] + args, **_call_kwargs)
def start_crossbar():
finished = Deferred()
launched = Deferred()
protocol = CrossbarProcessProtocol(finished, launched, Fore.YELLOW)
exe = 'crossbar'
args = [exe, 'start', '--cbdir', './router/.crossbar']
env = environ.copy()
env["PYTHONUNBUFFERED"] = "1"
reactor.spawnProcess(protocol, exe, args, path='.', env=env)
yield launched
yield sleep(2)
returnValue(protocol)
def start_example(py_fname, color, prefix='', exe=sys.executable):
finished = Deferred()
launched = Deferred()
protocol = CrossbarProcessProtocol(finished, launched, color, prefix)
args = [exe, py_fname]
env = environ.copy()
env["PYTHONUNBUFFERED"] = "1"
reactor.spawnProcess(protocol, exe, args, path='.', env=env)
yield launched
returnValue(protocol)
def copy_file(self, filename, dest):
info("Copy {} to {}".format(filename, dest))
filename = join(self.recipe_dir, filename)
dest = join(self.build_dir, dest)
shutil.copy(filename, dest)
def get_hostrecipe_env(self, arch):
env = environ.copy()
env['PYTHONPATH'] = join(dirname(self.real_hostpython_location), 'Lib', 'site-packages')
return env
def cythonize_file(self, env, build_dir, filename):
short_filename = filename
if filename.startswith(build_dir):
short_filename = filename[len(build_dir) + 1:]
info(u"Cythonize {}".format(short_filename))
cyenv = env.copy()
if 'CYTHONPATH' in cyenv:
cyenv['PYTHONPATH'] = cyenv['CYTHONPATH']
elif 'PYTHONPATH' in cyenv:
del cyenv['PYTHONPATH']
if 'PYTHONNOUSERSITE' in cyenv:
cyenv.pop('PYTHONNOUSERSITE')
cython = 'cython' if self.ctx.python_recipe.from_crystax else self.ctx.cython
cython_command = sh.Command(cython)
shprint(cython_command, filename, *self.cython_args, _env=cyenv)
def patch_and_capture_env_for_download_all_image_descriptions(testcase):
class CaptureEnv:
"""Fake function; records a copy of the environment."""
def __call__(self, *args, **kwargs):
self.args = args
self.env = environ.copy()
return MagicMock()
capture = testcase.patch(
bootsources, 'download_all_image_descriptions', CaptureEnv())
return capture
def patch_and_capture_env_for_download_all_boot_resources(self):
class CaptureEnv:
"""Fake function; records a copy of the environment."""
def __call__(self, *args, **kwargs):
self.args = args
self.env = environ.copy()
capture = self.patch(
bootresources, 'download_all_boot_resources', CaptureEnv())
return capture
def start_parent():
"""
Start the parent that will simply run the child forever until stopped.
"""
while True:
args = [sys.executable] + sys.argv
new_environ = environ.copy()
new_environ["_IN_CHILD"] = 'yes'
ret = subprocess.call(args, env=new_environ)
if ret != settings.CODE_RELOAD_EXIT:
return ret
def load_manifest(args, filename):
# Avoid modifying the system environment.
menv = environ.copy()
with open(filename, 'r') as f:
spec = f.read()
menv['IMAGE'] = args.image
menv['NAME'] = args.name
menv['NAMESPACE'] = args.namespace
for env in args.env:
(var, value) = env.split('=', 1)
menv[var] = value
def envrep(m):
funcs = {
'b64encode': lambda v: b64encode(v.encode('utf-8')).decode('utf-8'),
}
bits = m.group(2).split(':')
try:
var = menv[bits[0]]
except KeyError:
stderr.write(args.manifest+ ": $" + bits[0] + " not in environment.\n")
exit(1)
if len(bits) > 1:
if bits[1] not in funcs:
stderr.write(args.manifest + ": function " + bits[1] + " unknown.\n")
return funcs[bits[1]](var, *bits[2:])
else:
return var
spec = re.sub(r"\$({)?([A-Za-z_][A-Za-z0-9_:]+)(?(1)})", envrep, spec)
items = []
for item in yaml.load_all(spec):
items.append(item)
return items
def extract_source(self, source, cwd):
"""
(internal) Extract the `source` into the directory `cwd`.
"""
if not source:
return
if isfile(source):
info("Extract {} into {}".format(source, cwd))
if source.endswith(".tgz") or source.endswith(".tar.gz"):
shprint(sh.tar, "-C", cwd, "-xvzf", source)
elif source.endswith(".tbz2") or source.endswith(".tar.bz2"):
shprint(sh.tar, "-C", cwd, "-xvjf", source)
elif source.endswith(".zip"):
zf = zipfile.ZipFile(source)
zf.extractall(path=cwd)
zf.close()
else:
warning(
"Error: cannot extract, unrecognized extension for {}"
.format(source))
raise Exception()
elif isdir(source):
info("Copying {} into {}".format(source, cwd))
shprint(sh.cp, '-a', source, cwd)
else:
warning(
"Error: cannot extract or copy, unrecognized path {}"
.format(source))
raise Exception()
# def get_archive_rootdir(self, filename):
# if filename.endswith(".tgz") or filename.endswith(".tar.gz") or \
# filename.endswith(".tbz2") or filename.endswith(".tar.bz2"):
# archive = tarfile.open(filename)
# root = archive.next().path.split("/")
# return root[0]
# elif filename.endswith(".zip"):
# with zipfile.ZipFile(filename) as zf:
# return dirname(zf.namelist()[0])
# else:
# print("Error: cannot detect root directory")
# print("Unrecognized extension for {}".format(filename))
# raise Exception()
def install_python_package(self, arch, name=None, env=None, is_dir=True):
'''Automate the installation of a Python package (or a cython
package where the cython components are pre-built).'''
# arch = self.filtered_archs[0] # old kivy-ios way
if name is None:
name = self.name
if env is None:
env = self.get_recipe_env(arch)
info('Installing {} into site-packages'.format(self.name))
with current_directory(self.get_build_dir(arch.arch)):
hostpython = sh.Command(self.hostpython_location)
if self.ctx.python_recipe.from_crystax:
# hppath = join(dirname(self.hostpython_location), 'Lib',
# 'site-packages')
hpenv = env.copy()
# if 'PYTHONPATH' in hpenv:
# hpenv['PYTHONPATH'] = ':'.join([hppath] +
# hpenv['PYTHONPATH'].split(':'))
# else:
# hpenv['PYTHONPATH'] = hppath
# hpenv['PYTHONHOME'] = self.ctx.get_python_install_dir()
# shprint(hostpython, 'setup.py', 'build',
# _env=hpenv, *self.setup_extra_args)
shprint(hostpython, 'setup.py', 'install', '-O2',
'--root={}'.format(self.ctx.get_python_install_dir()),
'--install-lib=.',
# AND: will need to unhardcode the 3.5 when adding 2.7 (and other crystax supported versions)
_env=hpenv, *self.setup_extra_args)
# site_packages_dir = self.ctx.get_site_packages_dir()
# built_files = glob.glob(join('build', 'lib*', '*'))
# for filen in built_files:
# shprint(sh.cp, '-r', filen, join(site_packages_dir, split(filen)[-1]))
elif self.call_hostpython_via_targetpython:
shprint(hostpython, 'setup.py', 'install', '-O2', _env=env,
*self.setup_extra_args)
else:
hppath = join(dirname(self.hostpython_location), 'Lib',
'site-packages')
hpenv = env.copy()
if 'PYTHONPATH' in hpenv:
hpenv['PYTHONPATH'] = ':'.join([hppath] +
hpenv['PYTHONPATH'].split(':'))
else:
hpenv['PYTHONPATH'] = hppath
shprint(hostpython, 'setup.py', 'install', '-O2',
'--root={}'.format(self.ctx.get_python_install_dir()),
'--install-lib=lib/python2.7/site-packages',
_env=hpenv, *self.setup_extra_args)
# AND: Hardcoded python2.7 needs fixing
# If asked, also install in the hostpython build dir
if self.install_in_hostpython:
self.install_hostpython_package(arch)
