xref: /aosp_15_r20/external/autotest/client/bin/site_sysinfo_unittest.py (revision 9c5db1993ded3edbeafc8092d69fe5de2ee02df7)

#!/usr/bin/python3

"""Tests for site_sysinfo."""

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function


__author__ = '[email protected] (Dan Shi)'

import six.moves.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
from six.moves import range
from six.moves import zip


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 get_dest_path(self, src_path):
        """Get file path in dest dir from the one in src dir.

        @param src_path: File path in src dir.

        """
        # Make sure src_path is a subpath of self.src_dir
        self.assertEqual(os.path.commonprefix((src_path, self.src_dir)),
                         self.src_dir)
        rel_path = os.path.relpath(src_path, self.src_dir)
        return os.path.join(self.dest_dir, rel_path)

    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 = self.get_dest_path(file_path)
            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.get_dest_path(self.existing_fifo_path)),
                         msg='Existing FIFO present in diff sysinfo')
        self.assertFalse(os.path.exists(self.get_dest_path(new_fifo_path)),
                         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):
        """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):
        """Accurately 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):
        """Accurately 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()