Skip to content

S
slapos.core
Commit 4268d029 authored by Cédric Le Ninivin's avatar Cédric Le Ninivin Committed by Titouan Soulard
Browse files
Options

WIP slapos_jio_api_style: Continue Testing, this time instance interaction

parent 99365d61
Show whitespace changes
Inline Side-by-side
Showing with 190 additions and 545 deletions
master/bt5/slapos_jio_api_style/TestTemplateItem/portal_components/test.erp5.testSlapOSJIOAPI.py
View file @ 4268d029
...@@ -546,541 +546,173 @@ class TestSlapOSSlapToolComputeNodeAccess(TestSlapOSJIOAPIMixin): ...@@ -546,541 +546,173 @@ class TestSlapOSSlapToolComputeNodeAccess(TestSlapOSJIOAPIMixin):
self.assertEqual(expected_dict, software_dict) self.assertEqual(expected_dict, software_dict)
class TestSlapOSSlapToolInstanceAccess(TestSlapOSJIOAPIMixin): class TestSlapOSSlapToolInstanceAccess(TestSlapOSJIOAPIMixin):
def deactivated_test_getComputerPartitionCertificate(self): def test_10_getComputerPartitionCertificate(self):
self._makeComplexComputeNode() self._makeComplexComputeNode()
partition_id = self.start_requested_software_instance.getAggregateValue( self.callUpdateRevisionAndTic()
portal_type='Compute Partition').getReference()
self.login(self.start_requested_software_instance.getUserId())
response = self.portal_slap.getComputerPartitionCertificate(self.compute_node_id,
partition_id)
self.assertEqual(200, response.status)
self.assertEqual( 'public, max-age=0, must-revalidate',
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>certificate</string>
<string>%(instance_certificate)s</string>
<string>key</string>
<string>%(instance_key)s</string>
</dictionary>
</marshal>
""" % dict(
instance_certificate=self.start_requested_software_instance.getSslCertificate(),
instance_key=self.start_requested_software_instance.getSslKey()
)
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_getFullComputerInformation(self):
self._makeComplexComputeNode(with_slave=True)
self.login(self.start_requested_software_instance.getUserId())
response = self.portal_slap.getFullComputerInformation(self.compute_node_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.assertFalse('etag' 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='Computer'>
<tuple>
<string>%(compute_node_id)s</string>
</tuple>
<dictionary id='i3'>
<string>_computer_id</string>
<string>%(compute_node_id)s</string>
<string>_computer_partition_list</string>
<list id='i4'>
<object id='i5' module='slapos.slap.slap' class='ComputerPartition'>
<tuple>
<string>%(compute_node_id)s</string>
<string>partition1</string>
</tuple>
<dictionary id='i6'>
<string>_access_status</string>
<string>%(access_status)s</string>
<string>_computer_id</string>
<string>%(compute_node_id)s</string>
<string>_connection_dict</string>
<dictionary id='i7'/>
<string>_filter_dict</string>
<dictionary id='i8'>
<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='i9'>
<string>full_ip_list</string>
<list id='i10'/>
<string>instance_title</string>
<string>%(instance_title)s</string>
<string>ip_list</string>
<list id='i11'>
<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='i12'>
<dictionary id='i13'>
<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='i14' module='slapos.slap.slap' class='SoftwareRelease'>
<tuple>
<string>%(software_release_url)s</string>
<string>%(compute_node_id)s</string>
</tuple>
<dictionary id='i15'>
<string>_computer_guid</string>
<string>%(compute_node_id)s</string>
<string>_software_instance_list</string>
<list id='i16'/>
<string>_software_release</string>
<string>%(software_release_url)s</string>
</dictionary>
</object>
</dictionary>
</object>
</list>
<string>_software_release_list</string>
<list id='i17'/>
</dictionary>
</object>
</marshal>
""" % dict(
compute_node_id=self.compute_node_id,
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_release_url=self.start_requested_software_instance.getUrlString(),
software_type=self.start_requested_software_instance.getSourceReference(),
param=self.start_requested_software_instance.getInstanceXmlAsDict()['paramé'],
sla=self.start_requested_software_instance.getSlaXmlAsDict()['paramé'],
timestamp=int(self.start_requested_software_instance.getModificationDate()),
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(),
access_status="#error no data found for %s" % 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(self):
self._makeComplexComputeNode()
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()
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)
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(with_slave=True)
partition_id = self.start_requested_software_instance.getAggregateValue(
portal_type='Compute Partition').getReference()
self.login(self.start_requested_software_instance.getUserId())
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()
partition_id = self.start_requested_software_instance.getAggregateValue(
portal_type='Compute Partition').getReference()
self.login(self.start_requested_software_instance.getUserId()) self.login(self.start_requested_software_instance.getUserId())
response = self.portal_slap.registerComputerPartition(self.compute_node_id, partition_id) certificate_dict = self.getToApi({
self.assertEqual(200, response.status) "portal_type": "Software Instance Certificate Record",
self.assertEqual( 'public, max-age=1, stale-if-error=604800', "reference": self.start_requested_software_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 self.assertTrue(
xml_fp = StringIO.StringIO() certificate_dict.pop("$schema").endswith("SoftwareInstanceCertificateRecord_getFromJSON/getOutputJSONSchema")
)
self.assertEqual(certificate_dict, {
"key" :self.start_requested_software_instance.getSslKey(),
"certificate": self.start_requested_software_instance.getSslCertificate(),
"portal_type": "Software Instance Certificate Record",
"reference": self.start_requested_software_instance.getReference(),
})
xml.dom.ext.PrettyPrint(xml.dom.ext.reader.Sax.FromXml(response.body), def test_11_getFullComputerInformationWithSharedInstance(self, with_slave=True):
stream=xml_fp) self._makeComplexComputeNode(with_slave=with_slave)
xml_fp.seek(0) self.callUpdateRevisionAndTic()
got_xml = xml_fp.read() instance = self.start_requested_software_instance
expected_xml = """\ self.login(instance.getUserId())
<?xml version='1.0' encoding='UTF-8'?> instance_list_response = self.allDocsToApi({
<marshal> "compute_node_id": self.compute_node_id,
<object id='i2' module='slapos.slap.slap' class='ComputerPartition'> "portal_type": "Software Instance",
<tuple> })
<string>%(compute_node_id)s</string> response = self.portal.REQUEST.RESPONSE
<string>partition1</string> self.assertEqual(200, response.getStatus())
</tuple> self.assertEqual('application/json',
<dictionary id='i3'> response.headers.get('content-type'))
<string>_computer_id</string>
<string>%(compute_node_id)s</string> self.assertTrue(instance_list_response["$schema"].endswith("jIOWebSection_searchInstanceFromJSON/getOutputJSONSchema"))
<string>_connection_dict</string> result_list = instance_list_response["result_list"]
<dictionary id='i4'/> self.assertEqual(1, len(result_list))
<string>_filter_dict</string>
<dictionary id='i5'> self.login()
<string>paramé</string> # Check result_list match instance_list=
<string>%(sla)s</string> expected_instance_list = [{
</dictionary> "api_revision": instance.getJIOAPIRevision(self.web_site.api.getRelativeUrl()),
<string>_instance_guid</string> "compute_partition_id": instance.getAggregateReference(),
<string>%(instance_guid)s</string> "get_parameters": {
<string>_need_modification</string> "portal_type": "Software Instance",
<int>1</int> "reference": instance.getReference(),
<string>_parameter_dict</string> },
<dictionary id='i6'> "portal_type": "Software Instance",
<string>full_ip_list</string> "reference": instance.getReference(),
<list id='i7'/> "software_release_uri": instance.getUrlString(),
<string>instance_title</string> "state": self.getAPIStateFromSlapState(instance.getSlapState()),
<string>%(instance_title)s</string> "title": instance.getTitle(),
<string>ip_list</string> }]
<list id='i8'> self.assertEqual(expected_instance_list, instance_list_response["result_list"])
<tuple>
<string/> instance_resut_dict = expected_instance_list[0]
<string>ip_address_1</string> # Get instance as "user"
</tuple> self.login(instance.getUserId())
</list> instance_dict = self.getToApi(instance_resut_dict["get_parameters"])
<string>paramé</string> response = self.portal.REQUEST.RESPONSE
<string>%(param)s</string> self.assertEqual(200, response.getStatus())
<string>root_instance_short_title</string> self.assertEqual('application/json',
<string/> response.headers.get('content-type'))
<string>root_instance_title</string> # Check Data is correct
<string>%(root_instance_title)s</string> self.login()
<string>slap_computer_id</string> partition = instance.getAggregateValue(portal_type="Compute Partition")
<string>%(compute_node_id)s</string> self.assertEqual({
<string>slap_computer_partition_id</string> "$schema": instance.getJSONSchemaUrl(),
<string>partition1</string> "title": instance.getTitle(),
<string>slap_software_release_url</string> "reference": instance.getReference(),
<string>%(software_release_url)s</string> "software_release_uri": instance.getUrlString(),
<string>slap_software_type</string> "software_type": instance.getSourceReference(),
<string>%(software_type)s</string> "state": self.getAPIStateFromSlapState(instance.getSlapState()),
<string>slave_instance_list</string> "connection_parameters": instance.getConnectionXmlAsDict(),
<list id='i9'/> "parameters": instance.getInstanceXmlAsDict(),
<string>timestamp</string> "shared": False,
<string>%(timestamp)s</string> "root_instance_title": instance.getSpecialiseValue().getTitle(),
</dictionary> "ip_list":
<string>_partition_id</string> [
<string>partition1</string> [
<string>_request_dict</string> x.getNetworkInterface(''),
<none/> x.getIpAddress()
<string>_requested_state</string> ] for x in partition.contentValues(portal_type='Internet Protocol Address')
<string>started</string> ],
<string>_software_release_document</string> "full_ip_list": [],
<object id='i10' module='slapos.slap.slap' class='SoftwareRelease'> "sla_parameters": instance.getSlaXmlAsDict(),
<tuple> "compute_node_id": partition.getParentValue().getReference(),
<string>%(software_release_url)s</string> "compute_partition_id": partition.getReference(),
<string>%(compute_node_id)s</string> "processing_timestamp": instance.getSlapTimestamp(),
</tuple> "access_status_message": instance.getTextAccessStatus(),
<dictionary id='i11'> "api_revision": instance.getJIOAPIRevision(self.web_site.api.getRelativeUrl()),
<string>_computer_guid</string> "portal_type": instance.getPortalType(),
<string>%(compute_node_id)s</string> }, instance_dict)
<string>_software_instance_list</string>
<list id='i12'/> def test_11_bis_getFullComputerInformationNoSharedInstance(self):
<string>_software_release</string> self.test_11_getFullComputerInformationWithSharedInstance(with_slave=False)
<string>%(software_release_url)s</string>
</dictionary> def test_12_getSharedInstance(self):
</object> self._makeComplexComputeNode(with_slave=True)
<string>_synced</string> self.callUpdateRevisionAndTic()
<bool>1</bool> instance = self.start_requested_software_instance
</dictionary> # Check Slaves
</object> self.login(instance.getUserId())
</marshal> # XXX It should be the same portal_type
""" % dict( shared_instance_list_response = self.allDocsToApi({
compute_node_id=self.compute_node_id, "host_instance_reference": instance.getReference(),
param=self.start_requested_software_instance.getInstanceXmlAsDict()['paramé'], "portal_type": "Shared Instance",
sla=self.start_requested_software_instance.getSlaXmlAsDict()['paramé'], })
software_release_url=self.start_requested_software_instance.getUrlString(), self.maxDiff = None
timestamp=int(self.start_requested_software_instance.getModificationDate()), response = self.portal.REQUEST.RESPONSE
instance_guid=self.start_requested_software_instance.getReference(), self.assertEqual(200, response.getStatus())
instance_title=self.start_requested_software_instance.getTitle(), self.assertEqual('application/json',
root_instance_title=self.start_requested_software_instance.getSpecialiseValue().getTitle(), response.headers.get('content-type'))
software_type=self.start_requested_software_instance.getSourceReference()
) self.assertTrue(
self.assertEqual(expected_xml, got_xml, shared_instance_list_response.pop("$schema").endswith("jIOWebSection_searchInstanceFromJSON/getOutputJSONSchema")
'\n'.join([q for q in difflib.unified_diff(expected_xml.split('\n'), got_xml.split('\n'))])) )
shared_instance = self.start_requested_slave_instance
shared_instance_revision = shared_instance.getJIOAPIRevision(self.web_site.api.getRelativeUrl())
self.assertEqual(shared_instance_list_response,
{
'current_page_full': False,
'next_page_request': {'from_api_revision': shared_instance_revision,
'host_instance_reference': instance.getReference(),
'portal_type': 'Shared Instance'},
'result_list': [{'api_revision': shared_instance_revision,
'compute_partition_id': 'partition1',
'get_parameters': {'portal_type': 'Software Instance',
'reference': shared_instance.getReference()},
'portal_type': 'Software Instance',
'reference': shared_instance.getReference(),
'state': 'started',
'title': shared_instance.getTitle()}]
})
instance_dict = self.getToApi(shared_instance_list_response["result_list"][0]["get_parameters"])
response = self.portal.REQUEST.RESPONSE
self.assertEqual(200, response.getStatus())
self.assertEqual('application/json',
response.headers.get('content-type'))
# Check Data is correct
self.login()
partition = instance.getAggregateValue(portal_type="Compute Partition")
self.assertEqual({
"$schema": instance.getJSONSchemaUrl(),
"title": shared_instance.getTitle(),
"reference": shared_instance.getReference(),
"software_release_uri": shared_instance.getUrlString(),
"software_type": shared_instance.getSourceReference(),
"state": self.getAPIStateFromSlapState(shared_instance.getSlapState()),
"connection_parameters": shared_instance.getConnectionXmlAsDict(),
"parameters": shared_instance.getInstanceXmlAsDict(),
"shared": False,
"root_instance_title": shared_instance.getSpecialiseValue().getTitle(),
"ip_list": [],
"full_ip_list": [],
"sla_parameters": shared_instance.getSlaXmlAsDict(),
"compute_node_id": partition.getParentValue().getReference(),
"compute_partition_id": partition.getReference(),
"processing_timestamp": shared_instance.getSlapTimestamp(),
"access_status_message": shared_instance.getTextAccessStatus(),
"api_revision": shared_instance_revision,
"portal_type": "Slave Instance",
}, instance_dict)
def assertInstanceUpdateConnectionSimulator(self, args, kwargs): def assertInstanceUpdateConnectionSimulator(self, args, kwargs):
stored = eval(open(self.instance_update_connection_simulator).read()) #pylint: disable=eval-used stored = eval(open(self.instance_update_connection_simulator).read()) #pylint: disable=eval-used
...@@ -1090,39 +722,52 @@ class TestSlapOSSlapToolInstanceAccess(TestSlapOSJIOAPIMixin): ...@@ -1090,39 +722,52 @@ class TestSlapOSSlapToolInstanceAccess(TestSlapOSJIOAPIMixin):
[{'recargs': args, 'reckwargs': kwargs, [{'recargs': args, 'reckwargs': kwargs,
'recmethod': 'updateConnection'}]) 'recmethod': 'updateConnection'}])
def deactivated_test_setConnectionXml_withSlave(self): def test_13_setConnectionXml_withSlave(self):
# XXX CLN No idea how to deal with ascii
self._makeComplexComputeNode(with_slave=True) self._makeComplexComputeNode(with_slave=True)
partition_id = self.start_requested_software_instance.getAggregateValue( self.callUpdateRevisionAndTic()
portal_type='Compute Partition').getReference() connection_parameters_dict = {
slave_reference = self.start_requested_slave_instance.getReference() "p1e": "v1e",
connection_xml = """<marshal> "p2e": "v2e",
<dictionary id="i2"> }
<string>p1é</string>
<string>v1é</string>
<string>p2é</string>
<string>v2é</string>
</dictionary>
</marshal>"""
stored_xml = """<?xml version='1.0' encoding='utf-8'?> stored_xml = """<?xml version='1.0' encoding='utf-8'?>
<instance> <instance>
<parameter id="p1é">v1é</parameter> <parameter id="p2e">v2e</parameter>
<parameter id="p2é">v2é</parameter> <parameter id="p1e">v1e</parameter>
</instance> </instance>
""" """
self.login(self.start_requested_software_instance.getUserId()) self.called_update_connection_kw = ""
def calledupdateConnection(*args, **kw):
self.called_update_connection_kw = kw
start_date = DateTime().HTML4()
self.instance_update_connection_simulator = tempfile.mkstemp()[1]
try: try:
self.start_requested_slave_instance.updateConnection = Simulator( updateConnection = self.start_requested_slave_instance.__class__.updateConnection
self.instance_update_connection_simulator, 'updateConnection') self.start_requested_slave_instance.__class__.updateConnection = calledupdateConnection
response = self.portal_slap.setComputerPartitionConnectionXml( self.login(self.start_requested_software_instance.getUserId())
self.compute_node_id, partition_id, connection_xml, slave_reference) response_dict = self.putToApi({
self.assertEqual('None', response) "reference": self.start_requested_slave_instance.getReference(),
self.assertInstanceUpdateConnectionSimulator((), "portal_type": "Software Instance",
{'connection_xml': stored_xml}) "connection_parameters": connection_parameters_dict,
})
self.maxDiff = None
self.tic()
response = self.portal.REQUEST.RESPONSE
self.assertEqual(200, response.getStatus())
self.assertEqual('application/json',
response.headers.get('content-type'))
self.assertTrue(response_dict.pop("$schema").endswith("SoftwareInstance_updateFromJSON/getOutputJSONSchema"))
self.assertTrue(DateTime(response_dict.pop("date"))>= DateTime(start_date))
self.assertEqual(self.called_update_connection_kw, {"connection_xml": stored_xml})
self.assertEqual(response_dict, {
"reference": self.start_requested_slave_instance.getReference(),
"portal_type": "Software Instance",
"success": "Done"
})
finally: finally:
if os.path.exists(self.instance_update_connection_simulator): self.start_requested_slave_instance.__class__.updateConnection = updateConnection
os.unlink(self.instance_update_connection_simulator)
def deactivated_test_setConnectionXml(self): def deactivated_test_setConnectionXml(self):
self._makeComplexComputeNode() self._makeComplexComputeNode()
......
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