#!/usr/bin/python
"""Tests for site_sysinfo."""
__author__ = 'dshi@google.com (Dan Shi)'
import cPickle as pickle
import filecmp
import os
import random
import shutil
import tempfile
import unittest
import common
from autotest_lib.client.bin import site_sysinfo
from autotest_lib.client.common_lib import autotemp
class diffable_logdir_test(unittest.TestCase):
"""Tests for methods in class diffable_logdir."""
def setUp(self):
"""Initialize a temp direcotry with test files."""
self.tempdir = autotemp.tempdir(unique_id='diffable_logdir')
self.src_dir = os.path.join(self.tempdir.name, 'src')
self.dest_dir = os.path.join(self.tempdir.name, 'dest')
self.existing_files = ['existing_file_'+str(i) for i in range(3)]
self.existing_files_folder = ['', 'sub', 'sub/sub2']
self.existing_files_path = [os.path.join(self.src_dir, folder, f)
for f,folder in zip(self.existing_files,
self.existing_files_folder)]
self.new_files = ['new_file_'+str(i) for i in range(2)]
self.new_files_folder = ['sub', 'sub/sub3']
self.new_files_path = [os.path.join(self.src_dir, folder, f)
for f,folder in zip(self.new_files,
self.new_files_folder)]
# Create some file with random data in source directory.
for p in self.existing_files_path:
self.append_text_to_file(str(random.random()), p)
self.existing_fifo_path = os.path.join(
self.src_dir,'sub/sub2/existing_fifo')
os.mkfifo(self.existing_fifo_path)
def tearDown(self):
"""Clearn up."""
self.tempdir.clean()
def append_text_to_file(self, text, file_path):
"""Append text to the end of a file, create the file if not existed.
@param text: text to be appended to a file.
@param file_path: path to the file.
"""
dir_name = os.path.dirname(file_path)
if not os.path.exists(dir_name):
os.makedirs(dir_name)
with open(file_path, 'a') as f:
f.write(text)
def assert_trees_equal(self, dir1, dir2, ignore=None):
"""Assert two directory trees contain the same files.
@param dir1: the left comparison directory.
@param dir2: the right comparison directory.
@param ignore: filenames to ignore (in any directory).
"""
dircmps = []
dircmps.append(filecmp.dircmp(dir1, dir2, ignore))
while dircmps:
dcmp = dircmps.pop()
self.assertEqual(dcmp.left_list, dcmp.right_list)
self.assertEqual([], dcmp.diff_files)
dircmps.extend(dcmp.subdirs.values())
def test_diffable_logdir_success(self):
"""Test the diff function to save new data from a directory."""
info = site_sysinfo.diffable_logdir(self.src_dir,
keep_file_hierarchy=False,
append_diff_in_name=False)
# Run the first time to collect file status.
info.run(log_dir=None, collect_init_status=True)
# Add new files to the test directory.
for file_name, file_path in zip(self.new_files,
self.new_files_path):
self.append_text_to_file(file_name, file_path)
# Temp file for existing_file_2, used to hold on the inode. If the
# file is deleted and recreated, its inode might not change.
existing_file_2 = self.existing_files_path[2]
existing_file_2_tmp = existing_file_2 + '_tmp'
os.rename(existing_file_2, existing_file_2_tmp)
# Append data to existing file.
for file_name, file_path in zip(self.existing_files,
self.existing_files_path):
self.append_text_to_file(file_name, file_path)
# Remove the tmp file.
os.remove(existing_file_2_tmp)
# Add a new FIFO
new_fifo_path = os.path.join(self.src_dir, 'sub/sub2/new_fifo')
os.mkfifo(new_fifo_path)
# Run the second time to do diff.
info.run(self.dest_dir, collect_init_status=False, collect_all=True)
# Validate files in dest_dir.
for file_name, file_path in zip(self.existing_files+self.new_files,
self.existing_files_path+self.new_files_path):
file_path = file_path.replace('src', 'dest')
with open(file_path, 'r') as f:
self.assertEqual(file_name, f.read())
# Assert that FIFOs are not in the diff.
self.assertFalse(
os.path.exists(self.existing_fifo_path.replace('src', 'dest')),
msg='Existing FIFO present in diff sysinfo')
self.assertFalse(
os.path.exists(new_fifo_path.replace('src', 'dest')),
msg='New FIFO present in diff sysinfo')
# With collect_all=True, full sysinfo should also be present.
full_sysinfo_path = self.dest_dir + self.src_dir
self.assertTrue(os.path.exists(full_sysinfo_path),
msg='Full sysinfo not present')
# Assert that the full sysinfo is present.
self.assertNotEqual(self.src_dir, full_sysinfo_path)
self.assert_trees_equal(self.src_dir, full_sysinfo_path)
class LogdirTestCase(unittest.TestCase):
"""Tests logdir.run"""
def setUp(self):
self.tempdir = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, self.tempdir)
self.from_dir = os.path.join(self.tempdir, 'from')
self.to_dir = os.path.join(self.tempdir, 'to')
os.mkdir(self.from_dir)
os.mkdir(self.to_dir)
def _destination_path(self, relative_path, from_dir=None):
"""The expected destination path for a subdir of the source directory"""
if from_dir is None:
from_dir = self.from_dir
return '%s%s' % (self.to_dir, os.path.join(from_dir, relative_path))
def test_run_recreates_absolute_source_path(self):
"""When copying files, the absolute path from the source is recreated
in the destination folder.
"""
os.mkdir(os.path.join(self.from_dir, 'fubar'))
logdir = site_sysinfo.logdir(self.from_dir)
logdir.run(self.to_dir)
destination_path= self._destination_path('fubar')
self.assertTrue(os.path.exists(destination_path),
msg='Failed to copy to %s' % destination_path)
def test_run_skips_symlinks(self):
os.mkdir(os.path.join(self.from_dir, 'real'))
os.symlink(os.path.join(self.from_dir, 'real'),
os.path.join(self.from_dir, 'symlink'))
logdir = site_sysinfo.logdir(self.from_dir)
logdir.run(self.to_dir)
destination_path_real = self._destination_path('real')
self.assertTrue(os.path.exists(destination_path_real),
msg='real directory was not copied to %s' %
destination_path_real)
destination_path_link = self._destination_path('symlink')
self.assertFalse(
os.path.exists(destination_path_link),
msg='symlink was copied to %s' % destination_path_link)
def test_run_resolves_symlinks_to_source_root(self):
"""run tries hard to get to the source directory before copying.
Within the source folder, we skip any symlinks, but we first try to
resolve symlinks to the source root itself.
"""
os.mkdir(os.path.join(self.from_dir, 'fubar'))
from_symlink = os.path.join(self.tempdir, 'from_symlink')
os.symlink(self.from_dir, from_symlink)
logdir = site_sysinfo.logdir(from_symlink)
logdir.run(self.to_dir)
destination_path = self._destination_path('fubar')
self.assertTrue(os.path.exists(destination_path),
msg='Failed to copy to %s' % destination_path)
def test_run_excludes_common_patterns(self):
os.mkdir(os.path.join(self.from_dir, 'autoserv2344'))
# Create empty file.
open(os.path.join(self.from_dir, 'system.journal'), 'w').close()
deeper_subdir = os.path.join('prefix', 'autoserv', 'suffix')
os.makedirs(os.path.join(self.from_dir, deeper_subdir))
logdir = site_sysinfo.logdir(self.from_dir)
logdir.run(self.to_dir)
destination_path = self._destination_path('autoserv2344')
self.assertFalse(os.path.exists(destination_path),
msg='Copied banned file to %s' % destination_path)
destination_path = self._destination_path(deeper_subdir)
self.assertFalse(os.path.exists(destination_path),
msg='Copied banned file to %s' % destination_path)
destination_path = self._destination_path('system.journal')
self.assertFalse(os.path.exists(destination_path),
msg='Copied banned file to %s' % destination_path)
def test_run_ignores_exclude_patterns_in_leading_dirs(self):
"""Exclude patterns should only be applied to path suffixes within
from_dir, not to the root directory itself.
"""
exclude_pattern_dir = os.path.join(self.from_dir, 'autoserv2344')
os.makedirs(os.path.join(exclude_pattern_dir, 'fubar'))
logdir = site_sysinfo.logdir(exclude_pattern_dir)
logdir.run(self.to_dir)
destination_path = self._destination_path('fubar',
from_dir=exclude_pattern_dir)
self.assertTrue(os.path.exists(destination_path),
msg='Failed to copy to %s' % destination_path)
def test_pickle_unpickle_equal(self):
"""Sanity check pickle-unpickle round-trip."""
logdir = site_sysinfo.logdir(
self.from_dir,
excludes=(site_sysinfo.logdir.DEFAULT_EXCLUDES + ('a',)))
# base_job uses protocol 2 to pickle. We follow suit.
logdir_pickle = pickle.dumps(logdir, protocol=2)
unpickled_logdir = pickle.loads(logdir_pickle)
self.assertEqual(unpickled_logdir, logdir)
def test_pickle_enforce_required_attributes(self):
"""Some attributes from this object should never be dropped.
When running a client test against an older build, we pickle the objects
of this class from newer version of the class and unpicle to older
versions of the class. The older versions require some attributes to be
present.
"""
logdir = site_sysinfo.logdir(
self.from_dir,
excludes=(site_sysinfo.logdir.DEFAULT_EXCLUDES + ('a',)))
# base_job uses protocol 2 to pickle. We follow suit.
logdir_pickle = pickle.dumps(logdir, protocol=2)
logdir = pickle.loads(logdir_pickle)
self.assertEqual(logdir.additional_exclude, 'a')
def test_pickle_enforce_required_attributes_default(self):
"""Some attributes from this object should never be dropped.
When running a client test against an older build, we pickle the objects
of this class from newer version of the class and unpicle to older
versions of the class. The older versions require some attributes to be
present.
"""
logdir = site_sysinfo.logdir(self.from_dir)
# base_job uses protocol 2 to pickle. We follow suit.
logdir_pickle = pickle.dumps(logdir, protocol=2)
logdir = pickle.loads(logdir_pickle)
self.assertEqual(logdir.additional_exclude, None)
def test_unpickle_handle_missing__excludes(self):
"""Sanely handle missing _excludes attribute from pickles
This can happen when running brand new version of this class that
introduced this attribute from older server side code in prod.
"""
logdir = site_sysinfo.logdir(self.from_dir)
delattr(logdir, '_excludes')
# base_job uses protocol 2 to pickle. We follow suit.
logdir_pickle = pickle.dumps(logdir, protocol=2)
logdir = pickle.loads(logdir_pickle)
self.assertEqual(logdir._excludes,
site_sysinfo.logdir.DEFAULT_EXCLUDES)
def test_unpickle_handle_missing__excludes_default(self):
"""Sanely handle missing _excludes attribute from pickles
This can happen when running brand new version of this class that
introduced this attribute from older server side code in prod.
"""
logdir = site_sysinfo.logdir(
self.from_dir,
excludes=(site_sysinfo.logdir.DEFAULT_EXCLUDES + ('a',)))
delattr(logdir, '_excludes')
# base_job uses protocol 2 to pickle. We follow suit.
logdir_pickle = pickle.dumps(logdir, protocol=2)
logdir = pickle.loads(logdir_pickle)
self.assertEqual(
logdir._excludes,
(site_sysinfo.logdir.DEFAULT_EXCLUDES + ('a',)))
if __name__ == '__main__':
unittest.main()