# Copyright 2018 Google Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Module for the manager of services."""
# TODO(xpconanfan: move the device errors to a more generic location so
# other device controllers like iOS can share it.
import collections
import inspect
from mobly import expects
from mobly.controllers.android_device_lib import errors
from mobly.controllers.android_device_lib.services import base_service
[docs]class Error(errors.DeviceError):
"""Root error type for this module."""
[docs]class ServiceManager:
"""Manager for services of AndroidDevice.
A service is a long running process that involves an Android device, like
adb logcat or Snippet.
"""
def __init__(self, device):
self._service_objects = collections.OrderedDict()
self._device = device
[docs] def has_service_by_name(self, name):
"""Checks if the manager has a service registered with a specific name.
Args:
name: string, the name to look for.
Returns:
True if a service is registered with the specified name, False
otherwise.
"""
return name in self._service_objects
@property
def is_any_alive(self):
"""True if any service is alive; False otherwise."""
for service in self._service_objects.values():
if service.is_alive:
return True
return False
[docs] def register(self, alias, service_class, configs=None, start_service=True):
"""Registers a service.
This will create a service instance, starts the service, and adds the
instance to the mananger.
Args:
alias: string, the alias for this instance.
service_class: class, the service class to instantiate.
configs: (optional) config object to pass to the service class's
constructor.
start_service: bool, whether to start the service instance or not.
Default is True.
"""
if not inspect.isclass(service_class):
raise Error(self._device, '"%s" is not a class!' % service_class)
if not issubclass(service_class, base_service.BaseService):
raise Error(self._device,
'Class %s is not a subclass of BaseService!' % service_class)
if alias in self._service_objects:
raise Error(self._device,
'A service is already registered with alias "%s".' % alias)
service_obj = service_class(self._device, configs)
service_obj.alias = alias
if start_service:
service_obj.start()
self._service_objects[alias] = service_obj
[docs] def unregister(self, alias):
"""Unregisters a service instance.
Stops a service and removes it from the manager.
Args:
alias: string, the alias of the service instance to unregister.
"""
if alias not in self._service_objects:
raise Error(self._device,
'No service is registered with alias "%s".' % alias)
service_obj = self._service_objects.pop(alias)
if service_obj.is_alive:
with expects.expect_no_raises('Failed to stop service instance "%s".' %
alias):
service_obj.stop()
[docs] def for_each(self, func):
"""Executes a function with all registered services.
Args:
func: function, the function to execute. This function should take
a service object as args.
"""
aliases = list(self._service_objects.keys())
for alias in aliases:
with expects.expect_no_raises('Failed to execute "%s" for service "%s".' %
(func.__name__, alias)):
func(self._service_objects[alias])
[docs] def list_live_services(self):
"""Lists the aliases of all the services that are alive.
Order of this list is determined by the order the services are
registered in.
Returns:
list of strings, the aliases of the services that are running.
"""
aliases = []
self.for_each(lambda service: aliases.append(service.alias)
if service.is_alive else None)
return aliases
[docs] def create_output_excerpts_all(self, test_info):
"""Creates output excerpts from all services.
This calls `create_output_excerpts` on all registered services.
Args:
test_info: RuntimeTestInfo, the test info associated with the scope
of the excerpts.
Returns:
Dict, keys are the names of the services, values are the paths to
the excerpt files created by the corresponding services.
"""
excerpt_paths = {}
def create_output_excerpts_for_one(service):
if not service.is_alive:
return
paths = service.create_output_excerpts(test_info)
excerpt_paths[service.alias] = paths
self.for_each(create_output_excerpts_for_one)
return excerpt_paths
[docs] def unregister_all(self):
"""Safely unregisters all active instances.
Errors occurred here will be recorded but not raised.
"""
aliases = list(self._service_objects.keys())
for alias in aliases:
self.unregister(alias)
[docs] def start_all(self):
"""Starts all inactive service instances.
Services will be started in the order they were registered.
"""
for alias, service in self._service_objects.items():
if not service.is_alive:
with expects.expect_no_raises('Failed to start service "%s".' % alias):
service.start()
[docs] def start_services(self, service_alises):
"""Starts the specified services.
Services will be started in the order specified by the input list.
No-op for services that are already running.
Args:
service_alises: list of strings, the aliases of services to start.
"""
for name in service_alises:
if name not in self._service_objects:
raise Error(
self._device,
'No service is registered under the name "%s", cannot start.' %
name)
service = self._service_objects[name]
if not service.is_alive:
service.start()
[docs] def stop_all(self):
"""Stops all active service instances.
Services will be stopped in the reverse order they were registered.
"""
# OrdereDict#items does not return a sequence in Python 3.4, so we have
# to do a list conversion here.
for alias, service in reversed(list(self._service_objects.items())):
if service.is_alive:
with expects.expect_no_raises('Failed to stop service "%s".' % alias):
service.stop()
[docs] def pause_all(self):
"""Pauses all service instances.
Services will be paused in the reverse order they were registered.
"""
# OrdereDict#items does not return a sequence in Python 3.4, so we have
# to do a list conversion here.
for alias, service in reversed(list(self._service_objects.items())):
with expects.expect_no_raises('Failed to pause service "%s".' % alias):
service.pause()
[docs] def resume_all(self):
"""Resumes all service instances.
Services will be resumed in the order they were registered.
"""
for alias, service in self._service_objects.items():
with expects.expect_no_raises('Failed to resume service "%s".' % alias):
service.resume()
[docs] def resume_services(self, service_alises):
"""Resumes the specified services.
Services will be resumed in the order specified by the input list.
Args:
service_alises: list of strings, the names of services to start.
"""
for name in service_alises:
if name not in self._service_objects:
raise Error(
self._device,
'No service is registered under the name "%s", cannot resume.' %
name)
service = self._service_objects[name]
service.resume()
def __getattr__(self, name):
"""Syntactic sugar to enable direct access of service objects by alias.
Args:
name: string, the alias a service object was registered under.
"""
if self.has_service_by_name(name):
return self._service_objects[name]
return self.__getattribute__(name)