Source code for utah.cleanup
# Ubuntu Testing Automation Harness
# Copyright 2012 Canonical Ltd.
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License version 3, as published
# by the Free Software Foundation.
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU General Public License for more details.
# You should have received a copy of the GNU General Public License along
# with this program. If not, see <http://www.gnu.org/licenses/>.
"""Generic functionality to execute callbacks on exit."""
import atexit
import logging
import os
import shutil
import traceback
from utah.commandstr import commandstr
from utah.orderedcollections import (
HashableDict,
OrderedSet,
)
from utah.process import ProcessRunner
import utah.timeout
[docs]class _Cleanup(object):
"""Cleanup allocated resources on exit.
.. warning::
This is private class not expected to be instanciated
please use :attr:`utah.cleanup.cleanup` singleton
object to call any of the methods documented below.
"""
def __init__(self):
self.paths = OrderedSet()
self.functions = OrderedSet()
self.commands = OrderedSet()
self.logger = logging.getLogger('cleanup')
[docs] def run(self):
"""Run cleanup for the resources that have been allocated."""
self.logger.debug('Running cleanup')
for function in self.functions:
self._clean_function(function)
for command in self.commands:
self._clean_command(command)
for path in self.paths:
self._clean_path(path)
def _clean_function(self, function):
"""Run clean function.
:param function:
Data needed to run the clean function: timeout, callable,
positional arguments and keyword arguments.
:type function: tuple
"""
timeout, command, args, kw = function
self.logger.debug('Running: %s',
commandstr(command, *args, **kw))
try:
utah.timeout.timeout(timeout, command, *args, **kw)
except Exception as err:
self.logger.warning(
'Exception while running cleanup function: {}'
.format(str(err)))
self.functions.remove(function)
def _clean_command(self, command):
"""Run clean command.
:param command: A command as would be passed to `subprocess.Popen`
:type command: iterable
"""
# All exceptions are captured and logged to make sure the cleanup
# process is completed as much as possible
try:
ProcessRunner(command)
except Exception:
self.logger.error('Cleanup error: {}'
.format(traceback.format_exc()))
finally:
self.commands.remove(command)
def _clean_path(self, path):
"""Clean path.
:param path: Path to a link, file or directory.
:type path: str
"""
if os.path.islink(path):
self.logger.debug('Removing link %s', path)
os.unlink(path)
elif os.path.isfile(path):
self._clean_file(path)
elif os.path.isdir(path):
self._clean_dir(path)
else:
self.logger.debug(
'{} is not a link, file, or directory; not removing'
.format(path))
self.paths.remove(path)
def _clean_file(self, path):
"""Clean file.
:param path: Path to a file.
:type path: str
"""
self.logger.debug('Changing permissions of %s', path)
try:
os.chmod(path, 0664)
except OSError as err:
self.logger.warning(
'OSError when changing file permissions: {}'
.format(str(err)))
self.logger.debug('Removing file %s', path)
try:
os.unlink(path)
except OSError as err:
self.logger.warning('OSError when removing file: {}'
.format(str(err)))
def _clean_dir(self, path):
"""Clean directory.
:param path: Path to a directory.
:type paty: str
"""
# Cribbed from http://svn.python.org
# /projects/python/trunk/Mac/BuildScript/build-installer.py
for dirpath, dirnames, filenames in os.walk(path):
for name in (dirnames + filenames):
absolute_name = os.path.join(dirpath, name)
if not os.path.islink(absolute_name):
try:
os.chmod(absolute_name, 0775)
except OSError as err:
self.logger.warning(
'OSError when changing directory permissions: {}'
.format(str(err)))
self.logger.debug('Recursively Removing directory {}'.format(path))
try:
shutil.rmtree(path)
except OSError as err:
self.logger.warning('OSError when removing directory: {}'
.format(str(err)))
[docs] def add_path(self, path):
"""Register a path to be cleaned later.
The way to clean different paths is as follows:
- Links will be unlinked.
- Files will have permissions changed to 664 and unlinked.
- Directories will recursively have permissions changed to 775
(except for links contained within) and then be recursively
removed.
:param path: A link, file, or directory to be cleaned later.
:type path: str
"""
self.paths.add(path)
[docs] def add_function(self, timeout, function, *args, **kw):
"""Register a function to be run on cleanup.
The function will be run through :func:`utah.timeout.timeout`.
:param timeout: Timeout in seconds for the function to return a result.
:type timeout: int
:param function: A callable that will do some cleanup.
:type function: callable
:param args: Positional arguments to the function.
:type args: Tuple
:param kw: Keyword arguments to the function.
:type args: dict
"""
self.functions.add((timeout, function, args, HashableDict(kw)))
[docs] def add_command(self, cmd):
"""Register a command to be run on cleanup.
:param cmd: A command as would be passed `subprocess.Popen`.
:type cmd: iterable
"""
self.commands.add(cmd)
#: Singleton object used to cleanup everything
cleanup = _Cleanup()
# Make sure cleanup is executed even if it's not explicitly called
atexit.register(cleanup.run)