Fix get_hostname function

Change-Id: Ifa8cb018e6ee35db17ea222466e7713106fd834e
This commit is contained in:
Federico Ressi 2021-06-23 10:35:57 +02:00
parent 92dbb0d709
commit 2b79313978
2 changed files with 55 additions and 24 deletions

View File

@ -18,20 +18,35 @@ from __future__ import absolute_import
import socket
import tobiko
from tobiko.shell.sh import _exception
from tobiko.shell.sh import _execute
from tobiko.shell import ssh
class HostnameError(tobiko.TobikoException):
message = "Unable to get hostname from host: {error}"
def get_hostname(ssh_client=None, **execute_params):
def get_hostname(ssh_client: ssh.SSHClientType = None,
**execute_params) -> str:
if ssh_client is False:
return socket.gethostname()
tobiko.check_valid_type(ssh_client, ssh.SSHClientFixture,
type(None))
try:
result = _execute.execute('hostname',
ssh_client=ssh_client,
**execute_params)
except _exception.ShellCommandFailed as ex:
raise HostnameError(error=ex.stderr) from ex
line: str
for line in result.stdout.splitlines():
hostname = line.strip()
if hostname:
break
else:
result = _execute.execute('hostname', stdin=False, stdout=True,
stderr=True, expect_exit_status=None,
ssh_client=ssh_client, **execute_params)
if result.exit_status or not result.stdout:
raise HostnameError(error=result.stderr)
return result.stdout.splitlines()[0].strip()
raise HostnameError(error=f"Invalid result: '{result}'")
return hostname

View File

@ -17,7 +17,6 @@ from __future__ import absolute_import
import socket
import six
import testtools
import tobiko
@ -27,15 +26,13 @@ from tobiko.openstack import keystone
from tobiko.openstack import stacks
@keystone.skip_unless_has_keystone_credentials()
class HostnameTest(testtools.TestCase):
class GetHostnameTest(testtools.TestCase):
server_stack = tobiko.required_setup_fixture(
stacks.CirrosServerStackFixture)
def test_hostname(self, expect_hostname=None, **execute_params):
def test_hostname(self,
expect_hostname: str = None,
**execute_params):
hostname = sh.get_hostname(**execute_params)
self.assertIsInstance(hostname, six.string_types)
self.assertIsInstance(hostname, str)
if expect_hostname:
self.assertEqual(expect_hostname, hostname)
else:
@ -45,16 +42,35 @@ class HostnameTest(testtools.TestCase):
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())
def test_ssh_hostname(self,
ssh_client: ssh.SSHClientFixture = None):
fixture = ssh.ssh_client_fixture(ssh_client)
if fixture is None:
expect_hostname = socket.gethostname()
else:
stdin, stdput, stderr = fixture.connect().exec_command('hostname')
stdin.close()
self.assertEqual(b'', stderr.read())
expect_hostname = stdput.read().decode().strip()
self.test_hostname(ssh_client=ssh_client,
expect_hostname=stdput.read().decode().strip())
expect_hostname=expect_hostname)
def test_proxy_hostname(self):
def test_ssh_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)
tobiko.skip_test('SSH proxy server is not configured')
self.test_ssh_hostname(ssh_client=ssh_client)
cirros_server = tobiko.required_setup_fixture(
stacks.CirrosServerStackFixture)
@keystone.skip_unless_has_keystone_credentials()
def test_cirros_hostname(self):
self.test_ssh_hostname(ssh_client=self.cirros_server.ssh_client)
ubuntu_server = tobiko.required_setup_fixture(
stacks.UbuntuServerStackFixture)
@keystone.skip_unless_has_keystone_credentials()
def test_ubuntu_hostname(self):
self.test_ssh_hostname(ssh_client=self.ubuntu_server.ssh_client)