Home | History | Annotate | Download | only in cli
      1 #
      2 # Copyright 2008 Google Inc. All Rights Reserved.
      3 
      4 """Test for cli."""
      5 
      6 import os
      7 import sys
      8 import unittest
      9 
     10 import common
     11 from autotest_lib.cli import atest, rpc
     12 from autotest_lib.frontend.afe import rpc_client_lib
     13 from autotest_lib.frontend.afe.json_rpc import proxy
     14 from autotest_lib.client.common_lib.test_utils import mock
     15 from autotest_lib.client.common_lib import autotemp
     16 
     17 CLI_USING_PDB = False
     18 CLI_UT_DEBUG = False
     19 
     20 
     21 class ExitException(Exception):
     22     """Junk that should be removed."""
     23     pass
     24 
     25 def create_file(content):
     26     """Create a temporary file for testing.
     27 
     28     @param content: string contents for file.
     29 
     30     @return: Instance of autotemp.tempfile with specified contents.
     31     """
     32     file_temp = autotemp.tempfile(unique_id='cli_mock', text=True)
     33     os.write(file_temp.fd, content)
     34     return file_temp
     35 
     36 
     37 class cli_unittest(unittest.TestCase):
     38     """General mocks and setup / teardown for testing the atest cli.
     39     """
     40     def setUp(self):
     41         """Setup mocks for rpc calls and system exit.
     42         """
     43         super(cli_unittest, self).setUp()
     44         self.god = mock.mock_god(debug=CLI_UT_DEBUG, ut=self)
     45         self.god.stub_class_method(rpc.afe_comm, 'run')
     46         self.god.stub_function(sys, 'exit')
     47 
     48         def stub_authorization_headers(*args, **kwargs):
     49             """No auth headers required for testing."""
     50             return {}
     51         self.god.stub_with(rpc_client_lib, 'authorization_headers',
     52                            stub_authorization_headers)
     53 
     54 
     55     def tearDown(self):
     56         """Remove mocks.
     57         """
     58         # Unstub first because super may need exit
     59         self.god.unstub_all()
     60         super(cli_unittest, self).tearDown()
     61 
     62 
     63     def assertEqualNoOrder(self, x, y, message=None):
     64         """Assert x and y contain the same elements.
     65 
     66         @param x: list like object for comparison.
     67         @param y: second list like object for comparison
     68         @param message: Message for AssertionError if x and y contain different
     69                         elements.
     70 
     71         @raises: AssertionError
     72         """
     73         self.assertEqual(set(x), set(y), message)
     74 
     75 
     76     def assertWords(self, string, to_find=[], not_in=[]):
     77         """Assert the string contains all of the set of words to_find and none
     78         of the set not_in.
     79 
     80         @param string: String to search.
     81         @param to_find: List of strings that must be in string.
     82         @param not_in: List of strings that must NOT be in string.
     83 
     84         @raises: AssertionError
     85         """
     86         for word in to_find:
     87             self.assert_(string.find(word) >= 0,
     88                          "Could not find '%s' in: %s" % (word, string))
     89         for word in not_in:
     90             self.assert_(string.find(word) < 0,
     91                          "Found (and shouldn't have) '%s' in: %s" % (word,
     92                                                                      string))
     93 
     94 
     95     def _check_output(self, out='', out_words_ok=[], out_words_no=[],
     96                       err='', err_words_ok=[], err_words_no=[]):
     97         if out_words_ok or out_words_no:
     98             self.assertWords(out, out_words_ok, out_words_no)
     99         else:
    100             self.assertEqual('', out)
    101 
    102         if err_words_ok or err_words_no:
    103             self.assertWords(err, err_words_ok, err_words_no)
    104         else:
    105             self.assertEqual('', err)
    106 
    107 
    108     def assertOutput(self, obj, results,
    109                      out_words_ok=[], out_words_no=[],
    110                      err_words_ok=[], err_words_no=[]):
    111         """Assert that obj's output writes the expected strings to std(out/err).
    112 
    113         An empty list for out_words_ok or err_words_ok means that the stdout
    114         or stderr (respectively) must be empty.
    115 
    116         @param obj: Command object (such as atest_add_or_remove).
    117         @param results: Results of command for obj.output to format.
    118         @param out_words_ok: List of strings that must be in stdout.
    119         @param out_words_no: List of strings that must NOT be in stdout.
    120         @param err_words_ok: List of strings that must be in stderr.
    121         @param err_words_no: List of strings that must NOT be in stderr.
    122 
    123         @raises: AssertionError
    124         """
    125         self.god.mock_io()
    126         obj.output(results)
    127         obj.show_all_failures()
    128         (out, err) = self.god.unmock_io()
    129         self._check_output(out, out_words_ok, out_words_no,
    130                            err, err_words_ok, err_words_no)
    131 
    132 
    133     def mock_rpcs(self, rpcs):
    134         """Expect and mock the results of a list of RPCs.
    135 
    136         @param rpcs: A list of tuples, each representing one RPC:
    137                      (op, args(dict), success, expected)
    138         """
    139         for (op, dargs, success, expected) in rpcs:
    140             comm = rpc.afe_comm.run
    141             if success:
    142                 comm.expect_call(op, **dargs).and_return(expected)
    143             else:
    144                 (comm.expect_call(op, **dargs).
    145                  and_raises(proxy.JSONRPCException(expected)))
    146 
    147 
    148     def run_cmd(self, argv, rpcs=[], exit_code=None,
    149                 out_words_ok=[], out_words_no=[],
    150                 err_words_ok=[], err_words_no=[]):
    151         """Run an atest command with arguments.
    152 
    153         An empty list for out_words_ok or err_words_ok means that the stdout
    154         or stderr (respectively) must be empty.
    155 
    156         @param argv: List of command and arguments as strings.
    157         @param rpcs: List of rpcs to expect the command to perform.
    158         @param exit_code: Expected exit code of the command (if not 0).
    159         @param out_words_ok: List of strings to expect in stdout.
    160         @param out_words_no: List of strings that must not be in stdout.
    161         @param err_words_ok: List of strings to expect in stderr.
    162         @param err_words_no: List of strings that must not be in stderr.
    163 
    164         @raises: AssertionError or CheckPlaybackError.
    165 
    166         @returns: stdout, stderr
    167         """
    168         sys.argv = argv
    169 
    170         self.mock_rpcs(rpcs)
    171 
    172         if not (CLI_USING_PDB and CLI_UT_DEBUG):
    173             self.god.mock_io()
    174         if exit_code is not None:
    175             sys.exit.expect_call(exit_code).and_raises(ExitException)
    176             self.assertRaises(ExitException, atest.main)
    177         else:
    178             atest.main()
    179         (out, err) = self.god.unmock_io()
    180         self.god.check_playback()
    181         self._check_output(out, out_words_ok, out_words_no,
    182                            err, err_words_ok, err_words_no)
    183         return (out, err)
    184 
    185