Merge "Wraps 'hostname' command"

This commit is contained in:
Zuul 2019-10-03 21:21:25 +00:00 committed by Gerrit Code Review
commit 5e97f9e10c
5 changed files with 103 additions and 12 deletions

View File

@ -18,6 +18,7 @@ from __future__ import absolute_import
from tobiko.shell.sh import _command
from tobiko.shell.sh import _exception
from tobiko.shell.sh import _execute
from tobiko.shell.sh import _hostname
from tobiko.shell.sh import _local
from tobiko.shell.sh import _io
from tobiko.shell.sh import _process
@ -39,6 +40,9 @@ ShellExecuteResult = _execute.ShellExecuteResult
join_chunks = _io.join_chunks
HostNameError = _hostname.HostnameError
get_hostname = _hostname.get_hostname
local_execute = _local.local_execute
local_process = _local.local_process
LocalShellProcessFixture = _local.LocalShellProcessFixture

View File

@ -0,0 +1,33 @@
# Copyright (c) 2019 Red Hat, Inc.
#
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from __future__ import absolute_import
import tobiko
from tobiko.shell.sh import _execute
class HostnameError(tobiko.TobikoException):
"Unable to get hostname from host"
def get_hostname(**execute_params):
result = _execute.execute('hostname', stdin=False, stdout=True,
stderr=True, expect_exit_status=0,
**execute_params)
hostname = result.stdout.splitlines()[0].strip()
if not hostname:
raise HostnameError()
return hostname

View File

@ -0,0 +1,58 @@
# Copyright (c) 2019 Red Hat, Inc.
#
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from __future__ import absolute_import
import socket
import six
import testtools
import tobiko
from tobiko.shell import sh
from tobiko.shell import ssh
from tobiko.openstack import stacks
class HostnameTest(testtools.TestCase):
server_stack = tobiko.required_setup_fixture(
stacks.CirrosServerStackFixture)
def test_hostname(self, expect_hostname=None, **execute_params):
hostname = sh.get_hostname(**execute_params)
self.assertIsInstance(hostname, six.string_types)
if expect_hostname:
self.assertEqual(expect_hostname, hostname)
else:
self.assertNotEqual('', hostname)
def test_local_hostname(self):
self.test_hostname(expect_hostname=socket.gethostname(),
ssh_client=False)
def test_remote_hostname(self, ssh_client=None):
ssh_client = ssh_client or self.server_stack.ssh_client
stdin, stdput, stderr = ssh_client.connect().exec_command('hostname')
stdin.close()
self.assertEqual(b'', stderr.read())
self.test_hostname(ssh_client=ssh_client,
expect_hostname=stdput.read().decode().strip())
def test_proxy_hostname(self):
ssh_client = ssh.ssh_proxy_client()
if ssh_client is None:
self.skip('SSH proxy server not configured')
self.test_remote_hostname(ssh_client=ssh_client)

View File

@ -40,15 +40,13 @@ class FloatingIPTest(testtools.TestCase):
def test_ssh(self):
"""Test SSH connectivity to floating IP address"""
result = sh.execute("hostname", ssh_client=self.stack.ssh_client)
self.assertEqual(self.stack.server_name.lower(),
result.stdout.rstrip())
hostname = sh.get_hostname(ssh_client=self.stack.ssh_client)
self.assertEqual(self.stack.server_name.lower(), hostname)
def test_ssh_from_cli(self):
"""Test SSH connectivity to floating IP address from CLI"""
result = sh.execute("hostname", shell=self.stack.ssh_command)
self.assertEqual(self.stack.server_name.lower(),
result.stdout.rstrip())
hostname = sh.get_hostname(shell=self.stack.ssh_command)
self.assertEqual(self.stack.server_name.lower(), hostname)
def test_ping(self):
"""Test ICMP connectivity to floating IP address"""

View File

@ -38,15 +38,13 @@ class NetworkTest(testtools.TestCase):
def test_ssh(self):
"""Test SSH connectivity to floating IP address"""
result = sh.execute("hostname", ssh_client=self.stack.ssh_client)
self.assertEqual(self.stack.server_name.lower(),
result.stdout.rstrip())
hostname = sh.get_hostname(ssh_client=self.stack.ssh_client)
self.assertEqual(self.stack.server_name.lower(), hostname)
def test_ssh_from_cli(self):
"""Test SSH connectivity to floating IP address from CLI"""
result = sh.execute("hostname", shell=self.stack.ssh_command)
self.assertEqual(self.stack.server_name.lower(),
result.stdout.rstrip())
hostname = sh.get_hostname(shell=self.stack.ssh_command)
self.assertEqual(self.stack.server_name.lower(), hostname)
def test_ping(self):
"""Test ICMP connectivity to floating IP address"""