"""Module containing environments related classes."""
from testplan.common.entity import Resource, Environment
[docs]class EnvironmentCreator:
"""
Environment creator interface.
"""
def __init__(self, uid, resources=None):
self._uid = uid
self._resources = resources or []
[docs] def add_resource(self, resource):
"""
Add a resource to the future created environment.
"""
raise NotImplementedError
[docs] def uid(self):
"""
Environment uid.
"""
return self._uid
[docs] def create(self, parent):
"""
Create a new environment containing added resources.
"""
raise NotImplementedError
[docs]class LocalEnvironment(EnvironmentCreator):
"""
Creator class of a local environment.
"""
[docs] def add_resource(self, resource):
"""
Add a resource to the future created environment.
"""
if any(obj.uid() == resource.uid() for obj in self._resources):
raise RuntimeError(
"Resource with uid {} already exists.".format(resource.uid())
)
self._resources.append(resource)
[docs] def create(self, parent):
"""
Create a new environment containing added resources.
"""
env = Environment(parent=parent)
for item in self._resources:
env.add(item)
return env
[docs]class Environments(Resource):
"""
Environments holder resource.
Also inherits all
:py:class:`~testplan.common.entity.base.Resource` options.
"""
def __init__(self, uid="environments", **options):
super(Environments, self).__init__(**options)
self._envs = {}
self._uid = uid
@property
def envs(self):
"""
Returns all added environments.
"""
return self._envs
def __getattr__(self, item):
try:
return self.__getattribute__(item)
except AttributeError:
try:
return self._envs[item]
except KeyError:
# raise AttributeError so that below still works:
# getattr(env, "some_attr", "default_value")
raise AttributeError
def __getitem__(self, item):
return self._envs[item]
[docs] def delete(self, uid):
"""Delete an environment."""
del self._envs[uid]
[docs] def add(self, env, uid):
"""Add an environment."""
if uid in self._envs:
raise ValueError("{} already exists.".format(uid))
self._envs[uid] = env
[docs] def starting(self):
"""Start all added environments."""
for uid in self._envs:
self._envs[uid].start()
[docs] def stopping(self):
"""Stop all added environments."""
for uid in self._envs:
self._envs[uid].stop()
[docs] def abort_dependencies(self):
"""Abort all resources on all environments."""
for env in self._envs.values():
for resource in env:
yield resource
[docs] def aborting(self):
"""Abort logic."""
pass