Commit dc996cc6 by Marco Mariani Committed by Rafael Monnerat

update tests conforming to requests API

1 parent f86f0d39
......@@ -66,7 +66,8 @@ setup(name=name,
'bpython_console': ('bpython',)},
tests_require=[
'pyflakes',
'mock'
'mock',
'httmock',
],
zip_safe=False, # proxy depends on Flask, which has issues with
# accessing templates
......
......@@ -495,7 +495,7 @@ def forwardRequestToExternalMaster(master_url, request_form):
filter_kw['source_instance_id'] = partition_reference
new_request_form['filter_xml'] = dumps(filter_kw)
partition = loads(slap._connection_helper.POST('/requestComputerPartition', new_request_form))
partition = loads(slap._connection_helper.POST('/requestComputerPartition', data=new_request_form))
# XXX move to other end
partition._master_url = master_url
......
......@@ -338,7 +338,7 @@ class Computer(SlapDocument):
'message': message})
def getStatus(self):
xml = self._connection_helper.GET('getComputerStatus', {'computer_id': self._computer_id})
xml = self._connection_helper.GET('getComputerStatus', params={'computer_id': self._computer_id})
return xml_marshaller.loads(xml)
def revokeCertificate(self):
......@@ -548,7 +548,7 @@ class ComputerPartition(SlapRequester):
def getCertificate(self):
xml = self._connection_helper.GET('getComputerPartitionCertificate',
{
params={
'computer_id': self._computer_id,
'computer_partition_id': self._partition_id,
}
......@@ -557,7 +557,7 @@ class ComputerPartition(SlapRequester):
def getStatus(self):
xml = self._connection_helper.GET('getComputerPartitionStatus',
{
params={
'computer_id': self._computer_id,
'computer_partition_id': self._partition_id,
}
......@@ -579,7 +579,7 @@ class ConnectionHelper:
self.timeout = timeout
def getComputerInformation(self, computer_id):
xml = self.GET('getComputerInformation', {'computer_id': computer_id})
xml = self.GET('getComputerInformation', params={'computer_id': computer_id})
return xml_marshaller.loads(xml)
def getFullComputerInformation(self, computer_id):
......@@ -591,20 +591,21 @@ class ConnectionHelper:
params = {'computer_id': computer_id}
if not computer_id:
# XXX-Cedric: should raise something smarter than "NotFound".
raise NotFoundError('%r %r' (path, params))
raise NotFoundError('%r %r' % (path, params))
try:
xml = self.GET(path, params)
xml = self.GET(path, params=params)
except NotFoundError:
# XXX: This is a ugly way to keep backward compatibility,
# We should stablise slap library soon.
xml = self.GET('getComputerInformation', {'computer_id': computer_id})
xml = self.GET('getComputerInformation', params=params)
return xml_marshaller.loads(xml)
def do_request(self, method, path, params=None, data=None, headers=None):
url = urlparse.urljoin(self.slapgrid_uri, path)
if path.startswith('/'):
raise ValueError('method path should be relative: %s' % path)
path = path[1:]
# raise ValueError('method path should be relative: %s' % path)
try:
if url.startswith('https'):
......@@ -707,7 +708,7 @@ class slap:
raise NotFoundError
xml = self._connection_helper.GET('registerComputerPartition',
{
params = {
'computer_reference': computer_guid,
'computer_partition_reference': partition_id,
}
......@@ -726,18 +727,19 @@ class slap:
def getSoftwareReleaseListFromSoftwareProduct(self,
software_product_reference=None, software_release_url=None):
url = '/getSoftwareReleaseListFromSoftwareProduct?'
url = 'getSoftwareReleaseListFromSoftwareProduct'
params = {}
if software_product_reference:
if software_release_url is not None:
raise AttributeError('Both software_product_reference and '
'software_release_url parameters are specified.')
url += 'software_product_reference=%s' % software_product_reference
params['software_product_reference'] = software_product_reference
else:
if software_release_url is None:
raise AttributeError('None of software_product_reference and '
'software_release_url parameters are specified.')
url += 'software_release_url=%s' % software_release_url
params['software_release_url'] = software_release_url
result = xml_marshaller.loads(self._connection_helper.GET(url))
result = xml_marshaller.loads(self._connection_helper.GET(url, params=params))
assert(type(result) == list)
return result
......@@ -27,8 +27,6 @@
import logging
import unittest
# XXX: BasicMixin should be in a separated module, not in slapgrid test module.
from slapos.tests.slapgrid import BasicMixin
import slapos.slap
import slapos.client
......
......@@ -25,18 +25,16 @@
#
##############################################################################
import httplib
import logging
import os
import unittest
import urlparse
import httmock
import slapos.slap
import xml_marshaller
ORIGINAL_HTTPLIB_HTTPCONNECTION = httplib.HTTPConnection
ORIGINAL_HTTPLIB_HTTPSCONNECTION = httplib.HTTPSConnection
ORIGINAL_HTTPLIB_HTTPRESPONSE = httplib.HTTPResponse
class UndefinedYetException(Exception):
"""To catch exceptions which are not yet defined"""
......@@ -49,7 +47,6 @@ class SlapMixin(unittest.TestCase):
def setUp(self):
self._server_url = os.environ.get('TEST_SLAP_SERVER_URL', None)
if self._server_url is None:
self._patchHttplib()
self.server_url = 'http://localhost/'
else:
self.server_url = self._server_url
......@@ -58,31 +55,7 @@ class SlapMixin(unittest.TestCase):
self.partition_id = 'PARTITION_01'
def tearDown(self):
self._unpatchHttplib()
def _patchHttplib(self):
"""Overrides httplib"""
import slapmock.httplib
self.saved_httplib = {}
for fake in vars(slapmock.httplib):
self.saved_httplib[fake] = getattr(httplib, fake, None)
setattr(httplib, fake, getattr(slapmock.httplib, fake))
def _unpatchHttplib(self):
"""Restores httplib overriding"""
import httplib
# XXX not reliable
for name, original_value in self.saved_httplib.items():
setattr(httplib, name, original_value)
del self.saved_httplib
# XXX this fixes upper code, to be sure it is reliable
httplib.HTTPConnection = ORIGINAL_HTTPLIB_HTTPCONNECTION
httplib.HTTPSConnection = ORIGINAL_HTTPLIB_HTTPSCONNECTION
httplib.HTTPResponse = ORIGINAL_HTTPLIB_HTTPRESPONSE
pass
def _getTestComputerId(self):
"""
......@@ -103,19 +76,7 @@ class TestSlap(SlapMixin):
"""
slap_instance = slapos.slap.slap()
slap_instance.initializeConnection(self.server_url)
self.assertIn(slap_instance._connection_helper.host, self.server_url)
self.assertIn(slap_instance._connection_helper.path, self.server_url)
def test_slap_initialisation_wrong_url(self):
"""
Asserts that slap initialisation raises exception when passed url
is not correct
"""
server_url = 'https://user:pass@server/path/path?parameter=notAcceptable'
slap_instance = slapos.slap.slap()
self.assertRaises(AttributeError,
slap_instance.initializeConnection,
server_url)
self.assertEquals(slap_instance._connection_helper.slapgrid_uri, self.server_url)
def test_registerComputer_with_new_guid(self):
"""
......@@ -179,21 +140,26 @@ class TestSlap(SlapMixin):
self.slap.initializeConnection(self.server_url)
self.slap.registerComputer(computer_guid)
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):
partition = slapos.slap.ComputerPartition(
computer_guid, partition_id)
return (200, {}, xml_marshaller.xml_marshaller.dumps(partition))
def handler(url, req):
qs = urlparse.parse_qs(url.query)
if (url.path == '/registerComputerPartition'
and qs == {
'computer_reference': [computer_guid],
'computer_partition_reference': [partition_id]
}):
partition = slapos.slap.ComputerPartition(computer_guid, partition_id)
return {
'status_code': 200,
'content': xml_marshaller.xml_marshaller.dumps(partition)
}
else:
return (404, {}, '')
httplib.HTTPConnection._callback = server_response
return {'status_code': 400}
partition = self.slap.registerComputerPartition(computer_guid, partition_id)
self.assertIsInstance(partition, slapos.slap.ComputerPartition)
self._handler = handler
with httmock.HTTMock(handler):
partition = self.slap.registerComputerPartition(computer_guid, partition_id)
self.assertIsInstance(partition, slapos.slap.ComputerPartition)
def test_registerComputerPartition_existing_partition_id_known_computer_guid(self):
"""
......@@ -201,9 +167,10 @@ class TestSlap(SlapMixin):
returns ComputerPartition object
"""
self.test_registerComputerPartition_new_partition_id_known_computer_guid()
partition = self.slap.registerComputerPartition(self._getTestComputerId(),
self.partition_id)
self.assertIsInstance(partition, slapos.slap.ComputerPartition)
with httmock.HTTMock(self._handler):
partition = self.slap.registerComputerPartition(self._getTestComputerId(),
self.partition_id)
self.assertIsInstance(partition, slapos.slap.ComputerPartition)
def test_registerComputerPartition_unknown_computer_guid(self):
"""
......@@ -214,21 +181,22 @@ class TestSlap(SlapMixin):
self.slap.initializeConnection(self.server_url)
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 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, {}, '')
def handler(url, req):
qs = urlparse.parse_qs(url.query)
if (url.path == '/registerComputerPartition'
and qs == {
'computer_reference': [computer_guid],
'computer_partition_reference': [partition_id]
}):
return {'status_code': 404}
else:
return (0, {}, '')
httplib.HTTPConnection._callback = server_response
return {'status_code': 0}
with httmock.HTTMock(handler):
self.assertRaises(slapos.slap.NotFoundError,
self.slap.registerComputerPartition,
computer_guid, partition_id)
self.assertRaises(slapos.slap.NotFoundError,
self.slap.registerComputerPartition,
computer_guid, partition_id)
def test_getFullComputerInformation_empty_computer_guid(self):
"""
......@@ -237,15 +205,14 @@ class TestSlap(SlapMixin):
"""
self.slap.initializeConnection(self.server_url)
def server_response(self_httpconnection, path, method, body, header):
def handler(url, req):
# Shouldn't even be called
self.assertFalse(True)
httplib.HTTPConnection._callback = server_response
self.assertRaises(slapos.slap.NotFoundError,
self.slap._connection_helper.getFullComputerInformation,
None)
with httmock.HTTMock(handler):
self.assertRaises(slapos.slap.NotFoundError,
self.slap._connection_helper.getFullComputerInformation,
None)
def test_registerComputerPartition_empty_computer_guid(self):
"""
......@@ -254,15 +221,14 @@ class TestSlap(SlapMixin):
"""
self.slap.initializeConnection(self.server_url)
def server_response(self_httpconnection, path, method, body, header):
def handler(url, req):
# Shouldn't even be called
self.assertFalse(True)
httplib.HTTPConnection._callback = server_response
self.assertRaises(slapos.slap.NotFoundError,
self.slap.registerComputerPartition,
None, 'PARTITION_01')
with httmock.HTTMock(handler):
self.assertRaises(slapos.slap.NotFoundError,
self.slap.registerComputerPartition,
None, 'PARTITION_01')
def test_registerComputerPartition_empty_computer_partition_id(self):
"""
......@@ -271,15 +237,14 @@ class TestSlap(SlapMixin):
"""
self.slap.initializeConnection(self.server_url)
def server_response(self_httpconnection, path, method, body, header):
def handler(url, req):
# Shouldn't even be called
self.assertFalse(True)
httplib.HTTPConnection._callback = server_response
self.assertRaises(slapos.slap.NotFoundError,
self.slap.registerComputerPartition,
self._getTestComputerId(), None)
with httmock.HTTMock(handler):
self.assertRaises(slapos.slap.NotFoundError,
self.slap.registerComputerPartition,
self._getTestComputerId(), None)
def test_registerComputerPartition_empty_computer_guid_empty_computer_partition_id(self):
"""
......@@ -288,15 +253,14 @@ class TestSlap(SlapMixin):
"""
self.slap.initializeConnection(self.server_url)
def server_response(self_httpconnection, path, method, body, header):
def handler(url, req):
# Shouldn't even be called
self.assertFalse(True)
httplib.HTTPConnection._callback = server_response
self.assertRaises(slapos.slap.NotFoundError,
self.slap.registerComputerPartition,
None, None)
with httmock.HTTMock(handler):
self.assertRaises(slapos.slap.NotFoundError,
self.slap.registerComputerPartition,
None, None)
def test_getSoftwareReleaseListFromSoftwareProduct_software_product_reference(self):
......@@ -309,19 +273,21 @@ class TestSlap(SlapMixin):
software_product_reference = 'random_reference'
software_release_url_list = ['1', '2']
def server_response(self_httpconnection, path, method, body, header):
parsed_url = urlparse.urlparse(path.lstrip('/'))
parsed_qs = urlparse.parse_qs(parsed_url.query)
if parsed_url.path == 'getSoftwareReleaseListFromSoftwareProduct' \
and parsed_qs == {'software_product_reference': [software_product_reference]}:
return (200, {}, xml_marshaller.xml_marshaller.dumps(software_release_url_list))
httplib.HTTPConnection._callback = server_response
self.assertEqual(
self.slap.getSoftwareReleaseListFromSoftwareProduct(
def handler(url, req):
qs = urlparse.parse_qs(url.query)
if (url.path == '/getSoftwareReleaseListFromSoftwareProduct'
and qs == {'software_product_reference': [software_product_reference]}):
return {
'status_code': 200,
'content': xml_marshaller.xml_marshaller.dumps(software_release_url_list)
}
with httmock.HTTMock(handler):
self.assertEqual(
self.slap.getSoftwareReleaseListFromSoftwareProduct(
software_product_reference=software_product_reference),
software_release_url_list
)
software_release_url_list
)
def test_getSoftwareReleaseListFromSoftwareProduct_software_release_url(self):
"""
......@@ -333,19 +299,21 @@ class TestSlap(SlapMixin):
software_release_url = 'random_url'
software_release_url_list = ['1', '2']
def server_response(self_httpconnection, path, method, body, header):
parsed_url = urlparse.urlparse(path.lstrip('/'))
parsed_qs = urlparse.parse_qs(parsed_url.query)
if parsed_url.path == 'getSoftwareReleaseListFromSoftwareProduct' \
and parsed_qs == {'software_release_url': [software_release_url]}:
return (200, {}, xml_marshaller.xml_marshaller.dumps(software_release_url_list))
httplib.HTTPConnection._callback = server_response
self.assertEqual(
self.slap.getSoftwareReleaseListFromSoftwareProduct(
software_release_url=software_release_url),
software_release_url_list
)
def handler(url, req):
qs = urlparse.parse_qs(url.query)
if (url.path == '/getSoftwareReleaseListFromSoftwareProduct'
and qs == {'software_release_url': [software_release_url]}):
return {
'status_code': 200,
'content': xml_marshaller.xml_marshaller.dumps(software_release_url_list)
}
with httmock.HTTMock(handler):
self.assertEqual(
self.slap.getSoftwareReleaseListFromSoftwareProduct(
software_release_url=software_release_url),
software_release_url_list
)
def test_getSoftwareReleaseListFromSoftwareProduct_too_many_parameters(self):
"""
......@@ -381,30 +349,35 @@ class TestComputer(SlapMixin):
slap = self.slap
slap.initializeConnection(self.server_url)
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):
def handler(url, req):
qs = urlparse.parse_qs(url.query)
if (url.path == '/registerComputerPartition'
and 'computer_reference' in qs
and 'computer_partition_reference' in 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):
slap_computer = slapos.slap.Computer(parsed_qs['computer_id'][0])
qs['computer_reference'][0],
qs['computer_partition_reference'][0])
return {
'status_code': 200,
'content': xml_marshaller.xml_marshaller.dumps(slap_partition)
}
elif (url.path == '/getFullComputerInformation'
and 'computer_id' in qs):
slap_computer = slapos.slap.Computer(qs['computer_id'][0])
slap_computer._software_release_list = []
slap_computer._computer_partition_list = []
return (200, {}, xml_marshaller.xml_marshaller.dumps(slap_computer))
elif parsed_url.path == 'requestComputerPartition':
return (408, {}, '')
return {
'status_code': 200,
'content': xml_marshaller.xml_marshaller.dumps(slap_computer)
}
elif url.path == '/requestComputerPartition':
return {'status_code': 408}
else:
return (404, {}, '')
httplib.HTTPConnection._callback = server_response
return {'status_code': 404}
computer = self.slap.registerComputer(computer_guid)
self.assertEqual(computer.getComputerPartitionList(), [])
with httmock.HTTMock(handler):
computer = self.slap.registerComputer(computer_guid)
self.assertEqual(computer.getComputerPartitionList(), [])
def _test_computer_empty_computer_guid(self, computer_method):
"""
......@@ -413,15 +386,14 @@ class TestComputer(SlapMixin):
"""
self.slap.initializeConnection(self.server_url)
def server_response(self_httpconnection, path, method, body, header):
def handler(url, req):
# Shouldn't even be called
self.assertFalse(True)
httplib.HTTPConnection._callback = server_response
computer = self.slap.registerComputer(None)
self.assertRaises(slapos.slap.NotFoundError,
getattr(computer, computer_method))
with httmock.HTTMock(handler):
computer = self.slap.registerComputer(None)
self.assertRaises(slapos.slap.NotFoundError,
getattr(computer, computer_method))
def test_computer_getComputerPartitionList_empty_computer_guid(self):
"""
......@@ -446,10 +418,35 @@ class TestComputer(SlapMixin):
partition_id = 'PARTITION_01'
self.slap = slapos.slap.slap()
self.slap.initializeConnection(self.server_url)
self.computer = self.slap.registerComputer(self.computer_guid)
self.partition = self.slap.registerComputerPartition(self.computer_guid,
partition_id)
self.assertEqual(self.computer.getComputerPartitionList(), [])
def handler(url, req):
qs = urlparse.parse_qs(url.query)
if (url.path == '/registerComputerPartition'
and qs == {
'computer_reference': [self.computer_guid],
'computer_partition_reference': [partition_id]
}):
partition = slapos.slap.ComputerPartition(self.computer_guid, partition_id)
return {
'status_code': 200,
'content': xml_marshaller.xml_marshaller.dumps(partition)
}
elif (url.path == '/getFullComputerInformation'
and 'computer_id' in qs):
slap_computer = slapos.slap.Computer(qs['computer_id'][0])
slap_computer._computer_partition_list = []
return {
'status_code': 200,
'content': xml_marshaller.xml_marshaller.dumps(slap_computer)
}
else:
return {'status_code': 400}
with httmock.HTTMock(handler):
self.computer = self.slap.registerComputer(self.computer_guid)
self.partition = self.slap.registerComputerPartition(self.computer_guid,
partition_id)
self.assertEqual(self.computer.getComputerPartitionList(), [])
@unittest.skip("Not implemented")
def test_computer_reportUsage_non_valid_xml_raises(self):
......@@ -504,167 +501,194 @@ 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):
def handler(url, req):
qs = urlparse.parse_qs(url.query)
if (url.path == '/registerComputerPartition'
and 'computer_reference' in qs
and 'computer_partition_reference' in 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):
slap_computer = slapos.slap.Computer(parsed_qs['computer_id'][0])
qs['computer_reference'][0],
qs['computer_partition_reference'][0])
return {
'status_code': 200,
'content': xml_marshaller.xml_marshaller.dumps(slap_partition)
}
elif (url.path == '/getComputerInformation'
and 'computer_id' in qs):
slap_computer = slapos.slap.Computer(qs['computer_id'][0])
slap_computer._software_release_list = []
slap_partition = slapos.slap.ComputerPartition(
parsed_qs['computer_id'][0],
qs['computer_id'][0],
partition_id)
slap_computer._computer_partition_list = [slap_partition]
return (200, {}, xml_marshaller.xml_marshaller.dumps(slap_computer))
elif parsed_url.path == 'requestComputerPartition':
return {
'status_code': 200,
'content': xml_marshaller.xml_marshaller.dumps(slap_computer)
}
elif url.path == '/requestComputerPartition':
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,
'http://server/new/' + self._getTestComputerId(),
'software_type', 'myref')
return {
'status_code': 404
}
with httmock.HTTMock(handler):
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,
'http://server/new/' + self._getTestComputerId(),
'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):
def handler(url, req):
qs = urlparse.parse_qs(url.query)
if (url.path == '/registerComputerPartition'
and 'computer_reference' in qs
and 'computer_partition_reference' in 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):
slap_computer = slapos.slap.Computer(parsed_qs['computer_id'][0])
qs['computer_reference'][0],
qs['computer_partition_reference'][0])
return {
'status_code': 200,
'content': xml_marshaller.xml_marshaller.dumps(slap_partition)
}
elif (url.path == '/getComputerInformation'
and 'computer_id' in qs):
slap_computer = slapos.slap.Computer(qs['computer_id'][0])
slap_computer._software_release_list = []
slap_partition = slapos.slap.ComputerPartition(
parsed_qs['computer_id'][0],
qs['computer_id'][0],
partition_id)
slap_computer._computer_partition_list = [slap_partition]
return (200, {}, xml_marshaller.xml_marshaller.dumps(slap_computer))
elif parsed_url.path == 'requestComputerPartition':
return (408, {}, '')
return {
'status_code': 200,
'content': xml_marshaller.xml_marshaller.dumps(slap_computer)
}
elif url.path == '/requestComputerPartition':
return {'status_code': 408}
else:
return (404, {}, '')
return {'status_code': 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)
requested_partition = computer_partition.request(
'http://server/new/' + self._getTestComputerId(),
'software_type',
'myref')
self.assertIsInstance(requested_partition, slapos.slap.ComputerPartition)
with httmock.HTTMock(handler):
computer_partition = self.slap.registerComputerPartition(
self.computer_guid, partition_id)
requested_partition = computer_partition.request(
'http://server/new/' + self._getTestComputerId(),
'software_type',
'myref')
self.assertIsInstance(requested_partition, slapos.slap.ComputerPartition)
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):
def handler(url, req):
qs = urlparse.parse_qs(url.query)
if (url.path == '/registerComputerPartition' and
'computer_reference' in qs and
'computer_partition_reference' in 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):
slap_computer = slapos.slap.Computer(parsed_qs['computer_id'][0])
qs['computer_reference'][0],
qs['computer_partition_reference'][0])
return {
'status_code': 200,
'content': xml_marshaller.xml_marshaller.dumps(slap_partition)
}
elif (url.path == '/getComputerInformation'
and 'computer_id' in qs):
slap_computer = slapos.slap.Computer(qs['computer_id'][0])
slap_computer._software_release_list = []
slap_partition = slapos.slap.ComputerPartition(
parsed_qs['computer_id'][0],
qs['computer_id'][0],
partition_id)
slap_computer._computer_partition_list = [slap_partition]
return (200, {}, xml_marshaller.xml_marshaller.dumps(slap_computer))
elif parsed_url.path == 'requestComputerPartition':
return (408, {}, '')
return {
'status_code': 200,
'content': xml_marshaller.xml_marshaller.dumps(slap_computer)
}
elif url.path == '/requestComputerPartition':
return {'status_code': 408}
else:
return (404, {}, '')
return {'status_code': 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)
requested_partition = computer_partition.request(
'http://server/new/' + self._getTestComputerId(),
'software_type',
'myref')
self.assertIsInstance(requested_partition, slapos.slap.ComputerPartition)
# as request method does not raise, accessing data raises
self.assertRaises(slapos.slap.ResourceNotReady,
requested_partition.getId)
with httmock.HTTMock(handler):
computer_partition = self.slap.registerComputerPartition(
self.computer_guid, partition_id)
requested_partition = computer_partition.request(
'http://server/new/' + self._getTestComputerId(),
'software_type',
'myref')
self.assertIsInstance(requested_partition, slapos.slap.ComputerPartition)
# as request method does not raise, accessing data raises
self.assertRaises(slapos.slap.ResourceNotReady,
requested_partition.getId)
def test_request_fullfilled_work(self):
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):
def handler(url, req):
qs = urlparse.parse_qs(url.query)
if (url.path == '/registerComputerPartition' and
'computer_reference' in qs and
'computer_partition_reference' in 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):
slap_computer = slapos.slap.Computer(parsed_qs['computer_id'][0])
qs['computer_reference'][0],
qs['computer_partition_reference'][0])
return {
'status_code': 200,
'content': xml_marshaller.xml_marshaller.dumps(slap_partition)
}
elif (url.path == '/getComputerInformation' and 'computer_id' in qs):
slap_computer = slapos.slap.Computer(qs['computer_id'][0])
slap_computer._software_release_list = []
slap_partition = slapos.slap.ComputerPartition(
parsed_qs['computer_id'][0],
qs['computer_id'][0],
partition_id)
slap_computer._computer_partition_list = [slap_partition]
return (200, {}, xml_marshaller.xml_marshaller.dumps(slap_computer))
elif parsed_url.path == 'requestComputerPartition':
return {
'status_code': 200,
'content': xml_marshaller.xml_marshaller.dumps(slap_computer)
}
elif url.path == '/requestComputerPartition':
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))
return {
'status_code': 200,
'content': xml_marshaller.xml_marshaller.dumps(slap_partition)
}
else:
return (404, {}, '')
return {'status_code': 404}
httplib.HTTPConnection._callback = server_response
self.slap = slapos.slap.slap()
self.slap.initializeConnection(self.server_url)
computer_partition = self.slap.registerComputerPartition(
computer_guid, partition_id)
requested_partition = computer_partition.request(
'http://server/new/' + self._getTestComputerId(),
'software_type',
'myref')
self.assertIsInstance(requested_partition, slapos.slap.ComputerPartition)
# as request method does not raise, accessing data in case when
# request was done works correctly
self.assertEqual(requested_partition_id, requested_partition.getId())