site_utils/test_that_unittest.py

#!/usr/bin/python
# Copyright (c) 2013 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.
# pylint: disable-msg=C0111

import os, unittest
import mox
import common
from autotest_lib.site_utils import test_that

class StartsWithList(mox.Comparator):
    def __init__(self, start_of_list):
        self._lhs = start_of_list

    def equals(self, rhs):
        if len(rhs)<len(self._lhs):
            return False
        for (x,y) in zip(self._lhs, rhs):
            if x != y:
                return False
        return True

class TestThatUnittests(unittest.TestCase):
    def test_validate_arguments(self):
        # Deferred until validate_arguments allows for lab runs.
        pass

    def test_parse_arguments(self):
        args = test_that.parse_arguments(
                ['-b', 'some_board', '-i', 'some_image', '--args', 'some_args',
                 'some_remote', 'test1', 'test2'])
        self.assertEqual('some_board', args.board)
        self.assertEqual('some_image', args.build)
        self.assertEqual('some_args', args.args)
        self.assertEqual('some_remote', args.remote)
        self.assertEqual(['test1', 'test2'], args.tests)

    def test_schedule_local_suite(self):
        # Deferred until schedule_local_suite knows about non-local builds.
        pass

    def test_schedule_local_test(self):
        # Deferred until schedule_local_test knows about non-local builds.
        pass

    def test_run_job(self):
        class Object():
            pass

        autotest_path = 'htap_tsetotua'
        autoserv_command = os.path.join(autotest_path, 'server', 'autoserv')
        remote = 'etomer'
        self.mox = mox.Mox()

        # Create some dummy job objects.
        job1 = Object()
        job2 = Object()
        setattr(job1, 'control_type', 'cLiEnT')
        setattr(job1, 'control_file', 'c1')
        setattr(job2, 'control_type', 'Server')
        setattr(job2, 'control_file', 'c2')

        # Stub out subprocess.call, make it expect correct arguments.
        self.mox.StubOutWithMock(test_that.subprocess, 'call')
        test_that.subprocess.call(StartsWithList([autoserv_command, '-p', '-m',
                                                  remote, '-c']))
        test_that.subprocess.call(StartsWithList([autoserv_command, '-p', '-m',
                                                  remote, '-s']))

        # Test run_job.
        self.mox.ReplayAll()
        test_that.run_job(job1, remote, autotest_path)
        test_that.run_job(job2, remote, autotest_path)
        self.mox.UnsetStubs()
        self.mox.VerifyAll()
        self.mox.ResetAll()


    def test_perform_local_run(self):
        afe = test_that.setup_local_afe()
        autotest_path = 'ottotest_path'
        suite_name = 'sweet_name'
        remote = 'remoat'
        suite_control_files=['c1', 'c2', 'c3', 'c4']

        def fake_suite_callback(*args):
            for control_file in suite_control_files:
                afe.create_job(control_file, hosts=[remote])

        # Mock out scheduling of suite and running of jobs.
        self.mox = mox.Mox()
        self.mox.StubOutWithMock(test_that, 'schedule_local_suite')
        test_that.schedule_local_suite(autotest_path, suite_name,
                                     afe).WithSideEffects(fake_suite_callback)
        self.mox.StubOutWithMock(test_that, 'run_job')

        # Test perform_local_run. Enforce that run_job is called correctly.
        for control_file in suite_control_files:
            test_that.run_job(mox.ContainsAttributeValue('control_file',
                                                        control_file),
                             remote, autotest_path)
        self.mox.ReplayAll()
        test_that.perform_local_run(afe, autotest_path, ['suite:'+suite_name],
                                  remote)
        self.mox.UnsetStubs()
        self.mox.VerifyAll()


if __name__ == '__main__':
    unittest.main()