• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2020 The Chromium OS Authors. All rights reserved.
2# Use of this source code is governed by a BSD-style license that can be
3# found in the LICENSE file.
4
5import logging
6
7from autotest_lib.client.common_lib import error
8from autotest_lib.client.common_lib import utils
9from autotest_lib.server import test
10from autotest_lib.server.hosts.tls_client import connection
11from autotest_lib.server.hosts.tls_client import exec_dut_command
12
13
14class infra_TLSExecDUTCommand(test.test):
15    """
16    Run a command on the host via the TLS API (ExecDutCommand) and ensure the
17    behavior matches the desired test.
18
19    """
20
21    version = 1
22
23    def run_once(self, host, case):
24        """
25        Run the test.
26
27        @param host: A host object representing the DUT.
28        @param case: The case to run.
29
30        """
31        tlsconn = connection.TLSConnection()
32        self.tlsclient = exec_dut_command.TLSExecDutCommandClient(
33                tlsconn, host.hostname)
34        if case == "basic":
35            self.basic()
36        elif case == "stress":
37            self.stress()
38        elif case == "stress_fail":
39            self.stress_fail()
40        elif case == "timeout":
41            self.timeout()
42        else:
43            raise error.TestError("Case {} does not exist".format(case))
44
45    def timeout(self):
46        """Test that the timeout is respected."""
47        try:
48            self.tlsclient.run_cmd("sleep 10", timeout=5)
49        except error.CmdTimeoutError:
50            return
51        raise error.TestError("Command did not timeout.")
52
53    def stress(self):
54        """Basic command 500 times in a row."""
55        for i in range(500):
56            self.basic()
57
58    def stress_fail(self):
59        """Test a cmd that should return exit_status of 1 does so, reliably."""
60        for i in range(500):
61            res = self.tlsclient.run_cmd("NonExistingCommand")
62            if res.exit_status == 0:
63                raise error.TestError(
64                        "TLS SSH exit status was: '{}'. Expected != 0".format(
65                                res.exit_status))
66
67    def basic(self):
68        """Run a command over the TLS ExecDutCommand API. Verify output."""
69        res = self.tlsclient.run_cmd("echo success")
70        if not isinstance(res, utils.CmdResult):
71            raise error.TestError(
72                "Client returned type: '{}'. Expected type: 'utils.CmdResult'"
73                .format(type(res)))
74        if res.exit_status != 0:
75            logging.info("STD_ERR of res {}".format(res.stderr))
76            raise error.TestError(
77                    "TLS SSH exit status was: '{}'. Expected: '0'".format(
78                            res.exit_status))
79        if res.stdout != "success\n":
80            raise error.TestError("TLS returned: '{}'. Expected: '{}'".format(
81                    res.stdout, "success\n"))
82