blob: 0429ffb68ad10e9c8de7228b504126cf7555fdff [file] [log] [blame]
# -*- coding: utf-8 -*-
# Copyright 2020 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.
"""Module for updating the stateful partition on the device.
Use this module to update the stateful partition given a stateful payload
(e.g. stateful.tgz) on the device. This module untars/uncompresses the payload
on the device into var_new and dev_image_new directories. Optinonally, you can
ask this module to reset a stateful partition by preparing it to be clobbered on
from __future__ import print_function
import os
import tempfile
from chromite.lib import constants
from chromite.lib import cros_build_lib
from chromite.lib import cros_logging as logging
from chromite.lib import osutils
class Error(Exception):
"""Base exception class of StatefulUpdater errors."""
class StatefulUpdater(object):
"""The module for updating the stateful partition."""
_VAR_DIR = 'var_new'
_DEV_IMAGE_DIR = 'dev_image_new'
_UPDATE_TYPE_FILE = '.update_available'
def __init__(self, device, stateful_dir=constants.STATEFUL_DIR):
"""Initializes the module.
device: The ChromiumOsDevice to be updated.
stateful_dir: The stateful directory on the Chromium OS device.
self._device = device
self._stateful_dir = stateful_dir
self._var_dir = os.path.join(self._stateful_dir, self._VAR_DIR)
self._dev_image_dir = os.path.join(self._stateful_dir, self._DEV_IMAGE_DIR)
self._update_type_file = os.path.join(self._stateful_dir,
def Update(self, payload_path_on_device, update_type=None):
"""Updates the stateful partition given the update file.
payload_path_on_device: The path to the stateful update (stateful.tgz)
on the DUT.
update_type: The type of the stateful update to be marked. Accepted
values: 'standard' (default) and 'clobber'.
if not self._device.IfPathExists(payload_path_on_device):
raise Error('Missing the file: %s' % payload_path_on_device)
cmd = ['tar', '--ignore-command-error', '--overwrite',
'--directory', self._stateful_dir, '-xzf', payload_path_on_device]
except cros_build_lib.RunCommandError as e:
raise Error('Failed to untar the stateful update with error %s' % e)
# Make sure target directories are generated on the device.
if (not self._device.IfPathExists(self._var_dir) or
not self._device.IfPathExists(self._dev_image_dir)):
raise Error('Missing var or dev_image in stateful payload.')
self._MarkUpdateType(update_type if update_type is not None
def _MarkUpdateType(self, update_type):
"""Marks the type of the update.
update_type: The type of the update to be marked. See Update()
if update_type not in (self.UPDATE_TYPE_CLOBBER, self.UPDATE_TYPE_STANDARD):
raise Error('Invalid update type %s' % update_type)
with tempfile.NamedTemporaryFile() as f:
if update_type == self.UPDATE_TYPE_STANDARD:
logging.notice('Performing standard stateful update...')
elif update_type == self.UPDATE_TYPE_CLOBBER:
logging.notice('Restoring stateful to factory_install '
'with dev_image...')
osutils.WriteFile(, 'clobber')
self._device.CopyToDevice(, self._update_type_file, 'scp')
except cros_build_lib.RunCommandError as e:
raise Error('Failed to copy update type file to device with error %s' %
def Reset(self):
"""Resets the stateful partition."""'Resetting stateful update state.')
try:['rm', '-rf', self._update_type_file,
self._var_dir, self._dev_image_dir])
except cros_build_lib.RunCommandError as e:
logging.warning('(ignoring) Failed to delete stateful update paths with'
' error: %s', e)