Commit 2e925da6 authored by Antoine Catton's avatar Antoine Catton

Adding tests for slapgrid promises

parent 9af4f13a
...@@ -81,7 +81,6 @@ class HTTPSConnection(HTTPConnection): ...@@ -81,7 +81,6 @@ class HTTPSConnection(HTTPConnection):
source_address=None): source_address=None):
super().__init__(self, host, port, strict, timeout, super().__init__(self, host, port, strict, timeout,
source_address) source_address)
pass
class HTTPResponse(object): class HTTPResponse(object):
......
...@@ -68,6 +68,29 @@ class MasterMixin(BasicMixin): ...@@ -68,6 +68,29 @@ class MasterMixin(BasicMixin):
setattr(httplib, name, original_value) setattr(httplib, name, original_value)
del self.saved_httplib del self.saved_httplib
def _create_instance(self, name=0):
if not os.path.isdir(self.instance_root):
os.mkdir(self.instance_root)
partition_path = os.path.join(self.instance_root, str(name))
os.mkdir(partition_path, 0750)
return partition_path
def _bootstrap(self):
os.mkdir(self.software_root)
software_hash = slapos.grid.utils.getSoftwareUrlHash('http://sr/')
srdir = os.path.join(self.software_root, software_hash)
os.mkdir(srdir)
open(os.path.join(srdir, 'template.cfg'), 'w').write(
"""[buildout]""")
srbindir = os.path.join(srdir, 'bin')
os.mkdir(srbindir)
open(os.path.join(srbindir, 'buildout'), 'w').write("""#!/bin/sh
touch worked""")
os.chmod(os.path.join(srbindir, 'buildout'), 0755)
return software_hash
def setUp(self): def setUp(self):
self._patchHttplib() self._patchHttplib()
BasicMixin.setUp(self) BasicMixin.setUp(self)
...@@ -94,7 +117,7 @@ class TestSlapgridCPWithMaster(MasterMixin, unittest.TestCase): ...@@ -94,7 +117,7 @@ class TestSlapgridCPWithMaster(MasterMixin, unittest.TestCase):
parsed_qs = urlparse.parse_qs(parsed_url.query) parsed_qs = urlparse.parse_qs(parsed_url.query)
if parsed_url.path == 'getComputerInformation' and \ if parsed_url.path == 'getComputerInformation' and \
'computer_id' in parsed_qs: 'computer_id' in parsed_qs:
slap_computer = slapos.slap.Computer(parsed_qs['computer_id']) slap_computer = slapos.slap.Computer(parsed_qs['computer_id'][0])
slap_computer._software_release_list = [] slap_computer._software_release_list = []
slap_computer._computer_partition_list = [] slap_computer._computer_partition_list = []
return (200, {}, xml_marshaller.xml_marshaller.dumps(slap_computer)) return (200, {}, xml_marshaller.xml_marshaller.dumps(slap_computer))
...@@ -111,13 +134,13 @@ class TestSlapgridCPWithMaster(MasterMixin, unittest.TestCase): ...@@ -111,13 +134,13 @@ class TestSlapgridCPWithMaster(MasterMixin, unittest.TestCase):
def test_one_partition(self): def test_one_partition(self):
def server_response(self, path, method, body, header): def server_response(self, path, method, body, header):
parsed_url = urlparse.urlparse('/' + path) parsed_url = urlparse.urlparse(path.lstrip('/'))
parsed_qs = urlparse.parse_qs(parsed_url.query) parsed_qs = urlparse.parse_qs(parsed_url.query)
if parsed_url.path == '/getComputerInformation' and \ if parsed_url.path == 'getComputerInformation' and \
'computer_id' in parsed_qs: 'computer_id' in parsed_qs:
slap_computer = slapos.slap.Computer(parsed_qs['computer_id']) slap_computer = slapos.slap.Computer(parsed_qs['computer_id'][0])
slap_computer._software_release_list = [] slap_computer._software_release_list = []
partition = slapos.slap.ComputerPartition(parsed_qs['computer_id'], partition = slapos.slap.ComputerPartition(parsed_qs['computer_id'][0],
'0') '0')
partition._need_modification = True partition._need_modification = True
sr = slapos.slap.SoftwareRelease() sr = slapos.slap.SoftwareRelease()
...@@ -130,22 +153,106 @@ class TestSlapgridCPWithMaster(MasterMixin, unittest.TestCase): ...@@ -130,22 +153,106 @@ class TestSlapgridCPWithMaster(MasterMixin, unittest.TestCase):
return (404, {}, '') return (404, {}, '')
httplib.HTTPConnection._callback = server_response httplib.HTTPConnection._callback = server_response
os.mkdir(self.software_root) instance_path = self._create_instance('0')
os.mkdir(self.instance_root) software_hash = self._bootstrap()
partition_path = os.path.join(self.instance_root, '0')
os.mkdir(partition_path, 0750)
software_hash = slapos.grid.utils.getSoftwareUrlHash('http://sr/')
srdir = os.path.join(self.software_root, software_hash)
os.mkdir(srdir)
open(os.path.join(srdir, 'template.cfg'), 'w').write(
"""[buildout]""")
srbindir = os.path.join(srdir, 'bin')
os.mkdir(srbindir)
open(os.path.join(srbindir, 'buildout'), 'w').write("""#!/bin/sh
touch worked""")
os.chmod(os.path.join(srbindir, 'buildout'), 0755)
self.assertTrue(self.grid.processComputerPartitionList()) self.assertTrue(self.grid.processComputerPartitionList())
self.assertSortedListEqual(os.listdir(self.instance_root), ['0', 'etc', self.assertSortedListEqual(os.listdir(self.instance_root), ['0', 'etc',
'var']) 'var'])
self.assertSortedListEqual(os.listdir(self.software_root), self.assertSortedListEqual(os.listdir(self.software_root),
[software_hash]) [software_hash])
def test_one_failing_promise(self):
def server_response(self_httplib, path, method, body, header):
parsed_url = urlparse.urlparse(path.lstrip('/'))
if method == 'GET':
parsed_qs = urlparse.parse_qs(parsed_url.query)
else:
parsed_qs = urlparse.parse_qs(body)
if 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 = []
partition = slapos.slap.ComputerPartition(parsed_qs['computer_id'][0],
'0')
partition._need_modification = True
sr = slapos.slap.SoftwareRelease()
sr._software_release = 'http://sr/'
partition._software_release_document = sr
partition._requested_state = 'stopped'
slap_computer._computer_partition_list = [partition]
return (200, {}, xml_marshaller.xml_marshaller.dumps(slap_computer))
if parsed_url.path == 'softwareInstanceError' and \
method == 'POST' and 'computer_partition_id' in parsed_qs:
self.error = True
self.assertEqual(parsed_qs['computer_partition_id'][0], '0')
return (200, {}, '')
else:
return (404, {}, '')
httplib.HTTPConnection._callback = server_response
self.error = False
instance_path = self._create_instance('0')
software_hash = self._bootstrap()
promise_path = os.path.join(instance_path, 'etc', 'promise')
os.makedirs(promise_path)
fail = os.path.join(promise_path, 'fail')
with open(fail, 'w') as f:
f.write("""#!/usr/bin/env sh
exit 127""")
os.chmod(fail, 0777)
self.assertTrue(self.grid.processComputerPartitionList())
self.assertTrue(self.error)
def test_one_succeeding_promise(self):
def server_response(self_httplib, path, method, body, header):
parsed_url = urlparse.urlparse(path.lstrip('/'))
if method == 'GET':
parsed_qs = urlparse.parse_qs(parsed_url.query)
else:
parsed_qs = urlparse.parse_qs(body)
if 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 = []
partition = slapos.slap.ComputerPartition(parsed_qs['computer_id'][0],
'0')
partition._need_modification = True
sr = slapos.slap.SoftwareRelease()
sr._software_release = 'http://sr/'
partition._software_release_document = sr
partition._requested_state = 'stopped'
slap_computer._computer_partition_list = [partition]
return (200, {}, xml_marshaller.xml_marshaller.dumps(slap_computer))
if parsed_url.path == 'softwareInstanceError' and \
method == 'POST' and 'computer_partition_id' in parsed_qs:
self.error = True
raise AssertionError('ComputerPartition.error was raised')
return (200, {}, '')
else:
return (404, {}, '')
httplib.HTTPConnection._callback = server_response
self.error = False
instance_path = self._create_instance('0')
software_hash = self._bootstrap()
promise_path = os.path.join(instance_path, 'etc', 'promise')
os.makedirs(promise_path)
succeed = os.path.join(promise_path, 'succeed')
with open(succeed, 'w') as f:
f.write("""#!/usr/bin/env sh
exit 0""")
os.chmod(succeed, 0777)
self.assertTrue(self.grid.processComputerPartitionList())
self.assertFalse(self.error)
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