Python os.environ 模块，copy() 实例源码

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

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)