Skip to content

S
slapos.core
Commit e44bac78 authored by Cédric Le Ninivin's avatar Cédric Le Ninivin Committed by Cédric Le Ninivin
Browse files
Options

slapos_jio_api_style: Finish porting test from SlapTool

parent 8c812744
Show whitespace changes
Inline Side-by-side
Showing with 269 additions and 840 deletions
master/bt5/slapos_jio_api_style/TestTemplateItem/portal_components/test.erp5.testSlapOSJIOAPI.py
View file @ e44bac78
...@@ -28,18 +28,10 @@ ...@@ -28,18 +28,10 @@
from erp5.component.test.SlapOSTestCaseMixin import SlapOSTestCaseMixin from erp5.component.test.SlapOSTestCaseMixin import SlapOSTestCaseMixin
from DateTime import DateTime from DateTime import DateTime
from App.Common import rfc1123_date
import os
import tempfile
import time import time
import urllib import urllib
# blurb to make nice XML comparisions
import xml.dom.ext.reader.Sax
import xml.dom.ext
import StringIO
import difflib
import hashlib import hashlib
import json import json
from binascii import hexlify from binascii import hexlify
...@@ -692,14 +684,6 @@ class TestSlapOSSlapToolInstanceAccess(TestSlapOSJIOAPIMixin): ...@@ -692,14 +684,6 @@ class TestSlapOSSlapToolInstanceAccess(TestSlapOSJIOAPIMixin):
"portal_type": "Slave Instance", "portal_type": "Slave Instance",
}, instance_dict) }, instance_dict)
def assertInstanceUpdateConnectionSimulator(self, args, kwargs):
stored = eval(open(self.instance_update_connection_simulator).read()) #pylint: disable=eval-used
# do the same translation magic as in workflow
kwargs['connection_xml'] = kwargs.pop('connection_xml')
self.assertEqual(stored,
[{'recargs': args, 'reckwargs': kwargs,
'recmethod': 'updateConnection'}])
def test_13_setConnectionXml_withSlave(self): def test_13_setConnectionXml_withSlave(self):
# XXX CLN No idea how to deal with ascii # XXX CLN No idea how to deal with ascii
self._makeComplexComputeNode(with_slave=True) self._makeComplexComputeNode(with_slave=True)
...@@ -1212,881 +1196,304 @@ class TestSlapOSSlapToolPersonAccess(TestSlapOSJIOAPIMixin): ...@@ -1212,881 +1196,304 @@ class TestSlapOSSlapToolPersonAccess(TestSlapOSJIOAPIMixin):
self.person_user_id = person.getUserId() self.person_user_id = person.getUserId()
TestSlapOSJIOAPIMixin.afterSetUp(self) TestSlapOSJIOAPIMixin.afterSetUp(self)
def deactivated_test_not_accessed_getComputerStatus(self): def test_30_computerBang(self):
self.login(self.person_user_id) self.called_banged_kw = ""
created_at = rfc1123_date(DateTime()) def calledBang(*args, **kw):
since = created_at self.called_banged_kw = kw
response = self.portal_slap.getComputerStatus(self.compute_node_id) start_date = DateTime()
self.assertEqual(200, response.status)
self.assertEqual('public, max-age=60, stale-if-error=604800',
response.headers.get('cache-control'))
self.assertEqual('REMOTE_USER',
response.headers.get('vary'))
self.assertTrue('last-modified' in response.headers)
self.assertEqual('text/xml; charset=utf-8',
response.headers.get('content-type'))
# check returned XML
xml_fp = StringIO.StringIO()
xml.dom.ext.PrettyPrint(xml.dom.ext.reader.Sax.FromXml(response.body),
stream=xml_fp)
xml_fp.seek(0)
got_xml = xml_fp.read()
expected_xml = """\
<?xml version='1.0' encoding='UTF-8'?>
<marshal>
<dictionary id='i2'>
<string>created_at</string>
<string>%(created_at)s</string>
<string>no_data</string>
<int>1</int>
<string>since</string>
<string>%(since)s</string>
<string>state</string>
<string/>
<string>text</string>
<string>#error no data found for %(compute_node_id)s</string>
<string>user</string>
<string>SlapOS Master</string>
</dictionary>
</marshal>
""" % dict(
created_at=created_at,
since=since,
compute_node_id=self.compute_node_id
)
self.assertEqual(expected_xml, got_xml,
'\n'.join([q for q in difflib.unified_diff(expected_xml.split('\n'), got_xml.split('\n'))]))
def deactivated_test_accessed_getComputerStatus(self):
self.login(self.compute_node_user_id)
self.portal_slap.getFullComputerInformation(self.compute_node_id)
self.login(self.person_user_id)
created_at = rfc1123_date(DateTime())
since = created_at
response = self.portal_slap.getComputerStatus(self.compute_node_id)
self.assertEqual(200, response.status)
self.assertEqual('public, max-age=60, stale-if-error=604800',
response.headers.get('cache-control'))
self.assertEqual('REMOTE_USER',
response.headers.get('vary'))
self.assertTrue('last-modified' in response.headers)
self.assertEqual('text/xml; charset=utf-8',
response.headers.get('content-type'))
# check returned XML
xml_fp = StringIO.StringIO()
xml.dom.ext.PrettyPrint(xml.dom.ext.reader.Sax.FromXml(response.body),
stream=xml_fp)
xml_fp.seek(0)
got_xml = xml_fp.read()
expected_xml = """\
<?xml version='1.0' encoding='UTF-8'?>
<marshal>
<dictionary id='i2'>
<string>created_at</string>
<string>%(created_at)s</string>
<string>no_data_since_15_minutes</string>
<int>0</int>
<string>no_data_since_5_minutes</string>
<int>0</int>
<string>since</string>
<string>%(since)s</string>
<string>state</string>
<string/>
<string>text</string>
<string>#access %(compute_node_id)s</string>
<string>user</string>
<string>%(compute_node_id)s</string>
</dictionary>
</marshal>
""" % dict(
created_at=created_at,
since=since,
compute_node_id=self.compute_node_id
)
self.assertEqual(expected_xml, got_xml,
'\n'.join([q for q in difflib.unified_diff(expected_xml.split('\n'), got_xml.split('\n'))]))
def assertComputeNodeBangSimulator(self, args, kwargs):
stored = eval(open(self.compute_node_bang_simulator).read()) #pylint: disable=eval-used
# do the same translation magic as in workflow
kwargs['comment'] = kwargs.pop('comment')
self.assertEqual(stored,
[{'recargs': args, 'reckwargs': kwargs,
'recmethod': 'reportComputeNodeBang'}])
def deactivated_test_computerBang(self):
self.login(self.person_user_id)
self.compute_node_bang_simulator = tempfile.mkstemp()[1]
try: try:
self.compute_node.reportComputeNodeBang = Simulator( reportComputeNodeBang = self.compute_node.__class__.reportComputeNodeBang
self.compute_node_bang_simulator, 'reportComputeNodeBang') self.compute_node.__class__.reportComputeNodeBang = calledBang
self.login(self.person_user_id)
error_log = 'Please bang me' error_log = 'Please bang me'
response = self.portal_slap.computerBang(self.compute_node_id, response = self.putToApi({
error_log) "compute_node_id": self.compute_node_id,
self.assertEqual('None', response) "portal_type": "Compute Node",
# We do not assert getComputerStatus on this test, since "bang_status_message": error_log,
# the change of the timestamp is part of reportComputeNodeBang })
self.assertEqual(self.called_banged_kw, {"comment": error_log})
self.assertComputeNodeBangSimulator((), {'comment': error_log}) self.assertEqual(response["compute_node_id"], self.compute_node.getReference())
self.assertEqual(response["success"], "Done")
self.assertEqual(response["portal_type"], "Compute Node")
self.assertTrue(response["$schema"].endswith("ComputeNode_updateFromJSON/getOutputJSONSchema"))
self.assertTrue(DateTime(response["date"]) >= start_date)
finally: finally:
if os.path.exists(self.compute_node_bang_simulator): self.compute_node.__class__.reportComputeNodeBang = reportComputeNodeBang
os.unlink(self.compute_node_bang_simulator)
def deactivated_test_getComputerPartitionStatus(self): def test_31_getInstanceWithSharedInstance(self, with_slave=True):
self._makeComplexComputeNode() self._makeComplexComputeNode(person=self.person, with_slave=with_slave)
self.login(self.person_user_id) instance = self.start_requested_software_instance
partition_id = self.start_requested_software_instance.getAggregateValue(
portal_type='Compute Partition').getReference()
created_at = rfc1123_date(DateTime())
since = created_at
self.login(self.start_requested_software_instance.getUserId())
response = self.portal_slap.getComputerPartitionStatus(self.compute_node_id,
partition_id)
self.assertEqual(200, response.status)
self.assertEqual( 'public, max-age=60, stale-if-error=604800',
response.headers.get('cache-control'))
self.assertEqual('REMOTE_USER',
response.headers.get('vary'))
self.assertTrue('last-modified' in response.headers)
self.assertEqual('text/xml; charset=utf-8',
response.headers.get('content-type'))
# check returned XML
xml_fp = StringIO.StringIO()
xml.dom.ext.PrettyPrint(xml.dom.ext.reader.Sax.FromXml(response.body),
stream=xml_fp)
xml_fp.seek(0)
got_xml = xml_fp.read()
expected_xml = """\
<?xml version='1.0' encoding='UTF-8'?>
<marshal>
<dictionary id='i2'>
<string>created_at</string>
<string>%(created_at)s</string>
<string>no_data</string>
<int>1</int>
<string>since</string>
<string>%(since)s</string>
<string>state</string>
<string/>
<string>text</string>
<string>#error no data found for %(instance_guid)s</string>
<string>user</string>
<string>SlapOS Master</string>
</dictionary>
</marshal>
""" % dict(
created_at=created_at,
since=since,
instance_guid=self.start_requested_software_instance.getReference(),
)
self.assertEqual(expected_xml, got_xml,
'\n'.join([q for q in difflib.unified_diff(expected_xml.split('\n'), got_xml.split('\n'))]))
def deactivated_test_getComputerPartitionStatus_visited(self):
self._makeComplexComputeNode(person=self.person)
self.login(self.person_user_id)
partition_id = self.start_requested_software_instance.getAggregateValue(
portal_type='Compute Partition').getReference()
created_at = rfc1123_date(DateTime())
since = created_at
self.login(self.start_requested_software_instance.getUserId())
self.portal_slap.registerComputerPartition(self.compute_node_id, partition_id)
self.login(self.person_user_id)
response = self.portal_slap.getComputerPartitionStatus(self.compute_node_id,
partition_id)
self.assertEqual(200, response.status)
self.assertEqual( 'public, max-age=60, stale-if-error=604800',
response.headers.get('cache-control'))
self.assertEqual('REMOTE_USER',
response.headers.get('vary'))
self.assertTrue('last-modified' in response.headers)
self.assertEqual('text/xml; charset=utf-8',
response.headers.get('content-type'))
# check returned XML
xml_fp = StringIO.StringIO()
xml.dom.ext.PrettyPrint(xml.dom.ext.reader.Sax.FromXml(response.body),
stream=xml_fp)
xml_fp.seek(0)
got_xml = xml_fp.read()
expected_xml = """\
<?xml version='1.0' encoding='UTF-8'?>
<marshal>
<dictionary id='i2'>
<string>created_at</string>
<string>%(created_at)s</string>
<string>no_data</string>
<int>1</int>
<string>since</string>
<string>%(since)s</string>
<string>state</string>
<string/>
<string>text</string>
<string>#error no data found for %(instance_guid)s</string>
<string>user</string>
<string>SlapOS Master</string>
</dictionary>
</marshal>
""" % dict(
created_at=created_at,
since=since,
instance_guid=self.start_requested_software_instance.getReference(),
compute_node_id=self.compute_node_id,
partition_id=partition_id
)
self.assertEqual(expected_xml, got_xml,
'\n'.join([q for q in difflib.unified_diff(expected_xml.split('\n'), got_xml.split('\n'))]))
def deactivated_test_registerComputerPartition_withSlave(self):
self._makeComplexComputeNode(person=self.person, with_slave=True)
partition_id = self.start_requested_software_instance.getAggregateValue(
portal_type='Compute Partition').getReference()
self.login(self.person_user_id)
response = self.portal_slap.registerComputerPartition(self.compute_node_id, partition_id)
self.assertEqual(200, response.status)
self.assertEqual( 'public, max-age=1, stale-if-error=604800',
response.headers.get('cache-control'))
self.assertEqual('REMOTE_USER',
response.headers.get('vary'))
self.assertTrue('last-modified' in response.headers)
self.assertEqual('text/xml; charset=utf-8',
response.headers.get('content-type'))
# check returned XML
xml_fp = StringIO.StringIO()
xml.dom.ext.PrettyPrint(xml.dom.ext.reader.Sax.FromXml(response.body),
stream=xml_fp)
xml_fp.seek(0)
got_xml = xml_fp.read()
expected_xml = """\
<?xml version='1.0' encoding='UTF-8'?>
<marshal>
<object id='i2' module='slapos.slap.slap' class='ComputerPartition'>
<tuple>
<string>%(compute_node_id)s</string>
<string>partition1</string>
</tuple>
<dictionary id='i3'>
<string>_computer_id</string>
<string>%(compute_node_id)s</string>
<string>_connection_dict</string>
<dictionary id='i4'/>
<string>_filter_dict</string>
<dictionary id='i5'>
<string>paramé</string>
<string>%(sla)s</string>
</dictionary>
<string>_instance_guid</string>
<string>%(instance_guid)s</string>
<string>_need_modification</string>
<int>1</int>
<string>_parameter_dict</string>
<dictionary id='i6'>
<string>full_ip_list</string>
<list id='i7'/>
<string>instance_title</string>
<string>%(instance_title)s</string>
<string>ip_list</string>
<list id='i8'>
<tuple>
<string/>
<string>ip_address_1</string>
</tuple>
</list>
<string>paramé</string>
<string>%(param)s</string>
<string>root_instance_short_title</string>
<string/>
<string>root_instance_title</string>
<string>%(root_instance_title)s</string>
<string>slap_computer_id</string>
<string>%(compute_node_id)s</string>
<string>slap_computer_partition_id</string>
<string>partition1</string>
<string>slap_software_release_url</string>
<string>%(software_release_url)s</string>
<string>slap_software_type</string>
<string>%(software_type)s</string>
<string>slave_instance_list</string>
<list id='i9'>
<dictionary id='i10'>
<string>connection-parameter-hash</string>
<string>4f53cda18c2baa0c0354bb5f9a3ecbe5ed12ab4d8e11ba873c2f11161202b945</string>
<string>paramé</string>
<string>%(slave_1_param)s</string>
<string>slap_software_type</string>
<string>%(slave_1_software_type)s</string>
<string>slave_reference</string>
<string>%(slave_1_instance_guid)s</string>
<string>slave_title</string>
<string>%(slave_1_title)s</string>
<string>timestamp</string>
<int>%(timestamp)s</int>
</dictionary>
</list>
<string>timestamp</string>
<string>%(timestamp)s</string>
</dictionary>
<string>_partition_id</string>
<string>partition1</string>
<string>_request_dict</string>
<none/>
<string>_requested_state</string>
<string>started</string>
<string>_software_release_document</string>
<object id='i11' module='slapos.slap.slap' class='SoftwareRelease'>
<tuple>
<string>%(software_release_url)s</string>
<string>%(compute_node_id)s</string>
</tuple>
<dictionary id='i12'>
<string>_computer_guid</string>
<string>%(compute_node_id)s</string>
<string>_software_instance_list</string>
<list id='i13'/>
<string>_software_release</string>
<string>%(software_release_url)s</string>
</dictionary>
</object>
<string>_synced</string>
<bool>1</bool>
</dictionary>
</object>
</marshal>
""" % dict(
compute_node_id=self.compute_node_id,
param=self.start_requested_software_instance.getInstanceXmlAsDict()['paramé'],
sla=self.start_requested_software_instance.getSlaXmlAsDict()['paramé'],
software_release_url=self.start_requested_software_instance.getUrlString(),
timestamp=int(self.start_requested_software_instance.getModificationDate()),
instance_guid=self.start_requested_software_instance.getReference(),
instance_title=self.start_requested_software_instance.getTitle(),
root_instance_title=self.start_requested_software_instance.getSpecialiseValue().getTitle(),
software_type=self.start_requested_software_instance.getSourceReference(),
slave_1_param=self.start_requested_slave_instance.getInstanceXmlAsDict()['paramé'],
slave_1_software_type=self.start_requested_slave_instance.getSourceReference(),
slave_1_instance_guid=self.start_requested_slave_instance.getReference(),
slave_1_title=self.start_requested_slave_instance.getTitle(),
)
self.assertEqual(expected_xml, got_xml,
'\n'.join([q for q in difflib.unified_diff(expected_xml.split('\n'), got_xml.split('\n'))]))
def deactivated_test_registerComputerPartition(self):
self._makeComplexComputeNode(person=self.person)
partition_id = self.start_requested_software_instance.getAggregateValue(
portal_type='Compute Partition').getReference()
self.login(self.person_user_id) self.login(self.person_user_id)
response = self.portal_slap.registerComputerPartition(self.compute_node_id, partition_id) instance_dict = self.getToApi({
self.assertEqual(200, response.status) "portal_type": "Software Instance",
self.assertEqual( 'public, max-age=1, stale-if-error=604800', "reference": instance.getReference()
response.headers.get('cache-control')) })
self.assertEqual('REMOTE_USER', response = self.portal.REQUEST.RESPONSE
response.headers.get('vary')) self.assertEqual(200, response.getStatus())
self.assertTrue('last-modified' in response.headers) self.assertEqual('application/json',
self.assertEqual('text/xml; charset=utf-8',
response.headers.get('content-type')) response.headers.get('content-type'))
# check returned XML # Check Data is correct
xml_fp = StringIO.StringIO() self.login()
partition = instance.getAggregateValue(portal_type="Compute Partition")
xml.dom.ext.PrettyPrint(xml.dom.ext.reader.Sax.FromXml(response.body), self.assertEqual({
stream=xml_fp) "$schema": instance.getJSONSchemaUrl(),
xml_fp.seek(0) "title": instance.getTitle(),
got_xml = xml_fp.read() "reference": instance.getReference(),
expected_xml = """\ "software_release_uri": instance.getUrlString(),
<?xml version='1.0' encoding='UTF-8'?> "software_type": instance.getSourceReference(),
<marshal> "state": self.getAPIStateFromSlapState(instance.getSlapState()),
<object id='i2' module='slapos.slap.slap' class='ComputerPartition'> "connection_parameters": instance.getConnectionXmlAsDict(),
<tuple> "parameters": instance.getInstanceXmlAsDict(),
<string>%(compute_node_id)s</string> "shared": False,
<string>partition1</string> "root_instance_title": instance.getSpecialiseValue().getTitle(),
</tuple> "ip_list":
<dictionary id='i3'> [
<string>_computer_id</string> [
<string>%(compute_node_id)s</string> x.getNetworkInterface(''),
<string>_connection_dict</string> x.getIpAddress()
<dictionary id='i4'/> ] for x in partition.contentValues(portal_type='Internet Protocol Address')
<string>_filter_dict</string> ],
<dictionary id='i5'> "full_ip_list": [],
<string>paramé</string> "sla_parameters": instance.getSlaXmlAsDict(),
<string>%(sla)s</string> "compute_node_id": partition.getParentValue().getReference(),
</dictionary> "compute_partition_id": partition.getReference(),
<string>_instance_guid</string> "processing_timestamp": instance.getSlapTimestamp(),
<string>%(instance_guid)s</string> "access_status_message": instance.getTextAccessStatus(),
<string>_need_modification</string> "api_revision": instance.getJIOAPIRevision(self.web_site.api.getRelativeUrl()),
<int>1</int> "portal_type": instance.getPortalType(),
<string>_parameter_dict</string> }, instance_dict)
<dictionary id='i6'>
<string>full_ip_list</string> def test_31_bis_getInstance(self):
<list id='i7'/> self.test_31_getInstanceWithSharedInstance(with_slave=False)
<string>instance_title</string>
<string>%(instance_title)s</string>
<string>ip_list</string> def test_32_softwareInstanceBang(self):
<list id='i8'>
<tuple>
<string/>
<string>ip_address_1</string>
</tuple>
</list>
<string>paramé</string>
<string>%(param)s</string>
<string>root_instance_short_title</string>
<string/>
<string>root_instance_title</string>
<string>%(root_instance_title)s</string>
<string>slap_computer_id</string>
<string>%(compute_node_id)s</string>
<string>slap_computer_partition_id</string>
<string>partition1</string>
<string>slap_software_release_url</string>
<string>%(software_release_url)s</string>
<string>slap_software_type</string>
<string>%(software_type)s</string>
<string>slave_instance_list</string>
<list id='i9'/>
<string>timestamp</string>
<string>%(timestamp)s</string>
</dictionary>
<string>_partition_id</string>
<string>partition1</string>
<string>_request_dict</string>
<none/>
<string>_requested_state</string>
<string>started</string>
<string>_software_release_document</string>
<object id='i10' module='slapos.slap.slap' class='SoftwareRelease'>
<tuple>
<string>%(software_release_url)s</string>
<string>%(compute_node_id)s</string>
</tuple>
<dictionary id='i11'>
<string>_computer_guid</string>
<string>%(compute_node_id)s</string>
<string>_software_instance_list</string>
<list id='i12'/>
<string>_software_release</string>
<string>%(software_release_url)s</string>
</dictionary>
</object>
<string>_synced</string>
<bool>1</bool>
</dictionary>
</object>
</marshal>
""" % dict(
compute_node_id=self.compute_node_id,
param=self.start_requested_software_instance.getInstanceXmlAsDict()['paramé'],
sla=self.start_requested_software_instance.getSlaXmlAsDict()['paramé'],
software_release_url=self.start_requested_software_instance.getUrlString(),
timestamp=int(self.start_requested_software_instance.getModificationDate()),
instance_guid=self.start_requested_software_instance.getReference(),
instance_title=self.start_requested_software_instance.getTitle(),
root_instance_title=self.start_requested_software_instance.getSpecialiseValue().getTitle(),
software_type=self.start_requested_software_instance.getSourceReference()
)
self.assertEqual(expected_xml, got_xml,
'\n'.join([q for q in difflib.unified_diff(expected_xml.split('\n'), got_xml.split('\n'))]))
def assertInstanceBangSimulator(self, args, kwargs):
stored = eval(open(self.instance_bang_simulator).read()) #pylint: disable=eval-used
# do the same translation magic as in workflow
kwargs['comment'] = kwargs.pop('comment')
self.assertEqual(stored,
[{'recargs': args, 'reckwargs': kwargs,
'recmethod': 'bang'}])
def deactivated_test_softwareInstanceBang(self):
self._makeComplexComputeNode(person=self.person) self._makeComplexComputeNode(person=self.person)
self.instance_bang_simulator = tempfile.mkstemp()[1] self.called_instance_bang = ""
def calledBanged(*args, **kw):
self.called_instance_bang = kw
instance = self.start_requested_software_instance
error_log = 'Please bang me'
try: try:
partition_id = self.start_requested_software_instance.getAggregateValue( bang = instance.__class__.bang
portal_type='Compute Partition').getReference() instance.__class__.bang = calledBanged
self.login(self.person_user_id) self.login(self.person_user_id)
self.start_requested_software_instance.bang = Simulator(
self.instance_bang_simulator, 'bang')
error_log = 'Please bang me' error_log = 'Please bang me'
response = self.portal_slap.softwareInstanceBang(self.compute_node_id, self.callInstancePutToApiAndCheck({
partition_id, error_log) "reference": instance.getReference(),
self.assertEqual('OK', response) "portal_type": "Software Instance",
created_at = rfc1123_date(DateTime()) "reported_state": "bang",
since = created_at "status_message": error_log,
response = self.portal_slap.getComputerPartitionStatus(self.compute_node_id, })
partition_id) self.assertEqual(
# check returned XML self.called_instance_bang,
xml_fp = StringIO.StringIO() {'bang_tree': True, 'comment': 'Please bang me'}
xml.dom.ext.PrettyPrint(xml.dom.ext.reader.Sax.FromXml(response.body),
stream=xml_fp)
xml_fp.seek(0)
got_xml = xml_fp.read()
expected_xml = """\
<?xml version='1.0' encoding='UTF-8'?>
<marshal>
<dictionary id='i2'>
<string>created_at</string>
<string>%(created_at)s</string>
<string>no_data_since_15_minutes</string>
<int>0</int>
<string>no_data_since_5_minutes</string>
<int>0</int>
<string>since</string>
<string>%(since)s</string>
<string>state</string>
<string/>
<string>text</string>
<string>#error bang called</string>
<string>user</string>
<string>%(person_reference)s</string>
</dictionary>
</marshal>
""" % dict(
created_at=created_at,
since=since,
person_reference=self.person_reference,
) )
self.assertEqual(expected_xml, got_xml,
'\n'.join([q for q in difflib.unified_diff(expected_xml.split('\n'), got_xml.split('\n'))]))
self.assertInstanceBangSimulator((), {'comment': error_log, 'bang_tree': True})
finally: finally:
if os.path.exists(self.instance_bang_simulator): instance.__class__.bang = bang
os.unlink(self.instance_bang_simulator)
def assertInstanceRenameSimulator(self, args, kwargs): instance_dict = self.getToApi({
stored = eval(open(self.instance_rename_simulator).read()) #pylint: disable=eval-used "portal_type": "Software Instance",
# do the same translation magic as in workflow "reference": instance.getReference(),
self.assertEqual(stored, })
[{'recargs': args, 'reckwargs': kwargs, response = self.portal.REQUEST.RESPONSE
'recmethod': 'rename'}]) self.assertEqual(200, response.getStatus())
self.assertEqual('application/json',
response.headers.get('content-type'))
self.assertEqual(
instance_dict["access_status_message"],
"#error bang called"
)
def deactivated_test_softwareInstanceRename(self): def test_33_softwareInstanceRename(self):
self._makeComplexComputeNode(person=self.person) self._makeComplexComputeNode(person=self.person)
self.instance_rename_simulator = tempfile.mkstemp()[1] new_name = 'new me'
self.called_instance_rename = ""
def calledRename(*args, **kw):
self.called_instance_rename = kw
instance = self.start_requested_software_instance
try: try:
partition_id = self.start_requested_software_instance.getAggregateValue( rename = instance.__class__.rename
portal_type='Compute Partition').getReference() instance.__class__.rename = calledRename
self.login(self.person_user_id) self.login(self.person_user_id)
self.start_requested_software_instance.rename = Simulator( self.callInstancePutToApiAndCheck({
self.instance_rename_simulator, 'rename') "reference": instance.getReference(),
new_name = 'new me' "portal_type": "Software Instance",
response = self.portal_slap.softwareInstanceRename(new_name, self.compute_node_id, "title": new_name,
partition_id) })
self.assertEqual('None', response) self.assertEqual(self.called_instance_rename,
self.assertInstanceRenameSimulator((), { {
'comment': 'Rename %s into %s' % (self.start_requested_software_instance.getTitle(), 'comment': 'Rename %s into %s' % (
new_name), 'new_name': new_name}) instance.getTitle(),
new_name
),
'new_name': new_name
}
)
finally: finally:
if os.path.exists(self.instance_rename_simulator): instance.__class__.rename = rename
os.unlink(self.instance_rename_simulator)
def test_34_request_withSlave(self):
def assertInstanceRequestSimulator(self, args, kwargs): self.called_instance_request = ""
stored = eval(open(self.instance_request_simulator).read()) #pylint: disable=eval-used def calledRequestInstance(*args, **kw):
# do the same translation magic as in workflow self.called_instance_request = kw
self.assertEqual(stored,
[{'recargs': args, 'reckwargs': kwargs,
'recmethod': 'requestSoftwareInstance'}])
def deactivated_test_request_withSlave(self):
self.instance_request_simulator = tempfile.mkstemp()[1]
try: try:
requestSoftwareInstance = self.person.__class__.requestSoftwareInstance
self.person.__class__.requestSoftwareInstance = calledRequestInstance
self.login(self.person_user_id) self.login(self.person_user_id)
self.person.requestSoftwareInstance = Simulator( response_dict = self.postToApi({
self.instance_request_simulator, 'requestSoftwareInstance') "portal_type": "Software Instance",
response = self.portal_slap.requestComputerPartition( "software_release_uri": "req_release",
software_release='req_release', "software_type": "req_type",
software_type='req_type', "title": "req_reference",
partition_reference='req_reference', "shared": True,
partition_parameter_xml='<marshal><dictionary id="i2"/></marshal>', })
filter_xml='<marshal><dictionary id="i2"/></marshal>', response = self.portal.REQUEST.RESPONSE
state='<marshal><string>started</string></marshal>', self.assertEqual(400, response.getStatus())
shared_xml='<marshal><bool>1</bool></marshal>', self.assertEqual('application/json',
) response.headers.get('content-type'))
self.assertEqual(408, response.status) self.assertEqual(self.called_instance_request, {
self.assertEqual('private',
response.headers.get('cache-control'))
self.assertInstanceRequestSimulator((), {
'instance_xml': "<?xml version='1.0' encoding='utf-8'?>\n<instance/>\n", 'instance_xml': "<?xml version='1.0' encoding='utf-8'?>\n<instance/>\n",
'software_title': 'req_reference', 'software_title': 'req_reference',
'software_release': 'req_release', 'software_release': 'req_release',
'state': 'started', 'state': 'started',
'sla_xml': "<?xml version='1.0' encoding='utf-8'?>\n<instance/>\n", 'sla_xml': "<?xml version='1.0' encoding='utf-8'?>\n<instance/>\n",
'software_type': 'req_type', 'software_type': 'req_type',
'shared': True}) 'shared': True
})
self.assertTrue(response_dict.pop("$schema").endswith("error-response-schema.json"))
response_dict.pop("debug_id")
self.assertEqual(response_dict, {
'message': 'Software Instance Not Ready',
'name': 'SoftwareInstanceNotReady',
'status': 102
})
finally: finally:
if os.path.exists(self.instance_request_simulator): self.person.__class__.requestSoftwareInstance = requestSoftwareInstance
os.unlink(self.instance_request_simulator)
def test_35_request(self):
self.called_instance_request = ""
def calledRequestInstance(*args, **kw):
self.called_instance_request = kw
def deactivated_test_request(self):
self.instance_request_simulator = tempfile.mkstemp()[1]
try: try:
requestSoftwareInstance = self.person.__class__.requestSoftwareInstance
self.person.__class__.requestSoftwareInstance = calledRequestInstance
self.login(self.person_user_id) self.login(self.person_user_id)
self.person.requestSoftwareInstance = Simulator( response_dict = self.postToApi({
self.instance_request_simulator, 'requestSoftwareInstance') "portal_type": "Software Instance",
response = self.portal_slap.requestComputerPartition( "software_release_uri": "req_release",
software_release='req_release', "software_type": "req_type",
software_type='req_type', "title": "req_reference",
partition_reference='req_reference', })
partition_parameter_xml='<marshal><dictionary id="i2"/></marshal>', response = self.portal.REQUEST.RESPONSE
filter_xml='<marshal><dictionary id="i2"/></marshal>', self.assertEqual(400, response.getStatus())
state='<marshal><string>started</string></marshal>', self.assertEqual('application/json',
shared_xml='<marshal><bool>0</bool></marshal>', response.headers.get('content-type'))
) self.assertEqual(self.called_instance_request, {
self.assertEqual(408, response.status)
self.assertEqual('private',
response.headers.get('cache-control'))
self.assertInstanceRequestSimulator((), {
'instance_xml': "<?xml version='1.0' encoding='utf-8'?>\n<instance/>\n", 'instance_xml': "<?xml version='1.0' encoding='utf-8'?>\n<instance/>\n",
'software_title': 'req_reference', 'software_title': 'req_reference',
'software_release': 'req_release', 'software_release': 'req_release',
'state': 'started', 'state': 'started',
'sla_xml': "<?xml version='1.0' encoding='utf-8'?>\n<instance/>\n", 'sla_xml': "<?xml version='1.0' encoding='utf-8'?>\n<instance/>\n",
'software_type': 'req_type', 'software_type': 'req_type',
'shared': False}) 'shared': False
})
self.assertTrue(response_dict.pop("$schema").endswith("error-response-schema.json"))
response_dict.pop("debug_id")
self.assertEqual(response_dict, {
'message': 'Software Instance Not Ready',
'name': 'SoftwareInstanceNotReady',
'status': 102
})
finally: finally:
if os.path.exists(self.instance_request_simulator): self.person.__class__.requestSoftwareInstance = requestSoftwareInstance
os.unlink(self.instance_request_simulator)
def deactivated_test_request_allocated_instance(self): def test_36_request_allocated_instance(self):
self.tic() self.tic()
self.person.edit( self.person.edit(
default_email_coordinate_text="%s@example.org" % self.person.getReference(), default_email_coordinate_text="%s@example.org" % self.person.getReference(),
career_role='member', career_role='member',
) )
self._makeComplexComputeNode(person=self.person) self._makeComplexComputeNode(person=self.person)
self.start_requested_software_instance.updateLocalRolesOnSecurityGroups() instance = self.start_requested_software_instance
instance.updateLocalRolesOnSecurityGroups()
self.tic() self.tic()
self.login(self.person_user_id) self.login(self.person_user_id)
response = self.portal_slap.requestComputerPartition( instance_dict = self.postToApi({
software_release=self.start_requested_software_instance.getUrlString(), "portal_type": "Software Instance",
software_type=self.start_requested_software_instance.getSourceReference(), "software_release_uri": instance.getUrlString(),
partition_reference=self.start_requested_software_instance.getTitle(), "software_type": instance.getSourceReference(),
partition_parameter_xml='<marshal><dictionary id="i2"/></marshal>', "title": instance.getTitle(),
filter_xml='<marshal><dictionary id="i2"/></marshal>', })
state='<marshal><string>started</string></marshal>',
shared_xml='<marshal><bool>0</bool></marshal>', response = self.portal.REQUEST.RESPONSE
) self.assertEqual(201, response.getStatus())
self.assertEqual(type(response), str) self.assertEqual('application/json',
# check returned XML response.headers.get('content-type'))
xml_fp = StringIO.StringIO() # Check Data is correct
self.login()
xml.dom.ext.PrettyPrint(xml.dom.ext.reader.Sax.FromXml(response), partition = instance.getAggregateValue(portal_type="Compute Partition")
stream=xml_fp) self.assertEqual({
xml_fp.seek(0) "$schema": instance.getJSONSchemaUrl(),
got_xml = xml_fp.read() "title": instance.getTitle(),
expected_xml = """\ "reference": instance.getReference(),
<?xml version='1.0' encoding='UTF-8'?> "software_release_uri": instance.getUrlString(),
<marshal> "software_type": instance.getSourceReference(),
<object id='i2' module='slapos.slap.slap' class='SoftwareInstance'> "state": self.getAPIStateFromSlapState(instance.getSlapState()),
<tuple/> "connection_parameters": instance.getConnectionXmlAsDict(),
<dictionary id='i3'> "parameters": instance.getInstanceXmlAsDict(),
<string>_connection_dict</string> "shared": False,
<dictionary id='i4'/> "root_instance_title": instance.getSpecialiseValue().getTitle(),
<string>_filter_dict</string> "ip_list":
<dictionary id='i5'/> [
<string>_instance_guid</string> [
<string>%(instance_guid)s</string> x.getNetworkInterface(''),
<string>_parameter_dict</string> x.getIpAddress()
<dictionary id='i6'/> ] for x in partition.contentValues(portal_type='Internet Protocol Address')
<string>_requested_state</string> ],
<string>%(state)s</string> "full_ip_list": [],
<string>full_ip_list</string> "sla_parameters": instance.getSlaXmlAsDict(),
<list id='i7'/> "compute_node_id": partition.getParentValue().getReference(),
<string>instance_title</string> "compute_partition_id": partition.getReference(),
<string>%(instance_title)s</string> "processing_timestamp": instance.getSlapTimestamp(),
<string>ip_list</string> "access_status_message": instance.getTextAccessStatus(),
<list id='i8'> "api_revision": instance.getJIOAPIRevision(self.web_site.api.getRelativeUrl()),
<tuple> "portal_type": instance.getPortalType(),
<string/> }, instance_dict)
<string>%(ip)s</string>
</tuple> def test_37_ComputeNodeSupply(self):
</list> self.called_software_supply = ""
<string>root_instance_short_title</string> def calledSoftwareSupply(*args, **kw):
<string/> self.called_software_supply = kw
<string>root_instance_title</string>
<string>%(root_instance_title)s</string>
<string>slap_computer_id</string>
<string>%(compute_node_id)s</string>
<string>slap_computer_partition_id</string>
<string>%(partition_id)s</string>
<string>slap_software_release_url</string>
<string>%(url_string)s</string>
<string>slap_software_type</string>
<string>%(type)s</string>
<string>slave_instance_list</string>
<list id='i9'/>
<string>timestamp</string>
<string>%(timestamp)s</string>
</dictionary>
</object>
</marshal>
""" % dict(
instance_guid=self.start_requested_software_instance.getReference(),
instance_title=self.start_requested_software_instance.getTitle(),
root_instance_title=self.start_requested_software_instance.getSpecialiseTitle(),
state="started",
url_string=self.start_requested_software_instance.getUrlString(),
type=self.start_requested_software_instance.getSourceReference(),
timestamp=int(self.start_requested_software_instance.getModificationDate()),
compute_node_id=self.compute_node_id,
partition_id=self.start_requested_software_instance.getAggregateId(),
ip=self.start_requested_software_instance.getAggregateValue()\
.getDefaultNetworkAddressIpAddress(),
)
self.assertEqual(expected_xml, got_xml,
'\n'.join([q for q in difflib.unified_diff(expected_xml.split('\n'),
got_xml.split('\n'))]))
def assertSupplySimulator(self, args, kwargs):
stored = eval(open(self.compute_node_supply_simulator).read()) #pylint: disable=eval-used
# do the same translation magic as in workflow
kwargs['software_release_url'] = kwargs.pop('software_release_url')
kwargs['state'] = kwargs.pop('state')
self.assertEqual(stored,
[{'recargs': args, 'reckwargs': kwargs,
'recmethod': 'requestSoftwareRelease'}])
def deactivated_test_ComputeNodeSupply(self):
self.compute_node_supply_simulator = tempfile.mkstemp()[1]
try: try:
self.login(self.person_user_id) self.login(self.person_user_id)
self.compute_node.requestSoftwareRelease = Simulator( requestSoftwareRelease = self.compute_node.__class__.requestSoftwareRelease
self.compute_node_supply_simulator, 'requestSoftwareRelease') self.compute_node.__class__.requestSoftwareRelease = calledSoftwareSupply
software_url = 'live_test_url_%s' % self.generateNewId() software_url = 'live_test_url_%s' % self.generateNewId()
response = self.portal_slap.supplySupply( self.postToApi({
software_url, "portal_type": "Software Installation",
self.compute_node_id, "compute_node_id": self.compute_node.getReference(),
state='destroyed') "software_release_uri": urllib.quote(software_url)
self.assertEqual('None', response) })
self.assertSupplySimulator((), { response = self.portal.REQUEST.RESPONSE
self.assertEqual(201, response.getStatus())
self.assertEqual('application/json',
response.headers.get('content-type'))
self.assertEqual(self.called_software_supply, {
'software_release_url': software_url, 'software_release_url': software_url,
'state': 'destroyed'}) 'state': 'available'})
finally: finally:
if os.path.exists(self.compute_node_supply_simulator): self.compute_node.__class__.requestSoftwareRelease = requestSoftwareRelease
os.unlink(self.compute_node_supply_simulator)
def assertRequestComputeNodeSimulator(self, args, kwargs):
stored = eval(open(self.compute_node_request_compute_node_simulator).read()) #pylint: disable=eval-used
# do the same translation magic as in workflow
kwargs['compute_node_title'] = kwargs.pop('compute_node_title')
self.assertEqual(stored,
[{'recargs': args, 'reckwargs': kwargs,
'recmethod': 'requestComputeNode'}])
def deactivated_test_requestComputeNode(self):
self.compute_node_request_compute_node_simulator = tempfile.mkstemp()[1]
try:
self.login(self.person_user_id)
self.person.requestComputeNode = Simulator(
self.compute_node_request_compute_node_simulator, 'requestComputeNode')
compute_node_id = 'Foo Compute Node'
compute_node_reference = 'live_comp_%s' % self.generateNewId()
self.portal.REQUEST.set('compute_node_reference', compute_node_reference)
response = self.portal_slap.requestComputer(compute_node_id)
# check returned XML
xml_fp = StringIO.StringIO()
xml.dom.ext.PrettyPrint(xml.dom.ext.reader.Sax.FromXml(response),
stream=xml_fp)
xml_fp.seek(0)
got_xml = xml_fp.read()
expected_xml = """\
<?xml version='1.0' encoding='UTF-8'?>
<marshal>
<object id='i2' module='slapos.slap.slap' class='Computer'>
<tuple>
<string>%(compute_node_id)s</string>
</tuple>
<dictionary id='i3'>
<string>_computer_id</string>
<string>%(compute_node_id)s</string>
</dictionary>
</object>
</marshal>
""" % {'compute_node_id': compute_node_reference}
self.assertEqual(expected_xml, got_xml,
'\n'.join([q for q in difflib.unified_diff(expected_xml.split('\n'), got_xml.split('\n'))]))
self.assertRequestComputeNodeSimulator((), {'compute_node_title': compute_node_id})
finally:
if os.path.exists(self.compute_node_request_compute_node_simulator):
os.unlink(self.compute_node_request_compute_node_simulator)
def assertGenerateComputeNodeCertificateSimulator(self, args, kwargs):
stored = eval(open(self.generate_compute_node_certificate_simulator).read()) #pylint: disable=eval-used
# do the same translation magic as in workflow
self.assertEqual(stored,
[{'recargs': args, 'reckwargs': kwargs,
'recmethod': 'generateComputerCertificate'}])
def deactivated_test_generateComputerCertificate(self):
self.generate_compute_node_certificate_simulator = tempfile.mkstemp()[1]
try:
self.login(self.person_user_id)
self.compute_node.generateCertificate = Simulator(
self.generate_compute_node_certificate_simulator,
'generateComputerCertificate')
compute_node_certificate = 'live_\ncertificate_%s' % self.generateNewId()
compute_node_key = 'live_\nkey_%s' % self.generateNewId()
self.portal.REQUEST.set('compute_node_certificate', compute_node_certificate)
self.portal.REQUEST.set('compute_node_key', compute_node_key)
response = self.portal_slap.generateComputerCertificate(self.compute_node_id)
# check returned XML
xml_fp = StringIO.StringIO()
xml.dom.ext.PrettyPrint(xml.dom.ext.reader.Sax.FromXml(response),
stream=xml_fp)
xml_fp.seek(0)
got_xml = xml_fp.read()
expected_xml = """\
<?xml version='1.0' encoding='UTF-8'?>
<marshal>
<dictionary id='i2'>
<string>certificate</string>
<string>%(compute_node_certificate)s</string>
<string>key</string>
<string>%(compute_node_key)s</string>
</dictionary>
</marshal>
""" % {'compute_node_key': compute_node_key, 'compute_node_certificate': compute_node_certificate}
self.assertEqual(expected_xml, got_xml,
'\n'.join([q for q in difflib.unified_diff(expected_xml.split('\n'), got_xml.split('\n'))]))
self.assertGenerateComputeNodeCertificateSimulator((), {})
finally:
if os.path.exists(self.generate_compute_node_certificate_simulator):
os.unlink(self.generate_compute_node_certificate_simulator)
def assertRevokeComputeNodeCertificateSimulator(self, args, kwargs):
stored = eval(open(self.revoke_compute_node_certificate_simulator).read()) #pylint: disable=eval-used
# do the same translation magic as in workflow
self.assertEqual(stored,
[{'recargs': args, 'reckwargs': kwargs,
'recmethod': 'revokeComputerCertificate'}])
def deactivated_test_revokeComputerCertificate(self):
self.revoke_compute_node_certificate_simulator = tempfile.mkstemp()[1]
try:
self.login(self.person_user_id)
self.compute_node.revokeCertificate = Simulator(
self.revoke_compute_node_certificate_simulator,
'revokeComputerCertificate')
response = self.portal_slap.revokeComputerCertificate(self.compute_node_id) def test_38_getHateoasUrl_NotConfigured(self):
self.assertEqual('None', response)
self.assertRevokeComputeNodeCertificateSimulator((), {})
finally:
if os.path.exists(self.revoke_compute_node_certificate_simulator):
os.unlink(self.revoke_compute_node_certificate_simulator)
def deactivated_test_getHateoasUrl_NotConfigured(self):
for preference in \ for preference in \
self.portal.portal_catalog(portal_type="System Preference"): self.portal.portal_catalog(portal_type="System Preference"):
preference = preference.getObject() preference = preference.getObject()
...@@ -2096,7 +1503,7 @@ class TestSlapOSSlapToolPersonAccess(TestSlapOSJIOAPIMixin): ...@@ -2096,7 +1503,7 @@ class TestSlapOSSlapToolPersonAccess(TestSlapOSJIOAPIMixin):
self.login(self.person_user_id) self.login(self.person_user_id)
self.assertRaises(NotFound, self.portal_slap.getHateoasUrl) self.assertRaises(NotFound, self.portal_slap.getHateoasUrl)
def deactivated_test_getHateoasUrl(self): def test_39_getHateoasUrl(self):
for preference in \ for preference in \
self.portal.portal_catalog(portal_type="System Preference"): self.portal.portal_catalog(portal_type="System Preference"):
preference = preference.getObject() preference = preference.getObject()
...@@ -2107,3 +1514,25 @@ class TestSlapOSSlapToolPersonAccess(TestSlapOSJIOAPIMixin): ...@@ -2107,3 +1514,25 @@ class TestSlapOSSlapToolPersonAccess(TestSlapOSJIOAPIMixin):
response = self.portal_slap.getHateoasUrl() response = self.portal_slap.getHateoasUrl()
self.assertEqual(200, response.status) self.assertEqual(200, response.status)
self.assertEqual('foo', response.body) self.assertEqual('foo', response.body)
def test_40_getJIOAPIUrl_NotConfigured(self):
for preference in \
self.portal.portal_catalog(portal_type="System Preference"):
preference = preference.getObject()
if preference.getPreferenceState() == 'global':
preference.setPreferredJioApiUrl('')
self.tic()
self.login(self.person_user_id)
self.assertRaises(NotFound, self.portal_slap.getJIOAPIUrl)
def test_41_getJIOAPIUrl(self):
for preference in \
self.portal.portal_catalog(portal_type="System Preference"):
preference = preference.getObject()
if preference.getPreferenceState() == 'global':
preference.setPreferredJioApiUrl('bar')
self.tic()
self.login(self.person_user_id)
response = self.portal_slap.getJIOAPIUrl()
self.assertEqual(200, response.status)
self.assertEqual('bar', response.body)
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