| # Copyright 2018 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| import dbus |
| import subprocess |
| |
| from autotest_lib.client.common_lib.cros import dbus_send |
| |
| # Full path of the CUPS configuration file |
| _CUPS_CONF_FILE = '/etc/cups/cupsd.conf' |
| |
| class Configurator(): |
| """ |
| An instance of this class is responsible for an initial configuration of |
| the system. This is performed by the method configure(). To restore the |
| system to the state before a configure() call, the method restore() must |
| be called. |
| |
| """ |
| |
| def __init__(self): |
| """ |
| Constructor. |
| |
| """ |
| self._cupsd_conf_loglevel_line_no = None |
| self._cupsd_conf_loglevel_content = None |
| self._loaded_components = [] |
| self._components_to_load = ['epson-inkjet-printer-escpr', |
| 'star-cups-driver'] |
| |
| |
| def _load_component(self, component): |
| """ |
| Download filter component via dbus API |
| |
| @param component: name of component |
| |
| @raises Exception if component is not loaded. |
| |
| """ |
| if component in self._loaded_components: |
| return |
| res = dbus_send.dbus_send( |
| 'org.chromium.ComponentUpdaterService', |
| 'org.chromium.ComponentUpdaterService', |
| '/org/chromium/ComponentUpdaterService', |
| 'LoadComponent', |
| timeout_seconds=20, |
| user='root', |
| args=[dbus.String(component)]) |
| if res.response == '': |
| message = 'Component %s could not be loaded.' % component |
| raise Exception(message) |
| self._loaded_components.append(component) |
| |
| |
| def _delete_component(self, component): |
| """ |
| Delete filter component via dbus API |
| |
| @param component: name of component |
| |
| """ |
| if component not in self._loaded_components: |
| return |
| dbus_send.dbus_send( |
| 'org.chromium.ComponentUpdaterService', |
| 'org.chromium.ComponentUpdaterService', |
| '/org/chromium/ComponentUpdaterService', |
| 'UnloadComponent', |
| timeout_seconds=20, |
| user='root', |
| args=[dbus.String(component)]) |
| self._loaded_components.remove(component) |
| |
| |
| def _run_as_root(self, argv): |
| """ |
| Run given command as root. |
| |
| @param argv: an array of command-line parameters |
| |
| @returns standard output produced by the command |
| |
| @raises Exception if the command returns code different than 0 |
| |
| """ |
| p1 = subprocess.Popen(["echo", "test0000"], stdout=subprocess.PIPE) |
| p2 = subprocess.Popen(["sudo", "--stdin", "--prompt="] + argv, |
| stdin=p1.stdout, stdout=subprocess.PIPE) |
| p1.stdout.close() |
| out,err = p2.communicate() |
| if p2.returncode != 0: |
| raise Exception("The command '%s' returns %d" % |
| (' '.join(argv),p2.returncode)); |
| return out |
| |
| |
| def _set_cups_logging_level(self): |
| """ |
| Modify the CUPS configuration file to set log level to 'debug'. |
| |
| @raises Exception in case of any errors |
| |
| """ |
| # parse content of the CUPS configuration file and find a number of |
| # a line with 'LogLevel' option |
| lines = self._run_as_root(["cat", _CUPS_CONF_FILE]).splitlines() |
| |
| for index, line in enumerate(lines): |
| if line.startswith('LogLevel'): |
| line_no = index |
| break |
| if line_no is None: |
| raise Exception('Cannot find a line with LogLevel in cupsd.conf') |
| # save the original line and replace it with 'LogLevel debug' |
| self._cupsd_conf_loglevel_content = lines[line_no] |
| self._cupsd_conf_loglevel_line_no = line_no + 1 |
| self._run_as_root(['sed', '-i', '%ds/.*/LogLevel debug/' % (line_no+1), |
| _CUPS_CONF_FILE]) |
| # if CUPS is started, we have to stop |
| try: |
| self._run_as_root(['stop', 'cupsd']) |
| except: |
| None |
| |
| |
| def _restore_cups_logging_level(self): |
| """ |
| Restore content of the CUPS configuration file to this one before |
| calling _set_cups_logging_level(). Do nothing if the method |
| _set_cups_logging_level() was not called earlier. |
| |
| """ |
| if self._cupsd_conf_loglevel_content is None: |
| return |
| self._run_as_root(['sed', '-i', '%ds/.*/%s/' % |
| (self._cupsd_conf_loglevel_line_no, |
| self._cupsd_conf_loglevel_content), _CUPS_CONF_FILE]) |
| self._cupsd_conf_loglevel_content = None |
| self._cupsd_conf_loglevel_line_no = None |
| |
| |
| def _set_root_partition_as_read_write(self): |
| """ |
| Remount the root partition in read-write mode. |
| |
| """ |
| self._run_as_root(['mount', '-o', 'rw,remount', '/']) |
| |
| |
| def _set_root_partition_as_read_only(self): |
| """ |
| Remount the root partition in read-only mode. |
| |
| """ |
| self._run_as_root(['mount', '-o', 'ro,remount', '/']) |
| |
| |
| def configure(self, set_cups_logging_level): |
| """ |
| Apply the configuration required by the test. |
| |
| @param set_cups_logging_level: True or False; if True then |
| the root partition is remounted in R/W mode and the CUPS |
| configuration file is updated to set "LogLevel" to "debug". |
| """ |
| # Load components required by some printers (Epson and Star) |
| for component in self._components_to_load: |
| self._load_component(component) |
| # Update CUPS logging level |
| if set_cups_logging_level: |
| self._set_root_partition_as_read_write() |
| self._set_cups_logging_level() |
| self._set_root_partition_as_read_only() |
| |
| |
| def restore(self): |
| """ |
| Restore the system state before configure(). It is safe to run |
| this method, even if configure() failed or has not been called. |
| |
| """ |
| # Restore CUPS logging level |
| if self._cupsd_conf_loglevel_content is not None: |
| self._set_root_partition_as_read_write() |
| self._restore_cups_logging_level() |
| self._set_root_partition_as_read_write() |
| # Deleted components loaded during initialization |
| for component in self._components_to_load: |
| self._delete_component(component) |
