in src/google/appengine/ext/testbed/__init__.py [0:0]
def setup_env(self, overwrite=False, **kwargs):
"""Sets default and custom environment variables.
By default, all of the items in `DEFAULT_ENVIRONMENT` will be created
without being specified. To set a value other than the default, or to pass
a custom environment variable, pass a corresponding keyword argument.
Example:
# All defaults
testbed_instance.setup_env()
# All defaults, overriding AUTH_DOMAIN in both context modes
testbed_instance.setup_env(auth_domain='custom')
# All defaults; adds a custom os.environ['CUSTOM'] = 'foo'
testbed_instance.setup_env(custom='foo')
To overwrite the values set by a previous invocation, pass `overwrite=True`.
Passing this value will not result in an `OVERWRITE` entry in `os.environ`.
If the variable corresponds to a WSGI environ variable, it will be set in
the gen2 request context as contextvars as well as `os.environ`. To only set
the gen2 context, switch to `setup_wsgi_env()`.
Args:
overwrite: Boolean. Specifies whether to overwrite items with
corresponding entries in `os.environ`.
**kwargs: Environment variables to set. The name of the argument will be
uppercased and used as a key in `os.environ`.
"""
user_specified = {key.upper(): value for key, value in kwargs.items()}
specified_app_id = (user_specified.get('APP_ID')
or full_app_id.get(environ=user_specified)
or user_specified.get('GOOGLE_CLOUD_PROJECT'))
full_app_id.clear(environ=user_specified)
user_specified.pop('APP_ID', None)
if specified_app_id:
user_specified.setdefault('GAE_APPLICATION', specified_app_id)
user_specified.setdefault('GOOGLE_CLOUD_PROJECT',
full_app_id.project_id(specified_app_id))
merged_vars = user_specified.copy()
if not overwrite:
for key, value in six.iteritems(DEFAULT_ENVIRONMENT):
if key not in merged_vars:
merged_vars[key] = value
ctx = contextvars.copy_context()
gae_vars = {k: v for k, v in vars(context.gae_headers).items()
if isinstance(v, contextvars.ContextVar)}
wsgi_vars = {k: v for k, v in vars(context.wsgi).items()
if isinstance(v, contextvars.ContextVar)}
oauth_vars = {
v.name: v
for k, v in vars(oauth_api).items()
if isinstance(v, contextvars.ContextVar)
}
for key, value in six.iteritems(merged_vars):
if key == 'GAE_APPLICATION':
if overwrite or not full_app_id.get():
full_app_id.put(value)
elif oauth_vars.get(key):
ctxvar = oauth_vars.get(key)
if ctxvar.get(None) and overwrite:
ctxvar.set(value)
elif ctxvar.get(None) is None:
token = ctxvar.set(value)
oauth_api._TESTBED_RESET_TOKENS[ctxvar] = token
elif overwrite or key not in os.environ:
if key == 'GOOGLE_CLOUD_PROJECT':
validate_project_id(value)
os.environ[key] = value
if key.startswith(context.gae_headers.PREFIX):
key = key[len(context.gae_headers.PREFIX):]
ctxvar = gae_vars.get(key, wsgi_vars.get(key))
if ctxvar is not None and (overwrite or ctxvar not in ctx):
wsgi_key = (context.gae_headers.PREFIX + key
if key in gae_vars
else key)
self.setup_wsgi_env(**{wsgi_key: value})