82 lines
2.8 KiB
Python
82 lines
2.8 KiB
Python
# Copyright 2020 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.
|
|
|
|
import logging
|
|
|
|
from autotest_lib.client.common_lib import error
|
|
from autotest_lib.client.common_lib import utils
|
|
from autotest_lib.server import test
|
|
from autotest_lib.server.hosts.tls_client import connection
|
|
from autotest_lib.server.hosts.tls_client import exec_dut_command
|
|
|
|
|
|
class infra_TLSExecDUTCommand(test.test):
|
|
"""
|
|
Run a command on the host via the TLS API (ExecDutCommand) and ensure the
|
|
behavior matches the desired test.
|
|
|
|
"""
|
|
|
|
version = 1
|
|
|
|
def run_once(self, host, case):
|
|
"""
|
|
Run the test.
|
|
|
|
@param host: A host object representing the DUT.
|
|
@param case: The case to run.
|
|
|
|
"""
|
|
tlsconn = connection.TLSConnection()
|
|
self.tlsclient = exec_dut_command.TLSExecDutCommandClient(
|
|
tlsconn, host.hostname)
|
|
if case == "basic":
|
|
self.basic()
|
|
elif case == "stress":
|
|
self.stress()
|
|
elif case == "stress_fail":
|
|
self.stress_fail()
|
|
elif case == "timeout":
|
|
self.timeout()
|
|
else:
|
|
raise error.TestError("Case {} does not exist".format(case))
|
|
|
|
def timeout(self):
|
|
"""Test that the timeout is respected."""
|
|
try:
|
|
self.tlsclient.run_cmd("sleep 10", timeout=5)
|
|
except error.CmdTimeoutError:
|
|
return
|
|
raise error.TestError("Command did not timeout.")
|
|
|
|
def stress(self):
|
|
"""Basic command 500 times in a row."""
|
|
for i in range(500):
|
|
self.basic()
|
|
|
|
def stress_fail(self):
|
|
"""Test a cmd that should return exit_status of 1 does so, reliably."""
|
|
for i in range(500):
|
|
res = self.tlsclient.run_cmd("NonExistingCommand")
|
|
if res.exit_status == 0:
|
|
raise error.TestError(
|
|
"TLS SSH exit status was: '{}'. Expected != 0".format(
|
|
res.exit_status))
|
|
|
|
def basic(self):
|
|
"""Run a command over the TLS ExecDutCommand API. Verify output."""
|
|
res = self.tlsclient.run_cmd("echo success")
|
|
if not isinstance(res, utils.CmdResult):
|
|
raise error.TestError(
|
|
"Client returned type: '{}'. Expected type: 'utils.CmdResult'"
|
|
.format(type(res)))
|
|
if res.exit_status != 0:
|
|
logging.info("STD_ERR of res {}".format(res.stderr))
|
|
raise error.TestError(
|
|
"TLS SSH exit status was: '{}'. Expected: '0'".format(
|
|
res.exit_status))
|
|
if res.stdout != "success\n":
|
|
raise error.TestError("TLS returned: '{}'. Expected: '{}'".format(
|
|
res.stdout, "success\n"))
|