Skip to content

S
slapos.core
Commit 5c613eb5 authored by Marco Mariani's avatar Marco Mariani
Browse files
Options

whitespace, indentation, light cleanup

parent 630ed441
Show whitespace changes
Inline Side-by-side
Showing with 405 additions and 361 deletions
slapos/tests/slap.py
View file @ 5c613eb5
......@@ -37,6 +37,7 @@ import xml_marshaller
class UndefinedYetException(Exception):
"""To catch exceptions which are not yet defined"""
class SlapMixin(unittest.TestCase):
"""
Usefull methods for slap tests
......@@ -59,7 +60,7 @@ class SlapMixin(unittest.TestCase):
"""Overrides httplib"""
import mock.httplib
self.saved_httplib = dict()
self.saved_httplib = {}
for fake in vars(mock.httplib):
self.saved_httplib[fake] = getattr(httplib, fake, None)
......@@ -79,6 +80,7 @@ class SlapMixin(unittest.TestCase):
return self.id()
return os.environ.get('TEST_SLAP_COMPUTER_ID', self.id())
class TestSlap(SlapMixin):
"""
Test slap against slap server
......@@ -102,7 +104,8 @@ class TestSlap(SlapMixin):
server_url = 'https://user:pass@server/path/path?parameter=notAcceptable'
slap_instance = slapos.slap.slap()
self.assertRaises(AttributeError,
slap_instance.initializeConnection, server_url)
slap_instance.initializeConnection,
server_url)
def test_registerComputer_with_new_guid(self):
"""
......@@ -169,9 +172,9 @@ class TestSlap(SlapMixin):
def server_response(self, path, method, body, header):
parsed_url = urlparse.urlparse(path.lstrip('/'))
parsed_qs = urlparse.parse_qs(parsed_url.query)
if parsed_url.path == 'registerComputerPartition' and \
parsed_qs['computer_reference'][0] == computer_guid and \
parsed_qs['computer_partition_reference'][0] == partition_id:
if (parsed_url.path == 'registerComputerPartition'
and parsed_qs['computer_reference'][0] == computer_guid
and parsed_qs['computer_partition_reference'][0] == partition_id):
partition = slapos.slap.ComputerPartition(
computer_guid, partition_id)
return (200, {}, xml_marshaller.xml_marshaller.dumps(partition))
......@@ -204,9 +207,9 @@ class TestSlap(SlapMixin):
def server_response(self, path, method, body, header):
parsed_url = urlparse.urlparse(path.lstrip('/'))
parsed_qs = urlparse.parse_qs(parsed_url.query)
if parsed_url.path == 'registerComputerPartition' and \
parsed_qs['computer_reference'][0] == computer_guid and \
parsed_qs['computer_partition_reference'][0] == partition_id:
if (parsed_url.path == 'registerComputerPartition'
and parsed_qs['computer_reference'][0] == computer_guid
and parsed_qs['computer_partition_reference'][0] == partition_id):
slapos.slap.ComputerPartition(computer_guid, partition_id)
return (404, {}, '')
else:
......@@ -214,7 +217,8 @@ class TestSlap(SlapMixin):
httplib.HTTPConnection._callback = server_response
self.assertRaises(slapos.slap.NotFoundError,
self.slap.registerComputerPartition, computer_guid, partition_id)
self.slap.registerComputerPartition,
computer_guid, partition_id)
def test_getFullComputerInformation_empty_computer_guid(self):
"""
......@@ -226,11 +230,12 @@ class TestSlap(SlapMixin):
def server_response(self_httpconnection, path, method, body, header):
# Shouldn't even be called
self.assertFalse(True)
httplib.HTTPConnection._callback = server_response
self.assertRaises(
slapos.slap.NotFoundError,
self.slap._connection_helper.getFullComputerInformation, None)
self.assertRaises(slapos.slap.NotFoundError,
self.slap._connection_helper.getFullComputerInformation,
None)
def test_registerComputerPartition_empty_computer_guid(self):
"""
......@@ -242,11 +247,12 @@ class TestSlap(SlapMixin):
def server_response(self_httpconnection, path, method, body, header):
# Shouldn't even be called
self.assertFalse(True)
httplib.HTTPConnection._callback = server_response
self.assertRaises(
slapos.slap.NotFoundError,
self.slap.registerComputerPartition, None, 'PARTITION_01')
self.assertRaises(slapos.slap.NotFoundError,
self.slap.registerComputerPartition,
None, 'PARTITION_01')
def test_registerComputerPartition_empty_computer_partition_id(self):
"""
......@@ -258,10 +264,10 @@ class TestSlap(SlapMixin):
def server_response(self_httpconnection, path, method, body, header):
# Shouldn't even be called
self.assertFalse(True)
httplib.HTTPConnection._callback = server_response
self.assertRaises(
slapos.slap.NotFoundError,
self.assertRaises(slapos.slap.NotFoundError,
self.slap.registerComputerPartition,
self._getTestComputerId(), None)
......@@ -275,13 +281,14 @@ class TestSlap(SlapMixin):
def server_response(self_httpconnection, path, method, body, header):
# Shouldn't even be called
self.assertFalse(True)
httplib.HTTPConnection._callback = server_response
self.assertRaises(
slapos.slap.NotFoundError,
self.assertRaises(slapos.slap.NotFoundError,
self.slap.registerComputerPartition,
None, None)
class TestComputer(SlapMixin):
"""
Tests slapos.slap.slap.Computer class functionality
......@@ -299,15 +306,15 @@ class TestComputer(SlapMixin):
def server_response(self, path, method, body, header):
parsed_url = urlparse.urlparse(path.lstrip('/'))
parsed_qs = urlparse.parse_qs(parsed_url.query)
if parsed_url.path == 'registerComputerPartition' and \
'computer_reference' in parsed_qs and \
'computer_partition_reference' in parsed_qs:
if (parsed_url.path == 'registerComputerPartition'
and 'computer_reference' in parsed_qs
and 'computer_partition_reference' in parsed_qs):
slap_partition = slapos.slap.ComputerPartition(
parsed_qs['computer_reference'][0],
parsed_qs['computer_partition_reference'][0])
return (200, {}, xml_marshaller.xml_marshaller.dumps(slap_partition))
elif parsed_url.path == 'getFullComputerInformation' and \
'computer_id' in parsed_qs:
elif (parsed_url.path == 'getFullComputerInformation'
and 'computer_id' in parsed_qs):
slap_computer = slapos.slap.Computer(parsed_qs['computer_id'][0])
slap_computer._software_release_list = []
slap_computer._computer_partition_list = []
......@@ -331,11 +338,11 @@ class TestComputer(SlapMixin):
def server_response(self_httpconnection, path, method, body, header):
# Shouldn't even be called
self.assertFalse(True)
httplib.HTTPConnection._callback = server_response
computer = self.slap.registerComputer(None)
self.assertRaises(
slapos.slap.NotFoundError,
self.assertRaises(slapos.slap.NotFoundError,
getattr(computer, computer_method))
def test_computer_getComputerPartitionList_empty_computer_guid(self):
......@@ -373,7 +380,6 @@ class TestComputer(SlapMixin):
(not defined yet) XML raises (not defined yet) exception
"""
self.computer_guid = self._getTestComputerId()
self.slap = slapos.slap.slap()
self.slap.initializeConnection(self.server_url)
......@@ -381,7 +387,8 @@ class TestComputer(SlapMixin):
non_dtd_xml = """<xml>
<non-dtd-parameter name="xerxes">value<non-dtd-parameter name="xerxes">
</xml>"""
self.assertRaises(UndefinedYetException, self.computer.reportUsage,
self.assertRaises(UndefinedYetException,
self.computer.reportUsage,
non_dtd_xml)
@unittest.skip("Not implemented")
......@@ -402,12 +409,15 @@ class TestComputer(SlapMixin):
bad_partition_dtd_xml = """<xml>
<computer-partition id='ANOTHER_PARTITION>96.5% CPU</computer-partition>
</xml>"""
self.assertRaises(UndefinedYetException, self.computer.reportUsage,
self.assertRaises(UndefinedYetException,
self.computer.reportUsage,
bad_partition_dtd_xml)
class RequestWasCalled(Exception):
pass
class TestComputerPartition(SlapMixin):
"""
Tests slapos.slap.slap.ComputerPartition class functionality
......@@ -415,18 +425,19 @@ class TestComputerPartition(SlapMixin):
def test_request_sends_request(self):
partition_id = 'PARTITION_01'
def server_response(self, path, method, body, header):
parsed_url = urlparse.urlparse(path.lstrip('/'))
parsed_qs = urlparse.parse_qs(parsed_url.query)
if parsed_url.path == 'registerComputerPartition' and \
'computer_reference' in parsed_qs and \
'computer_partition_reference' in parsed_qs:
if (parsed_url.path == 'registerComputerPartition'
and 'computer_reference' in parsed_qs
and 'computer_partition_reference' in parsed_qs):
slap_partition = slapos.slap.ComputerPartition(
parsed_qs['computer_reference'][0],
parsed_qs['computer_partition_reference'][0])
return (200, {}, xml_marshaller.xml_marshaller.dumps(slap_partition))
elif parsed_url.path == 'getComputerInformation' and \
'computer_id' in parsed_qs:
elif (parsed_url.path == 'getComputerInformation'
and 'computer_id' in parsed_qs):
slap_computer = slapos.slap.Computer(parsed_qs['computer_id'][0])
slap_computer._software_release_list = []
slap_partition = slapos.slap.ComputerPartition(
......@@ -438,31 +449,33 @@ class TestComputerPartition(SlapMixin):
raise RequestWasCalled
else:
return (404, {}, '')
httplib.HTTPConnection._callback = server_response
self.computer_guid = self._getTestComputerId()
self.slap = slapos.slap.slap()
self.slap.initializeConnection(self.server_url)
computer_partition = self.slap.registerComputerPartition(
self.computer_guid, partition_id)
self.assertRaises(RequestWasCalled, computer_partition.request,
self.assertRaises(RequestWasCalled,
computer_partition.request,
'http://server/new/' + self._getTestComputerId(),
'software_type',
'myref')
'software_type', 'myref')
def test_request_not_raises(self):
partition_id = 'PARTITION_01'
def server_response(self, path, method, body, header):
parsed_url = urlparse.urlparse(path.lstrip('/'))
parsed_qs = urlparse.parse_qs(parsed_url.query)
if parsed_url.path == 'registerComputerPartition' and \
'computer_reference' in parsed_qs and \
'computer_partition_reference' in parsed_qs:
if (parsed_url.path == 'registerComputerPartition'
and 'computer_reference' in parsed_qs
and 'computer_partition_reference' in parsed_qs):
slap_partition = slapos.slap.ComputerPartition(
parsed_qs['computer_reference'][0],
parsed_qs['computer_partition_reference'][0])
return (200, {}, xml_marshaller.xml_marshaller.dumps(slap_partition))
elif parsed_url.path == 'getComputerInformation' and \
'computer_id' in parsed_qs:
elif (parsed_url.path == 'getComputerInformation'
and 'computer_id' in parsed_qs):
slap_computer = slapos.slap.Computer(parsed_qs['computer_id'][0])
slap_computer._software_release_list = []
slap_partition = slapos.slap.ComputerPartition(
......@@ -474,6 +487,7 @@ class TestComputerPartition(SlapMixin):
return (408, {}, '')
else:
return (404, {}, '')
httplib.HTTPConnection._callback = server_response
self.computer_guid = self._getTestComputerId()
self.slap = slapos.slap.slap()
......@@ -488,18 +502,19 @@ class TestComputerPartition(SlapMixin):
def test_request_raises_later(self):
partition_id = 'PARTITION_01'
def server_response(self, path, method, body, header):
parsed_url = urlparse.urlparse(path.lstrip('/'))
parsed_qs = urlparse.parse_qs(parsed_url.query)
if parsed_url.path == 'registerComputerPartition' and \
'computer_reference' in parsed_qs and \
'computer_partition_reference' in parsed_qs:
if (parsed_url.path == 'registerComputerPartition' and
'computer_reference' in parsed_qs and
'computer_partition_reference' in parsed_qs):
slap_partition = slapos.slap.ComputerPartition(
parsed_qs['computer_reference'][0],
parsed_qs['computer_partition_reference'][0])
return (200, {}, xml_marshaller.xml_marshaller.dumps(slap_partition))
elif parsed_url.path == 'getComputerInformation' and \
'computer_id' in parsed_qs:
elif (parsed_url.path == 'getComputerInformation'
and 'computer_id' in parsed_qs):
slap_computer = slapos.slap.Computer(parsed_qs['computer_id'][0])
slap_computer._software_release_list = []
slap_partition = slapos.slap.ComputerPartition(
......@@ -511,6 +526,7 @@ class TestComputerPartition(SlapMixin):
return (408, {}, '')
else:
return (404, {}, '')
httplib.HTTPConnection._callback = server_response
self.computer_guid = self._getTestComputerId()
self.slap = slapos.slap.slap()
......@@ -530,18 +546,18 @@ class TestComputerPartition(SlapMixin):
partition_id = 'PARTITION_01'
requested_partition_id = 'PARTITION_02'
computer_guid = self._getTestComputerId()
def server_response(self, path, method, body, header):
parsed_url = urlparse.urlparse(path.lstrip('/'))
parsed_qs = urlparse.parse_qs(parsed_url.query)
if parsed_url.path == 'registerComputerPartition' and \
'computer_reference' in parsed_qs and \
'computer_partition_reference' in parsed_qs:
if (parsed_url.path == 'registerComputerPartition' and
'computer_reference' in parsed_qs and
'computer_partition_reference' in parsed_qs):
slap_partition = slapos.slap.ComputerPartition(
parsed_qs['computer_reference'][0],
parsed_qs['computer_partition_reference'][0])
return (200, {}, xml_marshaller.xml_marshaller.dumps(slap_partition))
elif parsed_url.path == 'getComputerInformation' and \
'computer_id' in parsed_qs:
elif (parsed_url.path == 'getComputerInformation' and 'computer_id' in parsed_qs):
slap_computer = slapos.slap.Computer(parsed_qs['computer_id'][0])
slap_computer._software_release_list = []
slap_partition = slapos.slap.ComputerPartition(
......@@ -557,6 +573,7 @@ class TestComputerPartition(SlapMixin):
return (200, {}, xml_marshaller.xml_marshaller.dumps(slap_partition))
else:
return (404, {}, '')
httplib.HTTPConnection._callback = server_response
self.slap = slapos.slap.slap()
self.slap.initializeConnection(self.server_url)
......@@ -584,9 +601,9 @@ class TestComputerPartition(SlapMixin):
def server_response(self, path, method, body, header):
parsed_url = urlparse.urlparse(path.lstrip('/'))
parsed_qs = urlparse.parse_qs(parsed_url.query)
if parsed_url.path == 'registerComputerPartition' and \
parsed_qs['computer_reference'][0] == computer_guid and \
parsed_qs['computer_partition_reference'][0] == partition_id:
if (parsed_url.path == 'registerComputerPartition' and
parsed_qs['computer_reference'][0] == computer_guid and
parsed_qs['computer_partition_reference'][0] == partition_id):
partition = slapos.slap.ComputerPartition(
computer_guid, partition_id)
return (200, {}, xml_marshaller.xml_marshaller.dumps(partition))
......@@ -596,8 +613,8 @@ class TestComputerPartition(SlapMixin):
computer_partition = self.slap.registerComputerPartition(
computer_guid, partition_id)
method = getattr(computer_partition, state)
self.assertRaises(slapos.slap.NotFoundError, method)
self.assertRaises(slapos.slap.NotFoundError,
getattr(computer_partition, state))
def test_available_new_ComputerPartition_raises(self):
"""
......@@ -639,9 +656,9 @@ class TestComputerPartition(SlapMixin):
def server_response(self, path, method, body, header):
parsed_url = urlparse.urlparse(path.lstrip('/'))
parsed_qs = urlparse.parse_qs(parsed_url.query)
if parsed_url.path == 'registerComputerPartition' and \
parsed_qs['computer_reference'][0] == computer_guid and \
parsed_qs['computer_partition_reference'][0] == partition_id:
if (parsed_url.path == 'registerComputerPartition' and
parsed_qs['computer_reference'][0] == computer_guid and
parsed_qs['computer_partition_reference'][0] == partition_id):
partition = slapos.slap.ComputerPartition(
computer_guid, partition_id)
return (200, {}, xml_marshaller.xml_marshaller.dumps(partition))
......@@ -650,9 +667,9 @@ class TestComputerPartition(SlapMixin):
# XXX: why do we have computer_id and not computer_reference?
# XXX: why do we have computer_partition_id and not
# computer_partition_reference?
if parsed_qs_body['computer_id'][0] == computer_guid and \
parsed_qs_body['computer_partition_id'][0] == partition_id and \
parsed_qs_body['error_log'][0] == 'some error':
if (parsed_qs_body['computer_id'][0] == computer_guid and
parsed_qs_body['computer_partition_id'][0] == partition_id and
parsed_qs_body['error_log'][0] == 'some error'):
return (200, {}, '')
return (404, {}, '')
......@@ -663,6 +680,7 @@ class TestComputerPartition(SlapMixin):
# XXX: Interface does not define return value
computer_partition.error('some error')
class TestSoftwareRelease(SlapMixin):
"""
Tests slap.SoftwareRelease class functionality
......@@ -707,18 +725,20 @@ class TestSoftwareRelease(SlapMixin):
def server_response(self, path, method, body, header):
parsed_url = urlparse.urlparse(path.lstrip('/'))
parsed_qs = urlparse.parse_qs(body)
if parsed_url.path == 'softwareReleaseError' and \
parsed_qs['computer_id'][0] == computer_guid and \
parsed_qs['url'][0] == software_release_uri and \
parsed_qs['error_log'][0] == 'some error':
if (parsed_url.path == 'softwareReleaseError' and
parsed_qs['computer_id'][0] == computer_guid and
parsed_qs['url'][0] == software_release_uri and
parsed_qs['error_log'][0] == 'some error'):
return (200, {}, '')
return (404, {}, '')
httplib.HTTPConnection._callback = server_response
software_release = self.slap.registerSoftwareRelease(software_release_uri)
software_release._computer_guid = computer_guid
software_release.error('some error')
class TestOpenOrder(SlapMixin):
def test_request_sends_request(self):
software_release_uri = 'http://server/new/' + self._getTestComputerId()
......@@ -726,12 +746,16 @@ class TestOpenOrder(SlapMixin):
self.slap.initializeConnection(self.server_url)
# XXX: Interface lack registerOpenOrder method declaration
open_order = self.slap.registerOpenOrder()
def server_response(self, path, method, body, header):
parsed_url = urlparse.urlparse(path.lstrip('/'))
if parsed_url.path == 'requestComputerPartition':
raise RequestWasCalled
httplib.HTTPConnection._callback = server_response
self.assertRaises(RequestWasCalled, open_order.request, software_release_uri, 'myrefe')
self.assertRaises(RequestWasCalled,
open_order.request,
software_release_uri, 'myrefe')
def test_request_not_raises(self):
software_release_uri = 'http://server/new/' + self._getTestComputerId()
......@@ -748,13 +772,16 @@ class TestOpenOrder(SlapMixin):
self.slap.initializeConnection(self.server_url)
# XXX: Interface lack registerOpenOrder method declaration
open_order = self.slap.registerOpenOrder()
def server_response(self, path, method, body, header):
return (408, {}, '')
httplib.HTTPConnection._callback = server_response
computer_partition = open_order.request(software_release_uri, 'myrefe')
self.assertIsInstance(computer_partition, slapos.slap.ComputerPartition)
self.assertRaises(slapos.slap.ResourceNotReady, computer_partition.getId)
self.assertRaises(slapos.slap.ResourceNotReady,
computer_partition.getId)
def test_request_fullfilled_work(self):
software_release_uri = 'http://server/new/' + self._getTestComputerId()
......@@ -764,12 +791,14 @@ class TestOpenOrder(SlapMixin):
open_order = self.slap.registerOpenOrder()
computer_guid = self._getTestComputerId()
requested_partition_id = 'PARTITION_01'
def server_response(self, path, method, body, header):
from slapos.slap.slap import SoftwareInstance
slap_partition = SoftwareInstance(
slap_computer_id=computer_guid,
slap_computer_partition_id=requested_partition_id)
return (200, {}, xml_marshaller.xml_marshaller.dumps(slap_partition))
httplib.HTTPConnection._callback = server_response
computer_partition = open_order.request(software_release_uri, 'myrefe')
......
slapos/tests/slapformat.py
View file @ 5c613eb5
......@@ -45,9 +45,11 @@ USER_LIST = []
GROUP_LIST = []
INTERFACE_DICT = {}
class FakeConfig:
pass
class TestLoggerHandler(logging.Handler):
def __init__(self, *args, **kwargs):
self.bucket = []
......@@ -56,6 +58,7 @@ class TestLoggerHandler(logging.Handler):
def emit(self, record):
self.bucket.append(record.msg)
class FakeCallAndRead:
def __init__(self):
self.external_command_list = []
......@@ -89,6 +92,7 @@ class FakeCallAndRead:
self.external_command_list.append(' '.join(argument_list))
return retval
class LoggableWrapper:
def __init__(self, logger, name):
self.__logger = logger
......@@ -99,11 +103,13 @@ class LoggableWrapper:
'%s=%r' % (x, y) for x, y in kwargs.iteritems()]
self.__logger.debug('%s(%s)' % (self.__name, ', '.join(arg_list)))
class TimeMock:
@classmethod
def sleep(self, seconds):
return
class GrpMock:
@classmethod
def getgrnam(self, name):
......@@ -112,6 +118,7 @@ class GrpMock:
return True
raise KeyError
class PwdMock:
@classmethod
def getpwnam(self, name):
......@@ -123,6 +130,7 @@ class PwdMock:
return result
raise KeyError
class NetifacesMock:
@classmethod
def ifaddresses(self, name):
......@@ -136,12 +144,14 @@ class NetifacesMock:
global INTERFACE_DICT
return INTERFACE_DICT.keys()
class SlapformatMixin(unittest.TestCase):
# keep big diffs
maxDiff = None
def patchNetifaces(self):
self.netifaces = NetifacesMock()
self.saved_netifaces = dict()
self.saved_netifaces = {}
for fake in vars(NetifacesMock):
self.saved_netifaces[fake] = getattr(netifaces, fake, None)
setattr(netifaces, fake, getattr(self.netifaces, fake))
......@@ -152,7 +162,7 @@ class SlapformatMixin(unittest.TestCase):
del self.saved_netifaces
def patchPwd(self):
self.saved_pwd = dict()
self.saved_pwd = {}
for fake in vars(PwdMock):
self.saved_pwd[fake] = getattr(pwd, fake, None)
setattr(pwd, fake, getattr(PwdMock, fake))
......@@ -163,7 +173,7 @@ class SlapformatMixin(unittest.TestCase):
del self.saved_pwd
def patchTime(self):
self.saved_time = dict()
self.saved_time = {}
for fake in vars(TimeMock):
self.saved_time[fake] = getattr(time, fake, None)
setattr(time, fake, getattr(TimeMock, fake))
......@@ -174,7 +184,7 @@ class SlapformatMixin(unittest.TestCase):
del self.saved_time
def patchGrp(self):
self.saved_grp = dict()
self.saved_grp = {}
for fake in vars(GrpMock):
self.saved_grp[fake] = getattr(grp, fake, None)
setattr(grp, fake, getattr(GrpMock, fake))
......@@ -185,7 +195,7 @@ class SlapformatMixin(unittest.TestCase):
del self.saved_grp
def patchOs(self, logger):
self.saved_os = dict()
self.saved_os = {}
for fake in ['mkdir', 'chown', 'chmod', 'makedirs']:
self.saved_os[fake] = getattr(os, fake, None)
f = LoggableWrapper(logger, fake)
......@@ -231,6 +241,7 @@ class SlapformatMixin(unittest.TestCase):
self.restoreNetifaces()
slapos.format.callAndRead = self.real_callAndRead
class TestComputer(SlapformatMixin):
def test_getAddress_empty_computer(self):
computer = slapos.format.Computer('computer')
......@@ -276,8 +287,7 @@ class TestComputer(SlapformatMixin):
"makedirs('/software_root', 493)",
"chmod('/software_root', 493)"],
self.test_result.bucket)
self.assertEqual([
'ip addr list bridge',],
self.assertEqual(['ip addr list bridge'],
self.fakeCallAndRead.external_command_list)
@unittest.skip("Not implemented")
......@@ -377,7 +387,7 @@ class TestComputer(SlapformatMixin):
partition = slapos.format.Partition('partition', '/part_path',
slapos.format.User('testuser'), [], None)
global USER_LIST
USER_LIST=['testuser']
USER_LIST = ['testuser']
partition.tap = slapos.format.Tap('tap')
computer.partition_list = [partition]
global INTERFACE_DICT
......@@ -488,6 +498,7 @@ class TestComputer(SlapformatMixin):
],
self.fakeCallAndRead.external_command_list)
class TestPartition(SlapformatMixin):
def test_createPath(self):
......@@ -513,6 +524,7 @@ class TestPartition(SlapformatMixin):
self.test_result.bucket
)
class TestUser(SlapformatMixin):
def test_create(self):
user = slapos.format.User('doesnotexistsyet')
......
slapos/tests/slapgrid.py
View file @ 5c613eb5
......@@ -98,6 +98,7 @@ chmod 755 etc/service/daemon &&
touch worked
"""
class BasicMixin:
def setUp(self):
self._tempdir = tempfile.mkdtemp()
......@@ -125,8 +126,10 @@ class BasicMixin:
develop=develop,
logger=logging.getLogger())
# monkey patch buildout bootstrap
def dummy(*args, **kw):
pass
slapos.grid.utils.bootstrapBuildout = dummy
def launchSlapgrid(self, develop=False):
......@@ -186,6 +189,7 @@ class TestBasicSlapgridCP(BasicMixin, unittest.TestCase):
os.mkdir(self.instance_root)
self.assertRaises(socket.error, self.grid.processComputerPartitionList)
class MasterMixin(BasicMixin):
def _patchHttplib(self):
......@@ -218,6 +222,7 @@ class MasterMixin(BasicMixin):
def _unmock_sleep(self):
time.sleep = self.real_sleep
def setUp(self):
self._patchHttplib()
self._mock_sleep()
......@@ -310,8 +315,8 @@ class ComputerForTest:
parsed_qs = urlparse.parse_qs(parsed_url.query)
else:
parsed_qs = urlparse.parse_qs(body)
if parsed_url.path == 'getFullComputerInformation' and \
'computer_id' in parsed_qs:
if (parsed_url.path == 'getFullComputerInformation'
and 'computer_id' in parsed_qs):
slap_computer = self.getComputer(parsed_qs['computer_id'][0])
return (200, {}, xml_marshaller.xml_marshaller.dumps(slap_computer))
if method == 'POST' and 'computer_partition_id' in parsed_qs:
......@@ -332,9 +337,13 @@ class ComputerForTest:
if parsed_url.path == 'softwareInstanceBang':
return (200, {}, '')
if parsed_url.path == 'softwareInstanceError':
instance.error_log = '\n'.join([line for line \
in parsed_qs['error_log'][0].splitlines()
if 'dropPrivileges' not in line])
instance.error_log = '\n'.join(
[
line
for line in parsed_qs['error_log'][0].splitlines()
if 'dropPrivileges' not in line
]
)
instance.error = True
return (200, {}, '')
......@@ -345,9 +354,13 @@ class ComputerForTest:
if parsed_url.path == 'buildingSoftwareRelease':
return (200, {}, '')
if parsed_url.path == 'softwareReleaseError':
software.error_log = '\n'.join([line for line \
in parsed_qs['error_log'][0].splitlines()
if 'dropPrivileges' not in line])
software.error_log = '\n'.join(
[
line
for line in parsed_qs['error_log'][0].splitlines()
if 'dropPrivileges' not in line
]
)
software.error = True
return (200, {}, '')
......@@ -382,7 +395,7 @@ class InstanceForTest:
partition._software_release_document = self.getSoftwareRelease()
partition._requested_state = self.requested_state
if self.software is not None:
if self.timestamp is not None :
if self.timestamp is not None:
partition._parameter_dict = {'timestamp': self.timestamp}
return partition
......@@ -394,7 +407,8 @@ class InstanceForTest:
sr = slapos.slap.SoftwareRelease()
sr._software_release = self.software.name
return sr
else: return None
else:
return None
def setPromise(self, promise_name, promise_content):
"""
......@@ -415,10 +429,11 @@ class InstanceForTest:
self.certificate = str(random.random())
open(self.cert_file, 'w').write(self.certificate)
self.key_file = os.path.join(certificate_repository_path,
"%s.key" % self.name)
'%s.key' % self.name)
self.key = str(random.random())
open(self.key_file, 'w').write(self.key)
class SoftwareForTest:
"""
Class to prepare and simulate software.
......@@ -448,7 +463,6 @@ class SoftwareForTest:
software._requested_state = self.requested_state
return software
def setTemplateCfg(self, template="""[buildout]"""):
"""
Set template.cfg
......@@ -467,9 +481,8 @@ touch worked"""):
"""
Set a periodicity file
"""
open(os.path.join(self.srdir, 'periodicity'), 'w').write(
"""%s""" % (periodicity))
with open(os.path.join(self.srdir, 'periodicity'), 'w') as fout:
fout.write(str(periodicity))
class TestSlapgridCPWithMaster(MasterMixin, unittest.TestCase):
......@@ -548,7 +561,6 @@ class TestSlapgridCPWithMaster(MasterMixin, unittest.TestCase):
'startedComputerPartition'])
self.assertEqual(partition.state, 'started')
def test_one_partition_started_stopped(self):
computer = ComputerForTest(self.software_root, self.instance_root)
instance = computer.instance_list[0]
......@@ -683,7 +695,6 @@ exit 1
'softwareInstanceError'])
self.assertEqual(instance.state, 'started')
def test_one_partition_stopped_started(self):
computer = ComputerForTest(self.software_root, self.instance_root)
instance = computer.instance_list[0]
......@@ -756,11 +767,11 @@ class TestSlapgridCPWithMasterWatchdog(MasterMixin, unittest.TestCase):
# Prepare watchdog
self.watchdog_banged = os.path.join(self._tempdir, 'watchdog_banged')
watchdog_path = os.path.join(self._tempdir, 'watchdog')
open(watchdog_path, 'w').write(
WATCHDOG_TEMPLATE % {
open(watchdog_path, 'w').write(WATCHDOG_TEMPLATE % {
'python_path': sys.executable,
'sys_path': sys.path,
'watchdog_banged': self.watchdog_banged})
'watchdog_banged': self.watchdog_banged
})
os.chmod(watchdog_path, 0o755)
self.grid.watchdog_path = watchdog_path
slapos.grid.slapgrid.WATCHDOG_PATH = watchdog_path
......@@ -855,7 +866,6 @@ class TestSlapgridCPWithMasterWatchdog(MasterMixin, unittest.TestCase):
time.sleep(0.1)
self.assertFalse(os.path.exists(self.watchdog_banged))
def test_watched_by_watchdog_bang(self):
"""
Test that a process going to fatal or exited mode in supervisord
......@@ -877,8 +887,8 @@ class TestSlapgridCPWithMasterWatchdog(MasterMixin, unittest.TestCase):
instance.sequence = []
instance.header_list = []
headers = {'eventname': event}
payload = "processname:%s groupname:%s from_state:RUNNING"\
% ('daemon'+getWatchdogID(), instance.name)
payload = 'processname:%s groupname:%s from_state:RUNNING' % (
'daemon' + getWatchdogID(), instance.name)
watchdog.handle_event(headers, payload)
self.assertEqual(instance.sequence, ['softwareInstanceBang'])
self.assertEqual(instance.header_list[0]['key'], instance.key)
......@@ -901,12 +911,11 @@ class TestSlapgridCPWithMasterWatchdog(MasterMixin, unittest.TestCase):
'PROCESS_STATE_BACKOFF', 'PROCESS_STATE_STOPPED']:
computer.sequence = []
headers = {'eventname': event}
payload = "processname:%s groupname:%s from_state:RUNNING"\
% ('daemon'+getWatchdogID(), instance.name)
payload = 'processname:%s groupname:%s from_state:RUNNING' % (
'daemon' + getWatchdogID(), instance.name)
watchdog.handle_event(headers, payload)
self.assertEqual(instance.sequence, [])
def test_not_watched_by_watchdog_do_not_bang(self):
"""
Test that a process going to fatal or exited mode in supervisord
......@@ -951,7 +960,6 @@ class TestSlapgridCPPartitionProcessing(MasterMixin, unittest.TestCase):
self.assertEqual(instance.sequence,
['availableComputerPartition', 'stoppedComputerPartition'])
def test_partition_timestamp_develop(self):
computer = ComputerForTest(self.software_root, self.instance_root)
instance = computer.instance_list[0]
......@@ -990,7 +998,6 @@ class TestSlapgridCPPartitionProcessing(MasterMixin, unittest.TestCase):
self.assertEqual(instance.sequence,
['availableComputerPartition', 'stoppedComputerPartition'])
def test_partition_timestamp_new_timestamp(self):
computer = ComputerForTest(self.software_root, self.instance_root)
instance = computer.instance_list[0]
......@@ -1003,7 +1010,7 @@ class TestSlapgridCPPartitionProcessing(MasterMixin, unittest.TestCase):
self.assertItemsEqual(os.listdir(partition),
['.timestamp', 'buildout.cfg', 'software_release', 'worked'])
self.assertItemsEqual(os.listdir(self.software_root), [instance.software.software_hash])
instance.timestamp = str(int(timestamp)+1)
instance.timestamp = str(int(timestamp) + 1)
self.assertEqual(self.launchSlapgrid(), slapgrid.SLAPGRID_SUCCESS)
self.assertEqual(self.launchSlapgrid(), slapgrid.SLAPGRID_SUCCESS)
self.assertEqual(computer.sequence,
......@@ -1032,7 +1039,6 @@ class TestSlapgridCPPartitionProcessing(MasterMixin, unittest.TestCase):
'stoppedComputerPartition', 'getFullComputerInformation',
'availableComputerPartition', 'stoppedComputerPartition'])
def test_partition_periodicity_remove_timestamp(self):
"""
Check that if periodicity forces run of buildout for a partition, it
......@@ -1061,7 +1067,6 @@ class TestSlapgridCPPartitionProcessing(MasterMixin, unittest.TestCase):
self.assertItemsEqual(os.listdir(partition),
['.timestamp', 'buildout.cfg', 'software_release', 'worked'])
def test_partition_periodicity_is_not_overloaded_if_forced(self):
"""
If periodicity file in software directory but periodicity is forced
......@@ -1093,7 +1098,6 @@ class TestSlapgridCPPartitionProcessing(MasterMixin, unittest.TestCase):
['getFullComputerInformation', 'availableComputerPartition',
'startedComputerPartition', 'getFullComputerInformation'])
def test_one_partition_periodicity_from_file_does_not_disturb_others(self):
"""
If time between last processing of instance and now is superior
......@@ -1109,7 +1113,7 @@ class TestSlapgridCPPartitionProcessing(MasterMixin, unittest.TestCase):
"""
computer = ComputerForTest(self.software_root, self.instance_root, 20, 20)
instance0 = computer.instance_list[0]
timestamp = str(int(time.time()-5))
timestamp = str(int(time.time() - 5))
instance0.timestamp = timestamp
instance0.requested_state = 'started'
for instance in computer.instance_list[1:]:
......@@ -1142,7 +1146,6 @@ class TestSlapgridCPPartitionProcessing(MasterMixin, unittest.TestCase):
last_runtime)
self.assertNotEqual(wanted_periodicity, self.grid.maximum_periodicity)
def test_one_partition_stopped_is_not_processed_after_periodicity(self):
"""
Check that periodicity doesn't force processing a partition if it is not
......@@ -1150,7 +1153,7 @@ class TestSlapgridCPPartitionProcessing(MasterMixin, unittest.TestCase):
"""
computer = ComputerForTest(self.software_root, self.instance_root, 20, 20)
instance0 = computer.instance_list[0]
timestamp = str(int(time.time()-5))
timestamp = str(int(time.time() - 5))
instance0.timestamp = timestamp
for instance in computer.instance_list[1:]:
instance.software = \
......@@ -1171,13 +1174,12 @@ class TestSlapgridCPPartitionProcessing(MasterMixin, unittest.TestCase):
time.sleep(1)
self.launchSlapgrid()
self.assertEqual(instance0.sequence,
['availableComputerPartition', 'stoppedComputerPartition',
])
['availableComputerPartition', 'stoppedComputerPartition'])
for instance in computer.instance_list[1:]:
self.assertEqual(instance.sequence,
['availableComputerPartition', 'stoppedComputerPartition'])
self.assertEqual(
os.path.getmtime(os.path.join(instance0.partition_path, '.timestamp')),
self.assertEqual(os.path.getmtime(os.path.join(instance0.partition_path,
'.timestamp')),
last_runtime)
self.assertNotEqual(wanted_periodicity, self.grid.maximum_periodicity)
......@@ -1188,7 +1190,7 @@ class TestSlapgridCPPartitionProcessing(MasterMixin, unittest.TestCase):
"""
computer = ComputerForTest(self.software_root, self.instance_root, 20, 20)
instance0 = computer.instance_list[0]
timestamp = str(int(time.time()-5))
timestamp = str(int(time.time() - 5))
instance0.timestamp = timestamp
instance0.requested_state = 'stopped'
for instance in computer.instance_list[1:]:
......@@ -1211,13 +1213,12 @@ class TestSlapgridCPPartitionProcessing(MasterMixin, unittest.TestCase):
instance0.requested_state = 'destroyed'
self.launchSlapgrid()
self.assertEqual(instance0.sequence,
['availableComputerPartition', 'stoppedComputerPartition',
])
['availableComputerPartition', 'stoppedComputerPartition'])
for instance in computer.instance_list[1:]:
self.assertEqual(instance.sequence,
['availableComputerPartition', 'stoppedComputerPartition'])
self.assertEqual(
os.path.getmtime(os.path.join(instance0.partition_path, '.timestamp')),
self.assertEqual(os.path.getmtime(os.path.join(instance0.partition_path,
'.timestamp')),
last_runtime)
self.assertNotEqual(wanted_periodicity, self.grid.maximum_periodicity)
......@@ -1365,7 +1366,6 @@ class TestSlapgridUsageReport(MasterMixin, unittest.TestCase):
'destroyedComputerPartition'])
self.assertEqual(instance.state, 'destroyed')
def test_partition_list_is_complete_if_empty_destroyed_partition(self):
"""
Test that an empty partition with destroyed state but with SR informations
......@@ -1457,7 +1457,6 @@ class TestSlapgridUsageReport(MasterMixin, unittest.TestCase):
['getFullComputerInformation'])
self.assertEqual('started', instance.state)
def test_slapgrid_instance_ignore_free_instance(self):
"""
Test than a free instance (so in "destroyed" state, but empty, without
......@@ -1495,7 +1494,6 @@ class TestSlapgridUsageReport(MasterMixin, unittest.TestCase):
self.assertEqual(computer.sequence, ['getFullComputerInformation'])
class TestSlapgridSoftwareRelease(MasterMixin, unittest.TestCase):
def test_one_software_buildout_fail_is_correctly_logged(self):
"""
......@@ -1517,6 +1515,7 @@ echo %s; echo %s; exit 42""" % (line1, line2))
self.assertIn(line2, software.error_log)
self.assertIn('Failed to run buildout', software.error_log)
class SlapgridInitialization(unittest.TestCase):
"""
"Abstract" class setting setup and teardown for TestSlapgridArgumentTuple
......@@ -1558,6 +1557,7 @@ buildout = /path/to/buildout/binary
self.signature_key_file_descriptor.close()
shutil.rmtree(self.certificate_repository_path, True)
class TestSlapgridArgumentTuple(SlapgridInitialization):
"""
Test suite about arguments given to slapgrid command.
......@@ -1586,8 +1586,8 @@ class TestSlapgridArgumentTuple(SlapgridInitialization):
Raises if the signature_private_key_file does not exists.
"""
parser = parseArgumentTupleAndReturnSlapgridObject
argument_tuple = ("--signature_private_key_file", "/non/exists/path") + \
self.default_arg_tuple
argument_tuple = ("--signature_private_key_file",
"/non/exists/path") + self.default_arg_tuple
self.assertRaisesRegexp(RuntimeError,
"File '/non/exists/path' does not exist.",
parser, *argument_tuple)
......@@ -1599,8 +1599,7 @@ class TestSlapgridArgumentTuple(SlapgridInitialization):
"""
parser = parseArgumentTupleAndReturnSlapgridObject
argument_tuple = ("--signature_private_key_file",
self.signature_key_file_descriptor.name) + \
self.default_arg_tuple
self.signature_key_file_descriptor.name) + self.default_arg_tuple
slapgrid_object = parser(*argument_tuple)[0]
self.assertEquals(self.signature_key_file_descriptor.name,
slapgrid_object.signature_private_key_file)
......@@ -1639,6 +1638,7 @@ class TestSlapgridArgumentTuple(SlapgridInitialization):
slapgrid_object = parser('--maximum-periodicity', '40', *self.default_arg_tuple)[0]
self.assertTrue(slapgrid_object.force_periodicity)
class TestSlapgridConfigurationFile(SlapgridInitialization):
def test_upload_binary_cache_blacklist(self):
......@@ -1777,7 +1777,7 @@ class TestSlapgridCPWithMasterPromise(MasterMixin, unittest.TestCase):
self.assertNotEqual('started', instance.state)
def test_one_succeeding_promise(self):
computer = ComputerForTest(self.software_root,self.instance_root)
computer = ComputerForTest(self.software_root, self.instance_root)
instance = computer.instance_list[0]
instance.requested_state = 'started'
self.fake_waiting_time = 0.1
......@@ -1820,7 +1820,6 @@ class TestSlapgridCPWithMasterPromise(MasterMixin, unittest.TestCase):
self.assertTrue(instance.error)
self.assertIsNone(instance.state)
def test_timeout_works(self):
computer = ComputerForTest(self.software_root, self.instance_root)
instance = computer.instance_list[0]
......@@ -1915,8 +1914,8 @@ class TestSlapgridCPWithMasterPromise(MasterMixin, unittest.TestCase):
fi
exit 0""" % {
'worked_file': worked_file,
'lockfile': lockfile
})
'lockfile': lockfile}
)
instance.setPromise('promise_%d' % i, promise)
self.assertEqual(self.grid.processComputerPartitionList(),
......
slapos/tests/slapobject.py
View file @ 5c613eb5
......@@ -53,6 +53,7 @@ originalBootstrapBuildout = utils.bootstrapBuildout
originalLaunchBuildout = utils.launchBuildout
originalUploadSoftwareRelease = SlapObject.Software.uploadSoftwareRelease
class TestSoftwareSlapObject(BasicMixin, unittest.TestCase):
"""
Test for Software class.
......@@ -121,23 +122,21 @@ class TestSoftwareSlapObject(BasicMixin, unittest.TestCase):
Check if the networkcache parameters are not propagated if they are not
available.
"""
software = SlapObject.Software(
url='http://example.com/software.cfg',
software = SlapObject.Software(url='http://example.com/software.cfg',
software_root=self.software_root,
buildout=self.buildout,
logger=logging.getLogger())
software.install()
command_list = FakeCallAndRead.external_command_list
self.assertFalse('buildout:networkcache-section=networkcache'
in command_list)
self.assertFalse('networkcache:signature-private-key-file=%s' %
self.signature_private_key_file in command_list)
self.assertFalse('networkcache:upload-cache-url=%s' % self.upload_cache_url
in command_list)
self.assertFalse('networkcache:upload-dir-url=%s' % self.upload_dir_url
in command_list)
self.assertNotIn('buildout:networkcache-section=networkcache', command_list)
self.assertNotIn('networkcache:signature-private-key-file=%s' %
self.signature_private_key_file,
command_list)
self.assertNotIn('networkcache:upload-cache-url=%s' % self.upload_cache_url,
command_list)
self.assertNotIn('networkcache:upload-dir-url=%s' % self.upload_dir_url,
command_list)
# XXX-Cedric: do the same with upload
def test_software_install_networkcache_upload_blacklist(self):
......@@ -146,9 +145,12 @@ class TestSoftwareSlapObject(BasicMixin, unittest.TestCase):
"""
def fakeBuildout(*args, **kw):
pass
SlapObject.Software._install_from_buildout = fakeBuildout
def fake_upload_network_cached(*args, **kw):
self.assertFalse(True)
networkcache.upload_network_cached = fake_upload_network_cached
upload_to_binary_cache_url_blacklist = ["http://example.com"]
......@@ -165,7 +167,7 @@ class TestSoftwareSlapObject(BasicMixin, unittest.TestCase):
shacache_key_file=self.shacache_key_file,
shadir_cert_file=self.shadir_cert_file,
shadir_key_file=self.shadir_key_file,
upload_to_binary_cache_url_blacklist=\
upload_to_binary_cache_url_blacklist=
upload_to_binary_cache_url_blacklist,
)
software.install()
......@@ -178,10 +180,11 @@ class TestSoftwareSlapObject(BasicMixin, unittest.TestCase):
def fakeBuildout(*args, **kw):
pass
SlapObject.Software._install_from_buildout = fakeBuildout
def fakeUploadSoftwareRelease(*args, **kw):
self.uploaded = True
SlapObject.Software.uploadSoftwareRelease = fakeUploadSoftwareRelease
SlapObject.Software.uploadSoftwareRelease = fakeUploadSoftwareRelease
upload_to_binary_cache_url_blacklist = ["http://anotherexample.com"]
......@@ -199,7 +202,7 @@ class TestSoftwareSlapObject(BasicMixin, unittest.TestCase):
shacache_key_file=self.shacache_key_file,
shadir_cert_file=self.shadir_cert_file,
shadir_key_file=self.shadir_key_file,
upload_to_binary_cache_url_blacklist=\
upload_to_binary_cache_url_blacklist=
upload_to_binary_cache_url_blacklist,
)
software.install()
......
slapos/tests/slapproxy.py
View file @ 5c613eb5
......@@ -31,27 +31,31 @@ import ConfigParser
import os
import logging
import shutil
import slapos.proxy
import slapos.proxy.views as views
import slapos.slap.slap
import tempfile
import unittest
import xml_marshaller
import slapos.proxy
import slapos.proxy.views as views
import slapos.slap.slap
class WrongFormat(Exception):
pass
class ProxyOption:
class ProxyOption(object):
"""
Will simulate options given to slapproxy
"""
def __init__(self, proxy_db):
self.verbose=True
self.database_uri=proxy_db
self.verbose = True
self.database_uri = proxy_db
self.console = False
self.log_file = None
class BasicMixin:
class BasicMixin(object):
def setUp(self):
"""
Will set files and start slapproxy
......@@ -78,8 +82,8 @@ computer_id = computer
host = 127.0.0.1
port = 8080
database_uri = %(tempdir)s/lib/proxy.db
"""% dict (tempdir = self._tempdir, proxyaddr = self.proxyaddr))
for directory in ("opt", "srv", "lib"):
""" % {'tempdir': self._tempdir, 'proxyaddr': self.proxyaddr})
for directory in ['opt', 'srv', 'lib']:
path = os.path.join(self._tempdir, directory)
os.mkdir(path)
......@@ -99,26 +103,30 @@ database_uri = %(tempdir)s/lib/proxy.db
views.app.config['port'] = conf.port
self.app = views.app.test_client()
def add_free_partition (self, partition_amount):
def add_free_partition(self, partition_amount):
"""
Will simulate a slapformat first run
and create "partition_amount" partitions
"""
computer_dict = {'reference':self.computer_id,
'address':'123.456.789',
'netmask':'fffffffff',
'partition_list':[]}
for i in range(0, partition_amount):
partition_example = {'reference':'slappart%s' %i,
'address_list':[]}
computer_dict = {
'reference': self.computer_id,
'address': '123.456.789',
'netmask': 'fffffffff',
'partition_list': [],
}
for i in range(partition_amount):
partition_example = {
'reference': 'slappart%s' % i,
'address_list': [],
}
computer_dict['partition_list'].append(partition_example)
request_dict = { 'computer_id': self.computer_id,
request_dict = {
'computer_id': self.computer_id,
'xml': xml_marshaller.xml_marshaller.dumps(computer_dict),
}
self.app.post('/loadComputerConfigurationFromXML',
data = request_dict)
data=request_dict)
def tearDown(self):
"""
......@@ -132,46 +140,42 @@ class TestInformation(BasicMixin, unittest.TestCase):
Test Basic response of slapproxy
"""
def test_getComputerInformation (self):
def test_getComputerInformation(self):
"""
Check that getComputerInformation return a Computer
and database is generated
"""
rv = self.app.get('/getComputerInformation?computer_id='
+self.computer_id)
rv = self.app.get('/getComputerInformation?computer_id=%s' % self.computer_id)
self.assertIsInstance(
xml_marshaller.xml_marshaller.loads(rv.data),
slapos.slap.Computer)
self.assertTrue(os.path.exists(self.proxy_db))
def test_getFullComputerInformation (self):
def test_getFullComputerInformation(self):
"""
Check that getFullComputerInformation return a Computer
and database is generated
"""
rv = self.app.get('/getFullComputerInformation?computer_id='
+ self.computer_id)
rv = self.app.get('/getFullComputerInformation?computer_id=%s' % self.computer_id)
self.assertIsInstance(
xml_marshaller.xml_marshaller.loads(rv.data),
slapos.slap.Computer)
self.assertTrue(os.path.exists(self.proxy_db))
def test_getComputerInformation_wrong_computer (self):
def test_getComputerInformation_wrong_computer(self):
"""
Test that computer information won't be given to a requester different
from the one specified
"""
with self.assertRaises(slapos.slap.NotFoundError):
self.app.get('/getComputerInformation?computer_id='
+ self.computer_id + '42')
self.app.get('/getComputerInformation?computer_id=%s42' % self.computer_id)
def test_partition_are_empty (self):
def test_partition_are_empty(self):
"""
Test that empty partition are empty :)
"""
self.add_free_partition(10)
rv = self.app.get('/getFullComputerInformation?computer_id='
+ self.computer_id)
rv = self.app.get('/getFullComputerInformation?computer_id=%s' % self.computer_id)
computer = xml_marshaller.xml_marshaller.loads(rv.data)
for slap_partition in computer._computer_partition_list:
self.assertIsNone(slap_partition._software_release_document)
......@@ -179,7 +183,6 @@ class TestInformation(BasicMixin, unittest.TestCase):
self.assertEqual(slap_partition._need_modification, 0)
class MasterMixin(BasicMixin):
"""
Define advanced tool for test proxy simulating behavior slap library tools
......@@ -197,11 +200,12 @@ class MasterMixin(BasicMixin):
partition_parameter_kw = {}
if filter_kw is None:
filter_kw = {}
# Let enforce a default software type
# Let's enforce a default software type
if software_type is None:
software_type = 'default'
request_dict = { 'computer_id': self.computer_id,
request_dict = {
'computer_id': self.computer_id,
'computer_partition_id': partition_id,
'software_release': software_release,
'software_type': software_type,
......@@ -213,7 +217,7 @@ class MasterMixin(BasicMixin):
'state': xml_marshaller.xml_marshaller.dumps(state),
}
rv = self.app.post('/requestComputerPartition',
data = request_dict)
data=request_dict)
xml = rv.data
try:
software_instance = xml_marshaller.xml_marshaller.loads(xml)
......@@ -222,8 +226,7 @@ class MasterMixin(BasicMixin):
computer_partition = slapos.slap.ComputerPartition(
software_instance.slap_computer_id,
software_instance.slap_computer_partition_id,
)
software_instance.slap_computer_partition_id)
if shared:
computer_partition._synced = True
computer_partition._connection_dict = getattr(software_instance,
......@@ -232,32 +235,30 @@ class MasterMixin(BasicMixin):
'_parameter_dict', None)
return computer_partition
def setConnectionDict(self, partition_id,
connection_dict, slave_reference=None):
self.app.post('/setComputerPartitionConnectionXml', data = {
self.app.post('/setComputerPartitionConnectionXml', data={
'computer_id': self.computer_id,
'computer_partition_id': partition_id,
'connection_xml': xml_marshaller.xml_marshaller.dumps(connection_dict),
'slave_reference': slave_reference})
def getPartitionInformation(self,computer_partition_id):
def getPartitionInformation(self, computer_partition_id):
"""
Return computer information as stored in proxy for corresponding id
"""
rv = self.app.get('/getFullComputerInformation?computer_id='
+ self.computer_id)
rv = self.app.get('/getFullComputerInformation?computer_id=%s' % self.computer_id)
computer = xml_marshaller.xml_marshaller.loads(rv.data)
for instance in computer._computer_partition_list:
if instance._partition_id == computer_partition_id:
return instance
class TestRequest (MasterMixin, unittest.TestCase):
class TestRequest(MasterMixin, unittest.TestCase):
"""
Set of tests for requests
"""
def test_two_request_one_partition_free (self):
def test_two_request_one_partition_free(self):
"""
Since slapproxy does not implement scope, providing two partition_id
values will still succeed, even if only one partition is available.
......@@ -270,7 +271,7 @@ class TestRequest (MasterMixin, unittest.TestCase):
'Maria', 'slappart3'),
slapos.slap.ComputerPartition)
def test_two_request_two_partition_free (self):
def test_two_request_two_partition_free(self):
"""
If two requests are made with two available partition
both will succeed
......@@ -283,7 +284,7 @@ class TestRequest (MasterMixin, unittest.TestCase):
'Maria', 'slappart3'),
slapos.slap.ComputerPartition)
def test_two_same_request_from_one_partition (self):
def test_two_same_request_from_one_partition(self):
"""
Request will return same partition for two equal requests
"""
......@@ -292,7 +293,7 @@ class TestRequest (MasterMixin, unittest.TestCase):
self.request('http://sr//', None, 'Maria', 'slappart2').__dict__,
self.request('http://sr//', None, 'Maria', 'slappart2').__dict__)
def test_two_requests_with_different_parameters_but_same_reference (self):
def test_two_requests_with_different_parameters_but_same_reference(self):
"""
Request will return same partition for two different requests but will
only update parameters
......@@ -302,25 +303,25 @@ class TestRequest (MasterMixin, unittest.TestCase):
wanted_domain2 = 'carzy.org'
request1 = self.request('http://sr//', None, 'Maria', 'slappart2',
partition_parameter_kw = {'domain':wanted_domain1})
partition_parameter_kw={'domain': wanted_domain1})
request1_dict = request1.__dict__
requested_result1 = self.getPartitionInformation(
request1_dict['_partition_id'])
request2 = self.request('http://sr1//', 'Papa', 'Maria', 'slappart2',
partition_parameter_kw = {'domain':wanted_domain2})
partition_parameter_kw={'domain': wanted_domain2})
request2_dict = request2.__dict__
requested_result2 = self.getPartitionInformation(
request2_dict['_partition_id'])
# Test we received same partition
for key in request1_dict:
self.assertEqual(request1_dict[key],request2_dict[key])
self.assertEqual(request1_dict[key], request2_dict[key])
# Test that only parameters changed
for key in requested_result2.__dict__ :
if not key in ('_parameter_dict',
'_software_release_document'):
for key in requested_result2.__dict__:
if key not in ['_parameter_dict',
'_software_release_document']:
self.assertEqual(requested_result2.__dict__[key],
requested_result1.__dict__[key])
elif key in ('_software_release_document'):
elif key in ['_software_release_document']:
self.assertEqual(requested_result2.__dict__[key].__dict__,
requested_result1.__dict__[key].__dict__)
#Test parameters where set correctly
......@@ -329,7 +330,7 @@ class TestRequest (MasterMixin, unittest.TestCase):
self.assertEqual(wanted_domain2,
requested_result2._parameter_dict['domain'])
def test_two_different_request_from_two_partition (self):
def test_two_different_request_from_two_partition(self):
"""
Since slapproxy does not implement scope, two request with
different partition_id will still return the same partition.
......@@ -339,7 +340,7 @@ class TestRequest (MasterMixin, unittest.TestCase):
self.request('http://sr//', None, 'Maria', 'slappart2').__dict__,
self.request('http://sr//', None, 'Maria', 'slappart3').__dict__)
def test_two_different_request_from_one_partition (self):
def test_two_different_request_from_one_partition(self):
"""
Two different request from same partition
will return two different partitions
......@@ -349,7 +350,7 @@ class TestRequest (MasterMixin, unittest.TestCase):
self.request('http://sr//', None, 'Maria', 'slappart2').__dict__,
self.request('http://sr//', None, 'frontend', 'slappart2').__dict__)
def test_slave_request_no_corresponding_partition (self):
def test_slave_request_no_corresponding_partition(self):
"""
Slave instance request will fail if no corresponding are found
"""
......@@ -357,7 +358,7 @@ class TestRequest (MasterMixin, unittest.TestCase):
with self.assertRaises(WrongFormat):
self.request('http://sr//', None, 'Maria', 'slappart2', shared=True)
def test_slave_request_set_parameters (self):
def test_slave_request_set_parameters(self):
"""
Parameters sent in slave request must be put in slave master
slave instance list.
......@@ -372,14 +373,14 @@ class TestRequest (MasterMixin, unittest.TestCase):
# First request of slave instance
wanted_domain = 'fou.org'
self.request('http://sr//', None, 'Maria', 'slappart2', shared=True,
partition_parameter_kw = {'domain':wanted_domain})
partition_parameter_kw={'domain': wanted_domain})
# Get updated information for master partition
master_partition = self.getPartitionInformation(master_partition_id)
our_slave = master_partition._parameter_dict['slave_instance_list'][0]
self.assertEqual(our_slave.get('domain'),wanted_domain)
self.assertEqual(our_slave.get('domain'), wanted_domain)
def test_slave_request_set_parameters_are_updated (self):
def test_slave_request_set_parameters_are_updated(self):
"""
Parameters sent in slave request must be put in slave master
slave instance list and updated when they change.
......@@ -397,25 +398,24 @@ class TestRequest (MasterMixin, unittest.TestCase):
# First request of slave instance
wanted_domain_1 = 'crazy.org'
self.request('http://sr//', None, 'Maria', 'slappart2', shared=True,
partition_parameter_kw = {'domain':wanted_domain_1})
partition_parameter_kw={'domain': wanted_domain_1})
# Get updated information for master partition
master_partition = self.getPartitionInformation(master_partition_id)
our_slave = master_partition._parameter_dict['slave_instance_list'][0]
self.assertEqual(our_slave.get('domain'),wanted_domain_1)
self.assertEqual(our_slave.get('domain'), wanted_domain_1)
# Second request of slave instance
wanted_domain_2 = 'maluco.org'
self.request('http://sr//', None, 'Maria', 'slappart2', shared=True,
partition_parameter_kw = {'domain':wanted_domain_2})
partition_parameter_kw={'domain': wanted_domain_2})
# Get updated information for master partition
master_partition = self.getPartitionInformation(master_partition_id)
our_slave = master_partition._parameter_dict['slave_instance_list'][0]
self.assertNotEqual(our_slave.get('domain'),wanted_domain_1)
self.assertEqual(our_slave.get('domain'),wanted_domain_2)
self.assertNotEqual(our_slave.get('domain'), wanted_domain_1)
self.assertEqual(our_slave.get('domain'), wanted_domain_2)
def test_slave_request_one_corresponding_partition (self):
def test_slave_request_one_corresponding_partition(self):
"""
Successfull request slave instance follow these steps:
1. Provide one corresponding partition
......@@ -436,12 +436,12 @@ class TestRequest (MasterMixin, unittest.TestCase):
our_slave = self.request('http://sr//', None,
name, requester, shared=True)
self.assertIsInstance(our_slave, slapos.slap.ComputerPartition)
self.assertEqual(our_slave._connection_dict,{})
self.assertEqual(our_slave._connection_dict, {})
# Get updated information for master partition
master_partition = self.getPartitionInformation(master_partition_id)
slave_for_master = master_partition._parameter_dict['slave_instance_list'][0]
# Send information about slave
slave_address = {'url':'%s.master.com'}
slave_address = {'url': '%s.master.com'}
self.setConnectionDict(partition_id=master_partition._partition_id,
connection_dict=slave_address,
slave_reference=slave_for_master['slave_reference'])
......
slapos/tests/util.py
View file @ 5c613eb5
......@@ -29,6 +29,7 @@ import slapos.util
import tempfile
import unittest
class TestMkdirP(unittest.TestCase):
"""
Tests methods available in the slapos.util module.
......
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
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7