Commit 09611583 authored by Łukasz Nowak's avatar Łukasz Nowak Committed by Łukasz Nowak

software/kvm: Test access to default and kvm-cluster

Prove that:

 * VNC is accessible
 * monitor-base-url on each partition is accessible with password extracted
   from the main partition
parent 0f199b76
...@@ -25,7 +25,13 @@ ...@@ -25,7 +25,13 @@
# #
############################################################################## ##############################################################################
import httplib
import json
import os import os
import requests
import slapos.util
import sqlite3
import urlparse
from slapos.recipe.librecipe import generateHashFromFiles from slapos.recipe.librecipe import generateHashFromFiles
from slapos.testing.testcase import makeModuleSetUpAndTestCaseClass from slapos.testing.testcase import makeModuleSetUpAndTestCaseClass
...@@ -64,3 +70,185 @@ class ServicesTestCase(InstanceTestCase): ...@@ -64,3 +70,185 @@ class ServicesTestCase(InstanceTestCase):
expected_process_name = name.format(hash=h) expected_process_name = name.format(hash=h)
self.assertIn(expected_process_name, process_names) self.assertIn(expected_process_name, process_names)
class MonitorAccessMixin(object):
def sqlite3_connect(self):
sqlitedb_file = os.path.join(
os.path.abspath(
os.path.join(
self.slap.instance_directory, os.pardir
)
), 'var', 'proxy.db'
)
return sqlite3.connect(sqlitedb_file)
def get_all_instantiated_partition_list(self):
connection = self.sqlite3_connect()
def dict_factory(cursor, row):
d = {}
for idx, col in enumerate(cursor.description):
d[col[0]] = row[idx]
return d
connection.row_factory = dict_factory
cursor = connection.cursor()
cursor.execute(
"SELECT reference, xml, connection_xml, partition_reference, "
"software_release, requested_state, software_type "
"FROM partition14 "
"WHERE slap_state='busy'")
return cursor.fetchall()
def test_access_monitor(self):
connection_parameter_dict = self.computer_partition\
.getConnectionParameterDict()
monitor_setup_url = connection_parameter_dict['monitor-setup-url']
monitor_url_with_auth = 'https' + monitor_setup_url.split('https')[2]
auth = urlparse.parse_qs(urlparse.urlparse(monitor_url_with_auth).path)
# check that monitor-base-url for all partitions in the tree are accessible
# with published username and password
partition_with_monitor_base_url_count = 0
for partition_information in self.get_all_instantiated_partition_list():
connection_xml = partition_information.get('connection_xml')
if not connection_xml:
continue
connection_dict = slapos.util.xml2dict(
partition_information['connection_xml'].encode('utf-8'))
monitor_base_url = connection_dict.get('monitor-base-url')
if not monitor_base_url:
continue
result = requests.get(
monitor_base_url, verify=False, auth=(
auth['username'][0],
auth['password'][0])
)
self.assertEqual(
httplib.OK,
result.status_code
)
partition_with_monitor_base_url_count += 1
self.assertEqual(
self.expected_partition_with_monitor_base_url_count,
partition_with_monitor_base_url_count
)
class TestAccessDefault(MonitorAccessMixin, InstanceTestCase):
__partition_reference__ = 'ad'
expected_partition_with_monitor_base_url_count = 1
def test(self):
connection_parameter_dict = self.computer_partition\
.getConnectionParameterDict()
result = requests.get(connection_parameter_dict['url'], verify=False)
self.assertEqual(
httplib.OK,
result.status_code
)
self.assertTrue('<title>noVNC</title>' in result.text)
self.assertFalse('url-additional' in connection_parameter_dict)
class TestAccessDefaultAdditional(MonitorAccessMixin, InstanceTestCase):
__partition_reference__ = 'ada'
expected_partition_with_monitor_base_url_count = 1
@classmethod
def getInstanceParameterDict(cls):
return {
'frontend-additional-instance-guid': 'SOMETHING'
}
def test(self):
connection_parameter_dict = self.computer_partition\
.getConnectionParameterDict()
result = requests.get(connection_parameter_dict['url'], verify=False)
self.assertEqual(
httplib.OK,
result.status_code
)
self.assertTrue('<title>noVNC</title>' in result.text)
result = requests.get(
connection_parameter_dict['url-additional'], verify=False)
self.assertEqual(
httplib.OK,
result.status_code
)
self.assertTrue('<title>noVNC</title>' in result.text)
class TestAccessKvmCluster(MonitorAccessMixin, InstanceTestCase):
__partition_reference__ = 'akc'
expected_partition_with_monitor_base_url_count = 2
@classmethod
def getInstanceSoftwareType(cls):
return 'kvm-cluster'
@classmethod
def getInstanceParameterDict(cls):
return {'_': json.dumps({
"kvm-partition-dict": {
"KVM0": {
"disable-ansible-promise": True
}
}
})}
def test(self):
connection_parameter_dict = self.computer_partition\
.getConnectionParameterDict()
result = requests.get(connection_parameter_dict['kvm0-url'], verify=False)
self.assertEqual(
httplib.OK,
result.status_code
)
self.assertTrue('<title>noVNC</title>' in result.text)
self.assertFalse('kvm0-url-additional' in connection_parameter_dict)
class TestAccessKvmClusterAdditional(MonitorAccessMixin, InstanceTestCase):
__partition_reference__ = 'akca'
expected_partition_with_monitor_base_url_count = 2
@classmethod
def getInstanceSoftwareType(cls):
return 'kvm-cluster'
@classmethod
def getInstanceParameterDict(cls):
return {'_': json.dumps({
"frontend": {
'frontend-additional-instance-guid': 'SOMETHING',
},
"kvm-partition-dict": {
"KVM0": {
"disable-ansible-promise": True,
}
}
})}
def test(self):
connection_parameter_dict = self.computer_partition\
.getConnectionParameterDict()
result = requests.get(connection_parameter_dict['kvm0-url'], verify=False)
self.assertEqual(
httplib.OK,
result.status_code
)
self.assertTrue('<title>noVNC</title>' in result.text)
result = requests.get(
connection_parameter_dict['kvm0-url-additional'], verify=False)
self.assertEqual(
httplib.OK,
result.status_code
)
self.assertTrue('<title>noVNC</title>' in result.text)
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment