Skip to content

S
slapos.core
Commit 11e60887 authored by Marco Mariani's avatar Marco Mariani
Browse files
Options

update tests conforming to requests API

parent 4da82657
Hide whitespace changes
Inline Side-by-side
Showing with 1430 additions and 1483 deletions
setup.py
View file @ 11e60887
......@@ -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
......
slapos/proxy/views.py
View file @ 11e60887
......@@ -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
......
slapos/slap/slap.py
View file @ 11e60887
......@@ -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
slapos/tests/client.py
View file @ 11e60887
......@@ -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
......
slapos/tests/slap.py
View file @ 11e60887
......@@ -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())
with httmock.HTTMock(handler):
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())
def _test_new_computer_partition_state(self, state):
"""
......@@ -676,23 +700,26 @@ class TestComputerPartition(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
parsed_qs['computer_reference'][0] == computer_guid and
parsed_qs['computer_partition_reference'][0] == partition_id):
def handler(url, req):
qs = urlparse.parse_qs(url.query)
if (url.path == '/registerComputerPartition' and
qs['computer_reference'][0] == computer_guid and
qs['computer_partition_reference'][0] == partition_id):
partition = slapos.slap.ComputerPartition(
computer_guid, partition_id)
return (200, {}, xml_marshaller.xml_marshaller.dumps(partition))
return {
'status_code': 200,
'content': xml_marshaller.xml_marshaller.dumps(partition)
}
else:
return (404, {}, '')
httplib.HTTPConnection._callback = server_response
return {'status_code': 404}
computer_partition = self.slap.registerComputerPartition(
computer_guid, partition_id)
self.assertRaises(slapos.slap.NotFoundError,
getattr(computer_partition, state))
with httmock.HTTMock(handler):
computer_partition = self.slap.registerComputerPartition(
computer_guid, partition_id)
self.assertRaises(slapos.slap.NotFoundError,
getattr(computer_partition, state))
def test_available_new_ComputerPartition_raises(self):
"""
......@@ -731,32 +758,35 @@ class TestComputerPartition(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
parsed_qs['computer_reference'][0] == computer_guid and
parsed_qs['computer_partition_reference'][0] == partition_id):
def handler(url, req):
qs = urlparse.parse_qs(url.query)
if (url.path == '/registerComputerPartition' and
qs['computer_reference'][0] == computer_guid and
qs['computer_partition_reference'][0] == partition_id):
partition = slapos.slap.ComputerPartition(
computer_guid, partition_id)
return (200, {}, xml_marshaller.xml_marshaller.dumps(partition))
elif parsed_url.path == 'softwareInstanceError':
parsed_qs_body = urlparse.parse_qs(body)
return {
'statu_code': 200,
'content': xml_marshaller.xml_marshaller.dumps(partition)
}
elif url.path == '/softwareInstanceError':
parsed_qs_body = urlparse.parse_qs(req.body)
# 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'):
return (200, {}, '')
return {'status_code': 200}
return {'status_code': 404}
return (404, {}, '')
httplib.HTTPConnection._callback = server_response
computer_partition = slap.registerComputerPartition(
computer_guid, partition_id)
# XXX: Interface does not define return value
computer_partition.error('some error')
with httmock.HTTMock(handler):
computer_partition = slap.registerComputerPartition(
computer_guid, partition_id)
# XXX: Interface does not define return value
computer_partition.error('some error')
class TestSoftwareRelease(SlapMixin):
......@@ -800,21 +830,22 @@ class TestSoftwareRelease(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(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'):
return (200, {}, '')
return (404, {}, '')
def handler(url, req):
qs = urlparse.parse_qs(req.body)
if (url.path == '/softwareReleaseError' and
qs['computer_id'][0] == computer_guid and
qs['url'][0] == software_release_uri and
qs['error_log'][0] == 'some error'):
return {
'status_code': 200
}
return {'status_code': 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')
with httmock.HTTMock(handler):
software_release = self.slap.registerSoftwareRelease(software_release_uri)
software_release._computer_guid = computer_guid
software_release.error('some error')
class TestOpenOrder(SlapMixin):
......@@ -825,24 +856,31 @@ class TestOpenOrder(SlapMixin):
# 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':
def handler(url, req):
if url.path == '/requestComputerPartition':
raise RequestWasCalled
httplib.HTTPConnection._callback = server_response
self.assertRaises(RequestWasCalled,
open_order.request,
software_release_uri, 'myrefe')
with httmock.HTTMock(handler):
self.assertRaises(RequestWasCalled,
open_order.request,
software_release_uri, 'myrefe')
@unittest.skip('unclear what should be returned')
def test_request_not_raises(self):
software_release_uri = 'http://server/new/' + self._getTestComputerId()
self.slap = slapos.slap.slap()
self.slap.initializeConnection(self.server_url)
# XXX: Interface lack registerOpenOrder method declaration
open_order = self.slap.registerOpenOrder()
computer_partition = open_order.request(software_release_uri, 'myrefe')
self.assertIsInstance(computer_partition, slapos.slap.ComputerPartition)
def handler(url, req):
if url.path == '/requestComputerPartition':
pass
# XXX what to do here?
with httmock.HTTMock(handler):
open_order = self.slap.registerOpenOrder()
computer_partition = open_order.request(software_release_uri, 'myrefe')
self.assertIsInstance(computer_partition, slapos.slap.ComputerPartition)
def test_request_raises_later(self):
software_release_uri = 'http://server/new/' + self._getTestComputerId()
......@@ -851,15 +889,15 @@ class TestOpenOrder(SlapMixin):
# XXX: Interface lack registerOpenOrder method declaration
open_order = self.slap.registerOpenOrder()
def server_response(self, path, method, body, header):
return (408, {}, '')
def handler(url, req):
return {'status_code': 408}
httplib.HTTPConnection._callback = server_response
computer_partition = open_order.request(software_release_uri, 'myrefe')
self.assertIsInstance(computer_partition, slapos.slap.ComputerPartition)
with httmock.HTTMock(handler):
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()
......@@ -870,18 +908,20 @@ class TestOpenOrder(SlapMixin):
computer_guid = self._getTestComputerId()
requested_partition_id = 'PARTITION_01'
def server_response(self, path, method, body, header):
def handler(url, req):
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)
}
httplib.HTTPConnection._callback = server_response
computer_partition = open_order.request(software_release_uri, 'myrefe')
self.assertIsInstance(computer_partition, slapos.slap.ComputerPartition)
self.assertEqual(requested_partition_id, computer_partition.getId())
with httmock.HTTMock(handler):
computer_partition = open_order.request(software_release_uri, 'myrefe')
self.assertIsInstance(computer_partition, slapos.slap.ComputerPartition)
self.assertEqual(requested_partition_id, computer_partition.getId())
def test_request_getConnectionParameter(self):
......@@ -894,21 +934,24 @@ class TestOpenOrder(SlapMixin):
computer_guid = self._getTestComputerId()
requested_partition_id = 'PARTITION_01'
def server_response(self, path, method, body, header):
def handler(url, req):
from slapos.slap.slap import SoftwareInstance
slap_partition = SoftwareInstance(
_connection_dict = {"url": 'URL_CONNECTION_PARAMETER'},
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)
}
httplib.HTTPConnection._callback = server_response
computer_partition = open_order.request(software_release_uri, 'myrefe')
self.assertIsInstance(computer_partition, slapos.slap.ComputerPartition)
self.assertEqual(requested_partition_id, computer_partition.getId())
self.assertEqual("URL_CONNECTION_PARAMETER",
computer_partition.getConnectionParameter('url'))
with httmock.HTTMock(handler):
computer_partition = open_order.request(software_release_uri, 'myrefe')
self.assertIsInstance(computer_partition, slapos.slap.ComputerPartition)
self.assertEqual(requested_partition_id, computer_partition.getId())
self.assertEqual("URL_CONNECTION_PARAMETER",
computer_partition.getConnectionParameter('url'))
def test_request_connection_dict_backward_compatibility(self):
......@@ -921,7 +964,7 @@ class TestOpenOrder(SlapMixin):
computer_guid = self._getTestComputerId()
requested_partition_id = 'PARTITION_01'
def server_response(self, path, method, body, header):
def handler(url, req):
from slapos.slap.slap import SoftwareInstance
slap_partition = SoftwareInstance(
connection_xml="""<?xml version='1.0' encoding='utf-8'?>
......@@ -930,15 +973,17 @@ class TestOpenOrder(SlapMixin):
</instance>""",
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')
self.assertIsInstance(computer_partition, slapos.slap.ComputerPartition)
self.assertEqual(requested_partition_id, computer_partition.getId())
self.assertEqual("URL_CONNECTION_PARAMETER",
computer_partition.getConnectionParameter('url'))
return {
'status_code': 200,
'content': xml_marshaller.xml_marshaller.dumps(slap_partition)
}
with httmock.HTTMock(handler):
computer_partition = open_order.request(software_release_uri, 'myrefe')
self.assertIsInstance(computer_partition, slapos.slap.ComputerPartition)
self.assertEqual(requested_partition_id, computer_partition.getId())
self.assertEqual("URL_CONNECTION_PARAMETER",
computer_partition.getConnectionParameter('url'))
class TestSoftwareProductCollection(SlapMixin):
......
slapos/tests/slapgrid.py
View file @ 11e60887
This source diff could not be displayed because it is too large. You can view the blob instead.
slapos/tests/slapmock/httplib.py deleted 100644 → 0
View file @ 4da82657
#!/usr/bin/env python2
# -*- coding: utf-8 -*-
"""Mocked httplib
"""
__all__ = []
def log(message):
"""Need to be overridden to get a proper logger
"""
pass
class HTTPConnection(object):
scheme = 'http'
def _callback(self, path, method, body, headers):
"""To get it works properly, you need to override
HTTPConnection._callback. This method received the instance, the path,
method and request body as parameter, and it has to return a tuple with
headers dictionary and body response string.
@param self object instance reference
@param URL the parsed URL
@param method the http method
@param body the request body
@param headers the request headers
@return tuple containing status integer, headers dictionary and body
response"""
return (0, {}, '', )
def __init__(self, host, port=None, strict=None,
timeout=None, source_address=None):
self.host = host
self.port = port
self.strict = strict
self.timeout = timeout
self.source_address = source_address
self.__response = None
def request(self, method, url, body=None, headers=None):
status, headers, body = self._callback(url, method, body, headers)
self.__response = HTTPResponse('HTTP/1.1', status, 'OK', body, headers)
def getresponse(self):
response = self.__response
self.__response = None
return response
def set_debuglevel(self, level):
pass
def set_tunnel(self, host, port=None, headers=None):
pass
def connect(self):
pass
def close(self):
pass
def putrequest(self, request, selector, skip_host=None,
skip_accept_encoding=None):
pass
def putheader(self, *args):
pass
def endheaders(self):
pass
def send(self, data):
pass
class HTTPSConnection(HTTPConnection):
def __init__(self, host, port=None, key_file=None,
cert_file=None, strict=None, timeout=None,
source_address=None):
super(HTTPSConnection, self).__init__(host, port, strict, timeout,
source_address)
self.certificate = open(cert_file, 'r').read()
self.key = open(key_file, 'r').read()
def request(self, method, url, body=None, headers=None):
headers['certificate'] = self.certificate
headers['key'] = self.key
status, headers, body = self._callback(url, method, body, headers)
self.__response = HTTPResponse('HTTP/1.1', status, 'OK', body, headers)
def getresponse(self):
response = self.__response
self.__response = None
return response
class HTTPResponse(object):
def __init__(self, version, status, reason, content, headers=()):
self.version = version
self.status = status
self.reason = reason
self.__headers = headers
self.__content = content
def read(self, amt=None):
result = None
if amt is None:
result = self.__content
self.__content = ''
else:
end = max(amt, len(self.__content))
result = self.__content[:end]
del self.__content[:end]
return result
def getheader(self, name, default=None):
pass
def getheaders(self):
pass
slapos/tests/slapmock/requests.py 0 → 100644
View file @ 11e60887
# -*- coding: utf-8 -*-
def response_ok(url, request):
return {
'status_code': 200,
'content': ''
}
slapos/tests/slapproxy/__init__.py
View file @ 11e60887
......@@ -968,7 +968,7 @@ database_uri = %(tempdir)s/lib/external_proxy.db
'xml': xml_marshaller.xml_marshaller.dumps(computer_dict),
}
self.external_proxy_slap._connection_helper.POST('/loadComputerConfigurationFromXML',
parameter_dict=request_dict)
data=request_dict)
def _checkInstanceIsFowarded(self, name, partition_parameter_kw, software_release):
"""
......
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