Commit 5dc02df9 authored by Romain Courteaud's avatar Romain Courteaud

Draft version of hypermedia access to slapos.

Use http://tools.ietf.org/html/draft-kelly-json-hal-05 has base format.
Provide possibility to request instance, browse them and check their message
and connection parameters.
parent a1dcdb01
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="Folder" module="OFS.Folder"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>_local_properties</string> </key>
<value>
<tuple>
<dictionary>
<item>
<key> <string>id</string> </key>
<value> <string>business_template_skin_layer_priority</string> </value>
</item>
<item>
<key> <string>type</string> </key>
<value> <string>float</string> </value>
</item>
</dictionary>
</tuple>
</value>
</item>
<item>
<key> <string>_objects</string> </key>
<value>
<tuple/>
</value>
</item>
<item>
<key> <string>business_template_skin_layer_priority</string> </key>
<value> <float>60.0</float> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>slapos_hypermedia</string> </value>
</item>
<item>
<key> <string>title</string> </key>
<value> <string></string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="PythonScript" module="Products.PythonScripts.PythonScript"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>Script_magic</string> </key>
<value> <int>3</int> </value>
</item>
<item>
<key> <string>_bind_names</string> </key>
<value>
<object>
<klass>
<global name="NameAssignments" module="Shared.DC.Scripts.Bindings"/>
</klass>
<tuple/>
<state>
<dictionary>
<item>
<key> <string>_asgns</string> </key>
<value>
<dictionary>
<item>
<key> <string>name_container</string> </key>
<value> <string>container</string> </value>
</item>
<item>
<key> <string>name_context</string> </key>
<value> <string>context</string> </value>
</item>
<item>
<key> <string>name_m_self</string> </key>
<value> <string>script</string> </value>
</item>
<item>
<key> <string>name_subpath</string> </key>
<value> <string>traverse_subpath</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</state>
</object>
</value>
</item>
<item>
<key> <string>_body</string> </key>
<value> <string>from zExceptions import Unauthorized\n
from AccessControl import getSecurityManager\n
if REQUEST is None:\n
raise Unauthorized\n
\n
response = REQUEST.RESPONSE\n
type = \'application/vnd.slapos.org.hal+json; class=slapos.org.master\'\n
\n
if REQUEST.other[\'method\'] != "GET":\n
response.setStatus(405)\n
return ""\n
elif type != context.Base_handleAcceptHeader([type]):\n
response.setStatus(406)\n
return ""\n
else:\n
import json\n
result_dict = {\n
\'_class\': \'slapos.org.master\',\n
# \'title\': \'Vifib SlapOS Master\',\n
# \'description\': \'Reference implementation of the SlapOS Master specifications.\',\n
\'_links\': {\n
"self": { "href": context.Base_getRequestUrl(),\n
"type": type },\n
# "http://slapos.org/reg/edit": {\n
# "href": "/edit",\n
# "method": "PUT",\n
# "type": "application/json; class=slapos.org.master",\n
# },\n
# "http://slapos.org/reg/create": {\n
# "href": "/new",\n
# "method": "POST",\n
# "type": "application/json; class=slapos.org.person",\n
# },\n
},\n
}\n
\n
portal = context.getPortalObject()\n
\n
# XXX Handle also other kind of users: instance, computer, master\n
person = portal.ERP5Site_getAuthenticatedMemberPersonValue()\n
if person is not None:\n
result_dict[\'_links\'][\'http://slapos.org/reg/me\'] = {\n
\'href\': \'%s/Person_getHateoas\' % person.absolute_url(),\n
"type": "application/vnd.slapos.org.hal+json; class=slapos.org.person"\n
}\n
\n
response.setHeader(\'Content-Type\', type)\n
return json.dumps(result_dict)\n
</string> </value>
</item>
<item>
<key> <string>_params</string> </key>
<value> <string>REQUEST=None</string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>Base_getHateoasMaster</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="PythonScript" module="Products.PythonScripts.PythonScript"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>Script_magic</string> </key>
<value> <int>3</int> </value>
</item>
<item>
<key> <string>_bind_names</string> </key>
<value>
<object>
<klass>
<global name="NameAssignments" module="Shared.DC.Scripts.Bindings"/>
</klass>
<tuple/>
<state>
<dictionary>
<item>
<key> <string>_asgns</string> </key>
<value>
<dictionary>
<item>
<key> <string>name_container</string> </key>
<value> <string>container</string> </value>
</item>
<item>
<key> <string>name_context</string> </key>
<value> <string>context</string> </value>
</item>
<item>
<key> <string>name_m_self</string> </key>
<value> <string>script</string> </value>
</item>
<item>
<key> <string>name_subpath</string> </key>
<value> <string>traverse_subpath</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</state>
</object>
</value>
</item>
<item>
<key> <string>_body</string> </key>
<value> <string>from zExceptions import Unauthorized\n
if REQUEST is not None:\n
raise Unauthorized\n
\n
# XXX requested to simulate in unit test for now\n
return context.REQUEST.get(\'BODY\')\n
</string> </value>
</item>
<item>
<key> <string>_params</string> </key>
<value> <string>REQUEST=None</string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>Base_getRequestBody</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="PythonScript" module="Products.PythonScripts.PythonScript"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>Script_magic</string> </key>
<value> <int>3</int> </value>
</item>
<item>
<key> <string>_bind_names</string> </key>
<value>
<object>
<klass>
<global name="NameAssignments" module="Shared.DC.Scripts.Bindings"/>
</klass>
<tuple/>
<state>
<dictionary>
<item>
<key> <string>_asgns</string> </key>
<value>
<dictionary>
<item>
<key> <string>name_container</string> </key>
<value> <string>container</string> </value>
</item>
<item>
<key> <string>name_context</string> </key>
<value> <string>context</string> </value>
</item>
<item>
<key> <string>name_m_self</string> </key>
<value> <string>script</string> </value>
</item>
<item>
<key> <string>name_subpath</string> </key>
<value> <string>traverse_subpath</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</state>
</object>
</value>
</item>
<item>
<key> <string>_body</string> </key>
<value> <string>from zExceptions import Unauthorized\n
if REQUEST is not None:\n
raise Unauthorized\n
\n
# XXX requested to simulate in unit test for now\n
return context.REQUEST.getHeader(name, default)\n
</string> </value>
</item>
<item>
<key> <string>_params</string> </key>
<value> <string>name, default=None, REQUEST=None</string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>Base_getRequestHeader</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="PythonScript" module="Products.PythonScripts.PythonScript"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>Script_magic</string> </key>
<value> <int>3</int> </value>
</item>
<item>
<key> <string>_bind_names</string> </key>
<value>
<object>
<klass>
<global name="NameAssignments" module="Shared.DC.Scripts.Bindings"/>
</klass>
<tuple/>
<state>
<dictionary>
<item>
<key> <string>_asgns</string> </key>
<value>
<dictionary>
<item>
<key> <string>name_container</string> </key>
<value> <string>container</string> </value>
</item>
<item>
<key> <string>name_context</string> </key>
<value> <string>context</string> </value>
</item>
<item>
<key> <string>name_m_self</string> </key>
<value> <string>script</string> </value>
</item>
<item>
<key> <string>name_subpath</string> </key>
<value> <string>traverse_subpath</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</state>
</object>
</value>
</item>
<item>
<key> <string>_body</string> </key>
<value> <string>from zExceptions import Unauthorized\n
if REQUEST is not None:\n
raise Unauthorized\n
\n
# XXX requested to simulate in unit test for now\n
return context.REQUEST.getURL()\n
</string> </value>
</item>
<item>
<key> <string>_params</string> </key>
<value> <string>REQUEST=None</string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>Base_getRequestUrl</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="PythonScript" module="Products.PythonScripts.PythonScript"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>Script_magic</string> </key>
<value> <int>3</int> </value>
</item>
<item>
<key> <string>_bind_names</string> </key>
<value>
<object>
<klass>
<global name="NameAssignments" module="Shared.DC.Scripts.Bindings"/>
</klass>
<tuple/>
<state>
<dictionary>
<item>
<key> <string>_asgns</string> </key>
<value>
<dictionary>
<item>
<key> <string>name_container</string> </key>
<value> <string>container</string> </value>
</item>
<item>
<key> <string>name_context</string> </key>
<value> <string>context</string> </value>
</item>
<item>
<key> <string>name_m_self</string> </key>
<value> <string>script</string> </value>
</item>
<item>
<key> <string>name_subpath</string> </key>
<value> <string>traverse_subpath</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</state>
</object>
</value>
</item>
<item>
<key> <string>_body</string> </key>
<value> <string>from zExceptions import Unauthorized\n
if REQUEST is not None:\n
raise Unauthorized\n
\n
result = None\n
accept = context.Base_getRequestHeader(\'Accept\', default=\'\')\n
for accepted_type in accepted_type_list:\n
if accepted_type in accept:\n
# XXX Really simple and stupid matching.\n
# Better test to ensure best matching type\n
result = accepted_type\n
break\n
\n
if (result is None) and (\'*/*\' in accept):\n
result = accepted_type_list[0]\n
\n
return result\n
</string> </value>
</item>
<item>
<key> <string>_params</string> </key>
<value> <string>accepted_type_list, REQUEST=None</string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>Base_handleAcceptHeader</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="PythonScript" module="Products.PythonScripts.PythonScript"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>Script_magic</string> </key>
<value> <int>3</int> </value>
</item>
<item>
<key> <string>_bind_names</string> </key>
<value>
<object>
<klass>
<global name="NameAssignments" module="Shared.DC.Scripts.Bindings"/>
</klass>
<tuple/>
<state>
<dictionary>
<item>
<key> <string>_asgns</string> </key>
<value>
<dictionary>
<item>
<key> <string>name_container</string> </key>
<value> <string>container</string> </value>
</item>
<item>
<key> <string>name_context</string> </key>
<value> <string>context</string> </value>
</item>
<item>
<key> <string>name_m_self</string> </key>
<value> <string>script</string> </value>
</item>
<item>
<key> <string>name_subpath</string> </key>
<value> <string>traverse_subpath</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</state>
</object>
</value>
</item>
<item>
<key> <string>_body</string> </key>
<value> <string>from zExceptions import Unauthorized\n
from AccessControl import getSecurityManager\n
if REQUEST is None:\n
raise Unauthorized\n
\n
response = REQUEST.RESPONSE\n
type = \'application/vnd.slapos.org.hal+json; class=slapos.org.hosting_subscription\'\n
\n
if REQUEST.other[\'method\'] != "GET":\n
response.setStatus(405)\n
return ""\n
elif type != context.Base_handleAcceptHeader([type]):\n
response.setStatus(406)\n
return ""\n
elif context.getPortalType() != "Hosting Subscription":\n
response.setStatus(403)\n
return ""\n
else:\n
\n
# XXX Set more properties on the document\n
\n
import json\n
result_dict = {\n
\'_class\': \'slapos.org.hosting_subscription\',\n
\'title\': context.getTitle(),\n
\'_links\': {\n
"self": { "href": context.Base_getRequestUrl(),\n
"type": "application/vnd.slapos.org.hal+json; class=slapos.org.hosting_subscription" },\n
\n
"http://slapos.org/reg/instance": {\n
"href": "%s/HostingSubscription_getHateoasInstanceList" % context.absolute_url(),\n
"type": "application/vnd.slapos.org.hal+json; class=slapos.org.collection",\n
},\n
\n
},\n
}\n
\n
response.setHeader(\'Content-Type\', type)\n
return json.dumps(result_dict)\n
</string> </value>
</item>
<item>
<key> <string>_params</string> </key>
<value> <string>REQUEST=None</string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>HostingSubscription_getHateoas</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="PythonScript" module="Products.PythonScripts.PythonScript"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>Script_magic</string> </key>
<value> <int>3</int> </value>
</item>
<item>
<key> <string>_bind_names</string> </key>
<value>
<object>
<klass>
<global name="NameAssignments" module="Shared.DC.Scripts.Bindings"/>
</klass>
<tuple/>
<state>
<dictionary>
<item>
<key> <string>_asgns</string> </key>
<value>
<dictionary>
<item>
<key> <string>name_container</string> </key>
<value> <string>container</string> </value>
</item>
<item>
<key> <string>name_context</string> </key>
<value> <string>context</string> </value>
</item>
<item>
<key> <string>name_m_self</string> </key>
<value> <string>script</string> </value>
</item>
<item>
<key> <string>name_subpath</string> </key>
<value> <string>traverse_subpath</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</state>
</object>
</value>
</item>
<item>
<key> <string>_body</string> </key>
<value> <string>from zExceptions import Unauthorized\n
from AccessControl import getSecurityManager\n
if REQUEST is None:\n
raise Unauthorized\n
\n
response = REQUEST.RESPONSE\n
type = \'application/vnd.slapos.org.hal+json; class=slapos.org.collection\'\n
\n
if REQUEST.other[\'method\'] != "GET":\n
response.setStatus(405)\n
return ""\n
elif type != context.Base_handleAcceptHeader([type]):\n
response.setStatus(406)\n
return ""\n
elif context.getPortalType() != "Hosting Subscription":\n
response.setStatus(403)\n
return ""\n
else:\n
import json\n
result_dict = {\n
\'_class\': \'slapos.org.collection\',\n
\'_links\': {\n
"self": { "href": context.Base_getRequestUrl(),\n
"type": type },\n
"item": [],\n
},\n
}\n
\n
for sql_obj in context.getPortalObject().portal_catalog(\n
portal_type=[\'Software Instance\', \'Slave Instance\'],\n
default_specialise_uid=context.getUid(),\n
):\n
obj = sql_obj.getObject()\n
result_dict[\'_links\'][\'item\'].append({\n
\'href\': \'%s/Instance_getHateoas\' % obj.absolute_url(),\n
\'type\': \'application/vnd.slapos.org.hal+json; class=slapos.org.instance\',\n
})\n
\n
response.setHeader(\'Content-Type\', type)\n
return json.dumps(result_dict)\n
</string> </value>
</item>
<item>
<key> <string>_params</string> </key>
<value> <string>REQUEST=None</string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>HostingSubscription_getHateoasInstanceList</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="PythonScript" module="Products.PythonScripts.PythonScript"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>Script_magic</string> </key>
<value> <int>3</int> </value>
</item>
<item>
<key> <string>_bind_names</string> </key>
<value>
<object>
<klass>
<global name="NameAssignments" module="Shared.DC.Scripts.Bindings"/>
</klass>
<tuple/>
<state>
<dictionary>
<item>
<key> <string>_asgns</string> </key>
<value>
<dictionary>
<item>
<key> <string>name_container</string> </key>
<value> <string>container</string> </value>
</item>
<item>
<key> <string>name_context</string> </key>
<value> <string>context</string> </value>
</item>
<item>
<key> <string>name_m_self</string> </key>
<value> <string>script</string> </value>
</item>
<item>
<key> <string>name_subpath</string> </key>
<value> <string>traverse_subpath</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</state>
</object>
</value>
</item>
<item>
<key> <string>_body</string> </key>
<value> <string>from zExceptions import Unauthorized\n
from AccessControl import getSecurityManager\n
if REQUEST is None:\n
raise Unauthorized\n
\n
response = REQUEST.RESPONSE\n
type = \'application/vnd.slapos.org.hal+json; class=slapos.org.instance\'\n
\n
if REQUEST.other[\'method\'] != "GET":\n
response.setStatus(405)\n
return ""\n
elif type != context.Base_handleAcceptHeader([type]):\n
response.setStatus(406)\n
return ""\n
elif context.getPortalType() not in ["Software Instance", "Slave Instance"]:\n
response.setStatus(403)\n
return ""\n
else:\n
\n
if context.getSlapState() == "stop_requested":\n
state = \'stopped\'\n
elif context.getSlapState() == "start_requested":\n
state = \'started\'\n
else:\n
state = \'destroyed\'\n
\n
import json\n
result_dict = {\n
\'_class\': \'slapos.org.instance\',\n
\'title\': context.getTitle(),\n
\'slave\': context.getPortalType() == \'Slave Instance\',\n
\'software_type\': context.getSourceReference(),\n
\'parameter\': context.getInstanceXmlAsDict(),\n
\'sla\': context.getSlaXmlAsDict(),\n
\'connection\': context.getConnectionXmlAsDict(),\n
\'status\': state,\n
\'_links\': {\n
"self": { "href": context.Base_getRequestUrl(),\n
"type": "application/vnd.slapos.org.hal+json; class=slapos.org.instance" },\n
"http://slapos.org/reg/news": { "href": "%s/Instance_getHateoasNews" % context.absolute_url(),\n
"type": "application/vnd.slapos.org.hal+json; class=slapos.org.news" },\n
"http://slapos.org/reg/release": { "href": context.getUrlString()},\n
},\n
}\n
\n
response.setHeader(\'Content-Type\', type)\n
return json.dumps(result_dict)\n
</string> </value>
</item>
<item>
<key> <string>_params</string> </key>
<value> <string>REQUEST=None</string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>Instance_getHateoas</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="PythonScript" module="Products.PythonScripts.PythonScript"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>Script_magic</string> </key>
<value> <int>3</int> </value>
</item>
<item>
<key> <string>_bind_names</string> </key>
<value>
<object>
<klass>
<global name="NameAssignments" module="Shared.DC.Scripts.Bindings"/>
</klass>
<tuple/>
<state>
<dictionary>
<item>
<key> <string>_asgns</string> </key>
<value>
<dictionary>
<item>
<key> <string>name_container</string> </key>
<value> <string>container</string> </value>
</item>
<item>
<key> <string>name_context</string> </key>
<value> <string>context</string> </value>
</item>
<item>
<key> <string>name_m_self</string> </key>
<value> <string>script</string> </value>
</item>
<item>
<key> <string>name_subpath</string> </key>
<value> <string>traverse_subpath</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</state>
</object>
</value>
</item>
<item>
<key> <string>_body</string> </key>
<value> <string>from zExceptions import Unauthorized\n
from AccessControl import getSecurityManager\n
if REQUEST is None:\n
raise Unauthorized\n
\n
response = REQUEST.RESPONSE\n
type = \'application/vnd.slapos.org.hal+json; class=slapos.org.news\'\n
\n
if REQUEST.other[\'method\'] != "GET":\n
response.setStatus(405)\n
return ""\n
elif type != context.Base_handleAcceptHeader([type]):\n
response.setStatus(406)\n
return ""\n
elif context.getPortalType() not in ["Software Instance", "Slave Instance"]:\n
response.setStatus(403)\n
return ""\n
else:\n
\n
import json\n
\n
memcached_dict = context.getPortalObject().portal_memcached.getMemcachedDict(\n
key_prefix=\'slap_tool\',\n
plugin_path=\'portal_memcached/default_memcached_plugin\')\n
try:\n
d = memcached_dict[context.getReference()]\n
except KeyError:\n
d = {\n
"user": "SlapOS Master",\n
"text": "#error no data found for %s" % context.getReference()\n
}\n
else:\n
d = json.loads(d)\n
\n
result_dict = {\n
\'_class\': \'slapos.org.news\',\n
\'news\': [d],\n
\'_links\': {\n
"self": { "href": context.Base_getRequestUrl(),\n
"type": "application/vnd.slapos.org.hal+json; class=slapos.org.news" },\n
"http://slapos.org/reg/instance": { "href": "%s/Instance_getHateoas" % context.absolute_url(),\n
"type": "application/vnd.slapos.org.hal+json; class=slapos.org.instance" },\n
},\n
}\n
\n
response.setHeader(\'Content-Type\', type)\n
return json.dumps(result_dict)\n
</string> </value>
</item>
<item>
<key> <string>_params</string> </key>
<value> <string>REQUEST=None</string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>Instance_getHateoasNews</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="PythonScript" module="Products.PythonScripts.PythonScript"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>Script_magic</string> </key>
<value> <int>3</int> </value>
</item>
<item>
<key> <string>_bind_names</string> </key>
<value>
<object>
<klass>
<global name="NameAssignments" module="Shared.DC.Scripts.Bindings"/>
</klass>
<tuple/>
<state>
<dictionary>
<item>
<key> <string>_asgns</string> </key>
<value>
<dictionary>
<item>
<key> <string>name_container</string> </key>
<value> <string>container</string> </value>
</item>
<item>
<key> <string>name_context</string> </key>
<value> <string>context</string> </value>
</item>
<item>
<key> <string>name_m_self</string> </key>
<value> <string>script</string> </value>
</item>
<item>
<key> <string>name_subpath</string> </key>
<value> <string>traverse_subpath</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</state>
</object>
</value>
</item>
<item>
<key> <string>_body</string> </key>
<value> <string>from zExceptions import Unauthorized\n
from AccessControl import getSecurityManager\n
if REQUEST is None:\n
raise Unauthorized\n
\n
response = REQUEST.RESPONSE\n
type = \'application/vnd.slapos.org.hal+json; class=slapos.org.person\'\n
\n
if REQUEST.other[\'method\'] != "GET":\n
response.setStatus(405)\n
return ""\n
elif type != context.Base_handleAcceptHeader([type]):\n
response.setStatus(406)\n
return ""\n
elif context.getPortalType() != "Person":\n
response.setStatus(403)\n
return ""\n
else:\n
import json\n
result_dict = {\n
\'_class\': \'slapos.org.person\',\n
\'title\': context.getTitle(),\n
\'_links\': {\n
"self": { "href": context.Base_getRequestUrl(),\n
"type": "application/vnd.slapos.org.hal+json; class=slapos.org.person" },\n
\n
"http://slapos.org/reg/request": {\n
"href": "%s/Person_requestHateoasHostingSubscription" % context.absolute_url(),\n
"method": "POST",\n
"type": "application/json; class=slapos.org.hosting_subscription",\n
},\n
\n
"http://slapos.org/reg/hosting_subscription": {\n
"href": "%s/Person_getHateoasHostingSubscriptionList" % context.absolute_url(),\n
"type": "application/vnd.slapos.org.hal+json; class=slapos.org.collection",\n
},\n
\n
},\n
}\n
\n
response.setHeader(\'Content-Type\', type)\n
return json.dumps(result_dict)\n
</string> </value>
</item>
<item>
<key> <string>_params</string> </key>
<value> <string>REQUEST=None</string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>Person_getHateoas</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="PythonScript" module="Products.PythonScripts.PythonScript"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>Script_magic</string> </key>
<value> <int>3</int> </value>
</item>
<item>
<key> <string>_bind_names</string> </key>
<value>
<object>
<klass>
<global name="NameAssignments" module="Shared.DC.Scripts.Bindings"/>
</klass>
<tuple/>
<state>
<dictionary>
<item>
<key> <string>_asgns</string> </key>
<value>
<dictionary>
<item>
<key> <string>name_container</string> </key>
<value> <string>container</string> </value>
</item>
<item>
<key> <string>name_context</string> </key>
<value> <string>context</string> </value>
</item>
<item>
<key> <string>name_m_self</string> </key>
<value> <string>script</string> </value>
</item>
<item>
<key> <string>name_subpath</string> </key>
<value> <string>traverse_subpath</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</state>
</object>
</value>
</item>
<item>
<key> <string>_body</string> </key>
<value> <string>from zExceptions import Unauthorized\n
from AccessControl import getSecurityManager\n
if REQUEST is None:\n
raise Unauthorized\n
\n
response = REQUEST.RESPONSE\n
type = \'application/vnd.slapos.org.hal+json; class=slapos.org.collection\'\n
\n
if REQUEST.other[\'method\'] != "GET":\n
response.setStatus(405)\n
return ""\n
elif type != context.Base_handleAcceptHeader([type]):\n
response.setStatus(406)\n
return ""\n
elif context.getPortalType() != "Person":\n
response.setStatus(403)\n
return ""\n
else:\n
import json\n
result_dict = {\n
\'_class\': \'slapos.org.collection\',\n
\'_links\': {\n
"self": { "href": context.Base_getRequestUrl(),\n
"type": type },\n
"item": [],\n
},\n
}\n
\n
for sql_obj in context.getPortalObject().portal_catalog(\n
portal_type="Hosting Subscription",\n
default_destination_section_uid=context.getUid(),\n
):\n
obj = sql_obj.getObject()\n
result_dict[\'_links\'][\'item\'].append({\n
\'href\': \'%s/HostingSubscription_getHateoas\' % obj.absolute_url(),\n
\'type\': \'application/vnd.slapos.org.hal+json; class=slapos.org.hosting_subscription\',\n
})\n
\n
response.setHeader(\'Content-Type\', type)\n
return json.dumps(result_dict)\n
</string> </value>
</item>
<item>
<key> <string>_params</string> </key>
<value> <string>REQUEST=None</string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>Person_getHateoasHostingSubscriptionList</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="PythonScript" module="Products.PythonScripts.PythonScript"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>Script_magic</string> </key>
<value> <int>3</int> </value>
</item>
<item>
<key> <string>_bind_names</string> </key>
<value>
<object>
<klass>
<global name="NameAssignments" module="Shared.DC.Scripts.Bindings"/>
</klass>
<tuple/>
<state>
<dictionary>
<item>
<key> <string>_asgns</string> </key>
<value>
<dictionary>
<item>
<key> <string>name_container</string> </key>
<value> <string>container</string> </value>
</item>
<item>
<key> <string>name_context</string> </key>
<value> <string>context</string> </value>
</item>
<item>
<key> <string>name_m_self</string> </key>
<value> <string>script</string> </value>
</item>
<item>
<key> <string>name_subpath</string> </key>
<value> <string>traverse_subpath</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</state>
</object>
</value>
</item>
<item>
<key> <string>_body</string> </key>
<value> <string encoding="cdata"><![CDATA[
from zExceptions import Unauthorized\n
from AccessControl import getSecurityManager\n
if REQUEST is None:\n
raise Unauthorized\n
\n
response = REQUEST.RESPONSE\n
type = \'application/json; class=slapos.org.hosting_subscription\'\n
\n
if REQUEST.other[\'method\'] != "POST":\n
response.setStatus(405)\n
return ""\n
elif type != context.Base_getRequestHeader(\'Content-Type\'):\n
response.setStatus(406)\n
return ""\n
elif context.getPortalType() != "Person":\n
response.setStatus(403)\n
return ""\n
else:\n
import json\n
try:\n
data_dict = json.loads(context.Base_getRequestBody())\n
except (TypeError, ValueError):\n
response.setStatus(400)\n
return ""\n
else:\n
\n
def dictToXml(dict_data):\n
assert same_type(dict_data, {})\n
result = "<?xml version=\'1.0\' encoding=\'utf-8\'?><instance>\\n"\n
for key, value in dict_data.items():\n
result += """ <parameter id="%s">%s</parameter>\\n""" % (key, value)\n
result += "</instance>"\n
return result\n
\n
try:\n
parameter_kw = {\n
\'software_release\': data_dict[\'software_release\'],\n
\'software_title\': data_dict[\'title\'],\n
\'software_type\': data_dict[\'software_type\'],\n
\'instance_xml\': dictToXml(data_dict[\'parameter\']),\n
\'sla_xml\': dictToXml(data_dict[\'sla\']),\n
\'shared\': data_dict[\'slave\'],\n
\'state\': data_dict[\'status\'],\n
}\n
except KeyError:\n
response.setStatus(400)\n
return ""\n
else:\n
\n
context.requestSoftwareInstance(**parameter_kw)\n
# XXX Return hosting subscription link\n
response.setStatus(201)\n
return ""\n
]]></string> </value>
</item>
<item>
<key> <string>_params</string> </key>
<value> <string>REQUEST=None</string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>Person_requestHateoasHostingSubscription</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
# Copyright (c) 2002-2013 Nexedi SA and Contributors. All Rights Reserved.
import transaction
from Products.SlapOS.tests.testSlapOSMixin import \
testSlapOSMixin
import json
import httplib
import urlparse
import base64
class TestSlapOSHypermediaPersonScenario(testSlapOSMixin):
def _makeUser(self):
new_id = self.generateNewId()
person_user = self.portal.person_module.template_member.\
Base_createCloneDocument(batch_mode=1)
person_user.edit(
title="live_test_%s" % new_id,
reference="live_test_%s" % new_id,
password="live_test_%s" % new_id,
default_email_text="live_test_%s@example.org" % new_id,
)
person_user.validate()
for assignment in person_user.contentValues(portal_type="Assignment"):
assignment.open()
self.tic()
return person_user
def test(self):
erp5_person = self._makeUser()
authorization = 'Basic %s' % base64.b64encode(
"%s:%s" % (erp5_person.getReference(), erp5_person.getReference()))
# XXX Default home url. 'Hardcoded' in client.
api_scheme, api_netloc, api_path, api_query, \
api_fragment = urlparse.urlsplit('%s/Base_getHateoasMaster' % \
self.portal.absolute_url())
#####################################################
# Access the master home page hal
#####################################################
content_type = "application/vnd.slapos.org.hal+json; class=slapos.org.master"
connection = httplib.HTTPConnection(api_netloc)
connection.request(
method='GET',
url='%s/Base_getHateoasMaster' % \
self.portal.absolute_url(),
headers={
'Authorization': authorization,
'Accept': content_type,
},
body="",
)
response = connection.getresponse()
self.assertEquals(response.status, 200)
self.assertEquals(response.getheader('Content-Type'), content_type)
home_page_hal = json.loads(response.read())
#####################################################
# Fetch the user hal
#####################################################
content_type = "application/vnd.slapos.org.hal+json; class=slapos.org.person"
user_link_dict = home_page_hal['_links']['http://slapos.org/reg/me']
connection = httplib.HTTPConnection(api_netloc)
connection.request(
method=user_link_dict.get('method', 'GET'),
url=user_link_dict['href'],
headers={
'Authorization': authorization,
'Accept': user_link_dict['type'],
},
body="",
)
response = connection.getresponse()
self.assertEquals(response.status, 200)
self.assertEquals(response.getheader('Content-Type'), content_type)
user_hal = json.loads(response.read())
#####################################################
# Run method to request an hosting subscription
#####################################################
content_type = "application/json; " \
"class=slapos.org.hosting_subscription"
request_link_dict = user_hal['_links']['http://slapos.org/reg/request']
connection = httplib.HTTPConnection(api_netloc)
connection.request(
method=request_link_dict.get('method', 'GET'),
url=request_link_dict['href'],
headers={
'Authorization': authorization,
'Content-Type': request_link_dict['type'],
},
body=json.dumps({
'software_release': 'http://example.org',
'title': 'a great title',
'software_type': 'foo',
'parameter': {'param1': 'value1', 'param2': 'value2'},
'sla': {'param3': 'value3', 'param4': 'value4'},
'slave': False,
'status': 'started',
}),
)
response = connection.getresponse()
self.assertEquals(response.status, 201)
self.tic()
#####################################################
# Get user's hosting subscription list
#####################################################
content_type = "application/vnd.slapos.org.hal+json; " \
"class=slapos.org.collection"
user_link_dict = user_hal['_links']\
['http://slapos.org/reg/hosting_subscription']
connection = httplib.HTTPConnection(api_netloc)
connection.request(
method=user_link_dict.get('method', 'GET'),
url=user_link_dict['href'],
headers={
'Authorization': authorization,
'Accept': user_link_dict['type'],
},
body="",
)
response = connection.getresponse()
self.assertEquals(response.status, 200)
self.assertEquals(response.getheader('Content-Type'), content_type)
subscription_collection_hal = json.loads(response.read())
#####################################################
# Get user's hosting subscription
#####################################################
content_type = "application/vnd.slapos.org.hal+json; " \
"class=slapos.org.hosting_subscription"
subscription_link_dict = subscription_collection_hal['_links']\
['item'][0]
connection = httplib.HTTPConnection(api_netloc)
connection.request(
method=subscription_link_dict.get('method', 'GET'),
url=subscription_link_dict['href'],
headers={
'Authorization': authorization,
'Accept': subscription_link_dict['type'],
},
body="",
)
response = connection.getresponse()
self.assertEquals(response.status, 200)
self.assertEquals(response.getheader('Content-Type'), content_type)
subscription_hal = json.loads(response.read())
#####################################################
# Get hosting subscription's instance list
#####################################################
content_type = "application/vnd.slapos.org.hal+json; " \
"class=slapos.org.collection"
user_link_dict = subscription_hal['_links']\
['http://slapos.org/reg/instance']
connection = httplib.HTTPConnection(api_netloc)
connection.request(
method=user_link_dict.get('method', 'GET'),
url=user_link_dict['href'],
headers={
'Authorization': authorization,
'Accept': user_link_dict['type'],
},
body="",
)
response = connection.getresponse()
self.assertEquals(response.status, 200)
self.assertEquals(response.getheader('Content-Type'), content_type)
instance_collection_hal = json.loads(response.read())
#####################################################
# Get instance
#####################################################
content_type = "application/vnd.slapos.org.hal+json; " \
"class=slapos.org.instance"
subscription_link_dict = instance_collection_hal['_links']\
['item'][0]
connection = httplib.HTTPConnection(api_netloc)
connection.request(
method=subscription_link_dict.get('method', 'GET'),
url=subscription_link_dict['href'],
headers={
'Authorization': authorization,
'Accept': subscription_link_dict['type'],
},
body="",
)
response = connection.getresponse()
self.assertEquals(response.status, 200)
self.assertEquals(response.getheader('Content-Type'), content_type)
instance_hal = json.loads(response.read())
#####################################################
# Get instance news
#####################################################
content_type = "application/vnd.slapos.org.hal+json; " \
"class=slapos.org.news"
news_link_dict = instance_hal['_links']\
['http://slapos.org/reg/news']
connection = httplib.HTTPConnection(api_netloc)
connection.request(
method=news_link_dict.get('method', 'GET'),
url=news_link_dict['href'],
headers={
'Authorization': authorization,
'Accept': news_link_dict['type'],
},
body="",
)
response = connection.getresponse()
self.assertEquals(response.status, 200)
self.assertEquals(response.getheader('Content-Type'), content_type)
news_hal = json.loads(response.read())
# Copyright (c) 2002-2013 Nexedi SA and Contributors. All Rights Reserved.
import transaction
from Products.SlapOS.tests.testSlapOSMixin import \
testSlapOSMixin
from zExceptions import Unauthorized
from Products.ERP5Type.tests.utils import createZODBPythonScript
from Products.ERP5Type.tests.backportUnittest import skip
from functools import wraps
from ZPublisher.HTTPRequest import HTTPRequest
from ZPublisher.HTTPResponse import HTTPResponse
import os
import sys
import json
import StringIO
def simulate(script_id, params_string, code_string):
def upperWrap(f):
@wraps(f)
def decorated(self, *args, **kw):
if script_id in self.portal.portal_skins.custom.objectIds():
raise ValueError('Precondition failed: %s exists in custom' % script_id)
createZODBPythonScript(self.portal.portal_skins.custom,
script_id, params_string, code_string)
try:
result = f(self, *args, **kw)
finally:
if script_id in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_id)
transaction.commit()
return result
return decorated
return upperWrap
def do_fake_request(request_method, headers={}):
__version__ = "0.1"
env={}
env['SERVER_NAME']='bobo.server'
env['SERVER_PORT']='80'
env['REQUEST_METHOD']=request_method
env['REMOTE_ADDR']='204.183.226.81 '
env['REMOTE_HOST']='bobo.remote.host'
env['HTTP_USER_AGENT']='Bobo/%s' % __version__
env['HTTP_HOST']='127.0.0.1'
env['SERVER_SOFTWARE']='Bobo/%s' % __version__
env['SERVER_PROTOCOL']='HTTP/1.0 '
env['HTTP_ACCEPT']='image/gif, image/x-xbitmap, image/jpeg, */* '
env['SERVER_HOSTNAME']='bobo.server.host'
env['GATEWAY_INTERFACE']='CGI/1.1 '
env['SCRIPT_NAME']='Main'
env.update(headers)
return HTTPRequest(StringIO.StringIO(), env, HTTPResponse())
class TestSlapOSBase_getRequestHeader(testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def test_getRequestHeader_REQUEST_disallowed(self):
self.assertRaises(
Unauthorized,
self.portal.Base_getRequestHeader,
"foo",
REQUEST={})
def test_getRequestHeader_key_error(self):
self.assertEquals(
self.portal.Base_getRequestHeader('foo'),
None
)
def test_getRequestHeader_default_value(self):
self.assertEquals(
self.portal.Base_getRequestHeader('foo', default='bar'),
'bar'
)
@skip('TODO')
def test_getRequestHeader_matching_key(self):
pass
class TestSlapOSBase_getRequestUrl(testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def test_getRequestUrl_REQUEST_disallowed(self):
self.assertRaises(
Unauthorized,
self.portal.Base_getRequestUrl,
REQUEST={})
@skip('TODO')
def test_getRequestUrl_matching_key(self):
pass
class TestSlapOSBase_getRequestBody(testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def test_getRequestBody_REQUEST_disallowed(self):
self.assertRaises(
Unauthorized,
self.portal.Base_getRequestBody,
REQUEST={})
@skip('TODO')
def test_getRequestBody_matching_key(self):
pass
class TestSlapOSBase_handleAcceptHeader(testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def test_handleAcceptHeader_REQUEST_disallowed(self):
self.assertRaises(
Unauthorized,
self.portal.Base_handleAcceptHeader,
[],
REQUEST={})
@simulate('Base_getRequestHeader', '*args, **kwargs', 'return "*/*"')
def test_handleAcceptHeader_star_accept(self):
self.assertEquals(
self.portal.Base_handleAcceptHeader(['application/vnd+test',
'application/vnd+test2']),
'application/vnd+test'
)
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd+2test"')
def test_handleAcceptHeader_matching_type(self):
self.assertEquals(
self.portal.Base_handleAcceptHeader(['application/vnd+test',
'application/vnd+2test']),
'application/vnd+2test'
)
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd+2test"')
def test_handleAcceptHeader_non_matching_type(self):
self.assertEquals(
self.portal.Base_handleAcceptHeader(['application/vnd+test']),
None
)
class TestSlapOSBase_getHateoasMaster(testSlapOSMixin):
def _makePerson(self):
new_id = self.generateNewId()
person_user = self.portal.person_module.template_member.\
Base_createCloneDocument(batch_mode=1)
person_user.edit(
title="live_test_%s" % new_id,
reference="live_test_%s" % new_id,
default_email_text="live_test_%s@example.org" % new_id,
)
person_user.validate()
for assignment in person_user.contentValues(portal_type="Assignment"):
assignment.open()
self.tic()
return person_user
def beforeTearDown(self):
transaction.abort()
def test_getHateoasMaster_REQUEST_mandatory(self):
self.assertRaises(
Unauthorized,
self.portal.Base_getHateoasMaster
)
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd+bar"')
def test_getHateoasMaster_wrong_ACCEPT(self):
fake_request = do_fake_request("GET")
result = self.portal.Base_getHateoasMaster(REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 406)
self.assertEquals(result, "")
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd.slapos.org.hal+json; ' \
'class=slapos.org.master"')
def test_getHateoasMaster_bad_method(self):
fake_request = do_fake_request("POST")
result = self.portal.Base_getHateoasMaster(REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 405)
self.assertEquals(result, "")
@simulate('Base_getRequestUrl', '*args, **kwargs',
'return "http://example.org/bar"')
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd.slapos.org.hal+json; ' \
'class=slapos.org.master"')
def test_getHateoasMaster_anonymous_result(self):
self.logout()
fake_request = do_fake_request("GET")
result = self.portal.Base_getHateoasMaster(REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 200)
self.assertEquals(fake_request.RESPONSE.getHeader('Content-Type'),
"application/vnd.slapos.org.hal+json; class=slapos.org.master"
)
self.assertEquals(result, json.dumps({
'_class': 'slapos.org.master',
'_links': {
"self": {
"href": "http://example.org/bar",
"type": "application/vnd.slapos.org.hal+json; class=slapos.org.master"
},
},
}))
@simulate('Base_getRequestUrl', '*args, **kwargs',
'return "http://example.org/bar"')
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd.slapos.org.hal+json; ' \
'class=slapos.org.master"')
def test_getHateoasMaster_person_result(self):
person_user = self._makePerson()
self.login(person_user.getReference())
fake_request = do_fake_request("GET")
result = self.portal.Base_getHateoasMaster(REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 200)
self.assertEquals(fake_request.RESPONSE.getHeader('Content-Type'),
"application/vnd.slapos.org.hal+json; class=slapos.org.master"
)
self.assertEquals(result, json.dumps({
'_class': 'slapos.org.master',
'_links': {
"self": {
"href": "http://example.org/bar",
"type": "application/vnd.slapos.org.hal+json; class=slapos.org.master"
},
"http://slapos.org/reg/me": {
"href": "%s/Person_getHateoas" % person_user.absolute_url(),
"type": "application/vnd.slapos.org.hal+json; class=slapos.org.person"
},
},
}))
class TestSlapOSPerson_getHateoas(testSlapOSMixin):
def _makePerson(self):
new_id = self.generateNewId()
person_user = self.portal.person_module.template_member.\
Base_createCloneDocument(batch_mode=1)
person_user.edit(
title="live_test_%s" % new_id,
reference="live_test_%s" % new_id,
default_email_text="live_test_%s@example.org" % new_id,
)
person_user.validate()
for assignment in person_user.contentValues(portal_type="Assignment"):
assignment.open()
self.tic()
return person_user
def beforeTearDown(self):
transaction.abort()
def test_getHateoasPerson_REQUEST_mandatory(self):
self.assertRaises(
Unauthorized,
self.portal.Person_getHateoas
)
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd+bar"')
def test_getHateoasPerson_wrong_ACCEPT(self):
person_user = self._makePerson()
fake_request = do_fake_request("GET")
result = person_user.Person_getHateoas(REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 406)
self.assertEquals(result, "")
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd.slapos.org.hal+json; ' \
'class=slapos.org.person"')
def test_getHateoasPerson_bad_method(self):
person_user = self._makePerson()
fake_request = do_fake_request("POST")
result = person_user.Person_getHateoas(REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 405)
self.assertEquals(result, "")
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd.slapos.org.hal+json; ' \
'class=slapos.org.person"')
def test_getHateoasPerson_not_person_context(self):
fake_request = do_fake_request("GET")
result = self.portal.Person_getHateoas(REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 403)
self.assertEquals(result, "")
@simulate('Base_getRequestUrl', '*args, **kwargs',
'return "http://example.org/bar"')
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd.slapos.org.hal+json; ' \
'class=slapos.org.person"')
def test_getHateoasPerson_result(self):
person_user = self._makePerson()
fake_request = do_fake_request("GET")
result = person_user.Person_getHateoas(REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 200)
self.assertEquals(fake_request.RESPONSE.getHeader('Content-Type'),
"application/vnd.slapos.org.hal+json; class=slapos.org.person"
)
self.assertEquals(result, json.dumps({
'_class': 'slapos.org.person',
'title': person_user.getTitle(),
'_links': {
"self": {
"href": "http://example.org/bar",
"type": "application/vnd.slapos.org.hal+json; class=slapos.org.person"
},
"http://slapos.org/reg/request": {
"href": "%s/Person_requestHateoasHostingSubscription" % \
person_user.absolute_url(),
"method": "POST",
"type": "application/json; class=slapos.org.hosting_subscription",
},
"http://slapos.org/reg/hosting_subscription": {
"href": "%s/Person_getHateoasHostingSubscriptionList" % \
person_user.absolute_url(),
"type": "application/vnd.slapos.org.hal+json; " \
"class=slapos.org.collection"
},
},
}))
class TestSlapOSPerson_requestHateoasHostingSubscription(testSlapOSMixin):
def _makePerson(self):
new_id = self.generateNewId()
person_user = self.portal.person_module.template_member.\
Base_createCloneDocument(batch_mode=1)
person_user.edit(
title="live_test_%s" % new_id,
reference="live_test_%s" % new_id,
default_email_text="live_test_%s@example.org" % new_id,
)
person_user.validate()
for assignment in person_user.contentValues(portal_type="Assignment"):
assignment.open()
self.tic()
return person_user
def beforeTearDown(self):
transaction.abort()
def test_requestHateoasHostingSubscription_REQUEST_mandatory(self):
self.assertRaises(
Unauthorized,
self.portal.Person_requestHateoasHostingSubscription
)
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd+bar"')
def test_requestHateoasHostingSubscription_wrong_CONTENT(self):
person_user = self._makePerson()
fake_request = do_fake_request("POST")
result = person_user.Person_requestHateoasHostingSubscription(
REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 406)
self.assertEquals(result, "")
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/json; ' \
'class=slapos.org.hosting_subscription"')
def test_requestHateoasHostingSubscription_bad_method(self):
person_user = self._makePerson()
fake_request = do_fake_request("GET")
result = person_user.Person_requestHateoasHostingSubscription(
REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 405)
self.assertEquals(result, "")
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/json; ' \
'class=slapos.org.hosting_subscription"')
def test_requestHateoasHostingSubscription_not_person_context(self):
fake_request = do_fake_request("POST")
result = self.portal.Person_requestHateoasHostingSubscription(
REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 403)
self.assertEquals(result, "")
@simulate('Base_getRequestBody', '*args, **kwargs',
'return "[}"')
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/json; ' \
'class=slapos.org.hosting_subscription"')
def test_requestHateoasHostingSubscription_no_json(self):
person_user = self._makePerson()
fake_request = do_fake_request("POST")
result = person_user.Person_requestHateoasHostingSubscription(
REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 400)
self.assertEquals(result, "")
@simulate('Base_getRequestBody', '*args, **kwargs',
'return "%s"' % json.dumps({
}))
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/json; ' \
'class=slapos.org.hosting_subscription"')
def test_requestHateoasHostingSubscription_missing_parameter(self):
person_user = self._makePerson()
fake_request = do_fake_request("POST")
result = person_user.Person_requestHateoasHostingSubscription(
REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 400)
self.assertEquals(result, "")
@simulate('Base_getRequestBody', '*args, **kwargs',
'return """%s"""' % json.dumps({
'software_release': 'http://example.org',
'title': 'a great title',
'software_type': 'foo',
'parameter': {'param1': 'value1', 'param2': 'value2'},
'sla': {'param3': 'value3', 'param4': 'value4'},
'slave': False,
'status': 'started',
}))
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/json; ' \
'class=slapos.org.hosting_subscription"')
def test_requestHateoasHostingSubscription_result(self):
person_user = self._makePerson()
fake_request = do_fake_request("POST")
result = person_user.Person_requestHateoasHostingSubscription(
REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 201)
self.assertEquals(result, "")
# XXX Test that person.request is called.
class TestSlapOSPerson_getHateoasHostingSubscriptionList(testSlapOSMixin):
def _makePerson(self):
new_id = self.generateNewId()
person_user = self.portal.person_module.template_member.\
Base_createCloneDocument(batch_mode=1)
person_user.edit(
title="live_test_%s" % new_id,
reference="live_test_%s" % new_id,
default_email_text="live_test_%s@example.org" % new_id,
)
person_user.validate()
for assignment in person_user.contentValues(portal_type="Assignment"):
assignment.open()
self.tic()
return person_user
def _makeHostingSubscription(self):
hosting_subscription = self.portal.hosting_subscription_module\
.template_hosting_subscription.Base_createCloneDocument(batch_mode=1)
hosting_subscription.validate()
# hosting_subscription.edit(
# title=self.generateNewSoftwareTitle(),
# reference="TESTHS-%s" % self.generateNewId(),
# )
# self.request_kw = dict(
# software_release=\
# self.generateNewSoftwareReleaseUrl(),
# software_type=self.generateNewSoftwareType(),
# instance_xml=self.generateSafeXml(),
# sla_xml=self.generateSafeXml(),
# shared=False,
# software_title=hosting_subscription.getTitle(),
# state='started'
# )
# hosting_subscription.requestStart(**self.request_kw)
# hosting_subscription.requestInstance(**self.request_kw)
self.tic()
return hosting_subscription
def beforeTearDown(self):
transaction.abort()
def test_getHateoasHostingSubscriptionList_REQUEST_mandatory(self):
self.assertRaises(
Unauthorized,
self.portal.Person_getHateoasHostingSubscriptionList
)
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd+bar"')
def test_getHateoasHostingSubscriptionList_wrong_ACCEPT(self):
person_user = self._makePerson()
fake_request = do_fake_request("GET")
result = person_user.Person_getHateoasHostingSubscriptionList(
REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 406)
self.assertEquals(result, "")
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd.slapos.org.hal+json; ' \
'class=slapos.org.collection"')
def test_getHateoasHostingSubscriptionList_bad_method(self):
fake_request = do_fake_request("POST")
result = self.portal.Person_getHateoasHostingSubscriptionList(
REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 405)
self.assertEquals(result, "")
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd.slapos.org.hal+json; ' \
'class=slapos.org.collection"')
def test_requestHateoasHostingSubscription_not_person_context(self):
fake_request = do_fake_request("GET")
result = self.portal.Person_getHateoasHostingSubscriptionList(
REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 403)
self.assertEquals(result, "")
@simulate('Base_getRequestUrl', '*args, **kwargs',
'return "http://example.org/foo"')
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd.slapos.org.hal+json; ' \
'class=slapos.org.collection"')
def test_getHateoasHostingSubscriptionList_person_result(self):
person_user = self._makePerson()
hosting_subscription = self._makeHostingSubscription()
hosting_subscription.edit(destination_section_value=person_user)
self.tic()
self.login(person_user.getReference())
fake_request = do_fake_request("GET")
result = person_user.Person_getHateoasHostingSubscriptionList(
REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 200)
self.assertEquals(fake_request.RESPONSE.getHeader('Content-Type'),
"application/vnd.slapos.org.hal+json; class=slapos.org.collection"
)
self.assertEquals(result, json.dumps({
'_class': 'slapos.org.collection',
'_links': {
"self": {
"href": "http://example.org/foo",
"type": "application/vnd.slapos.org.hal+json; " \
"class=slapos.org.collection"
},
"item": [{
"href": "%s/HostingSubscription_getHateoas" % \
hosting_subscription.absolute_url(),
"type": "application/vnd.slapos.org.hal+json; " \
"class=slapos.org.hosting_subscription"
}],
},
}))
class TestSlapOSHostingSubscription_getHateoas(testSlapOSMixin):
def _makeHostingSubscription(self):
hosting_subscription = self.portal.hosting_subscription_module\
.template_hosting_subscription.Base_createCloneDocument(batch_mode=1)
hosting_subscription.edit(
title=self.generateNewSoftwareTitle(),
reference="TESTHS-%s" % self.generateNewId(),
)
self.tic()
return hosting_subscription
def beforeTearDown(self):
transaction.abort()
def test_getHateoasHostingSubscription_REQUEST_mandatory(self):
self.assertRaises(
Unauthorized,
self.portal.HostingSubscription_getHateoas
)
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd+bar"')
def test_getHateoasHostingSubscription_wrong_ACCEPT(self):
hosting_subscription = self._makeHostingSubscription()
fake_request = do_fake_request("GET")
result = hosting_subscription.HostingSubscription_getHateoas(
REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 406)
self.assertEquals(result, "")
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd.slapos.org.hal+json; ' \
'class=slapos.org.hosting_subscription"')
def test_getHateoasHostingSubscription_bad_method(self):
hosting_subscription = self._makeHostingSubscription()
fake_request = do_fake_request("POST")
result = hosting_subscription.HostingSubscription_getHateoas(
REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 405)
self.assertEquals(result, "")
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd.slapos.org.hal+json; ' \
'class=slapos.org.hosting_subscription"')
def test_getHateoasHostingSubscription_not_hosting_subscription_context(self):
fake_request = do_fake_request("GET")
result = self.portal.HostingSubscription_getHateoas(REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 403)
self.assertEquals(result, "")
@simulate('Base_getRequestUrl', '*args, **kwargs',
'return "http://example.org/bar"')
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd.slapos.org.hal+json; ' \
'class=slapos.org.hosting_subscription"')
def test_getHateoasHostingSubscription_result(self):
hosting_subscription = self._makeHostingSubscription()
fake_request = do_fake_request("GET")
result = hosting_subscription.HostingSubscription_getHateoas(
REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 200)
self.assertEquals(fake_request.RESPONSE.getHeader('Content-Type'),
"application/vnd.slapos.org.hal+json; class=slapos.org.hosting_subscription"
)
self.assertEquals(result, json.dumps({
'_class': 'slapos.org.hosting_subscription',
'title': hosting_subscription.getTitle(),
'_links': {
"self": {
"href": "http://example.org/bar",
"type": "application/vnd.slapos.org.hal+json; " \
"class=slapos.org.hosting_subscription"
},
"http://slapos.org/reg/instance": {
"href": "%s/HostingSubscription_getHateoasInstanceList" % \
hosting_subscription.absolute_url(),
"type": "application/vnd.slapos.org.hal+json; " \
"class=slapos.org.collection"
},
},
}))
class TestSlapOSHostingSubscription_getHateoasInstanceList(testSlapOSMixin):
def _makeHostingSubscription(self):
hosting_subscription = self.portal.hosting_subscription_module\
.template_hosting_subscription.Base_createCloneDocument(batch_mode=1)
hosting_subscription.validate()
self.tic()
return hosting_subscription
def _makeInstance(self):
instance = self.portal.software_instance_module\
.template_software_instance.Base_createCloneDocument(batch_mode=1)
instance.validate()
self.tic()
return instance
def beforeTearDown(self):
transaction.abort()
def test_getHateoasInstanceList_REQUEST_mandatory(self):
self.assertRaises(
Unauthorized,
self.portal.HostingSubscription_getHateoasInstanceList
)
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd+bar"')
def test_getHateoasInstanceList_wrong_ACCEPT(self):
subscription = self._makeHostingSubscription()
fake_request = do_fake_request("GET")
result = subscription.HostingSubscription_getHateoasInstanceList(
REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 406)
self.assertEquals(result, "")
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd.slapos.org.hal+json; ' \
'class=slapos.org.collection"')
def test_getHateoasInstanceList_bad_method(self):
fake_request = do_fake_request("POST")
result = self.portal.HostingSubscription_getHateoasInstanceList(
REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 405)
self.assertEquals(result, "")
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd.slapos.org.hal+json; ' \
'class=slapos.org.collection"')
def test_requestHateoasHostingSubscription_not_person_context(self):
fake_request = do_fake_request("GET")
result = self.portal.HostingSubscription_getHateoasInstanceList(
REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 403)
self.assertEquals(result, "")
@simulate('Base_getRequestUrl', '*args, **kwargs',
'return "http://example.org/bar"')
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd.slapos.org.hal+json; ' \
'class=slapos.org.collection"')
def test_getHateoasInstanceList_person_result(self):
subscription = self._makeHostingSubscription()
instance= self._makeInstance()
instance.edit(specialise_value=subscription)
self.tic()
fake_request = do_fake_request("GET")
result = subscription.HostingSubscription_getHateoasInstanceList(
REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 200)
self.assertEquals(fake_request.RESPONSE.getHeader('Content-Type'),
"application/vnd.slapos.org.hal+json; class=slapos.org.collection"
)
self.assertEquals(result, json.dumps({
'_class': 'slapos.org.collection',
'_links': {
"self": {
"href": "http://example.org/bar",
"type": "application/vnd.slapos.org.hal+json; " \
"class=slapos.org.collection"
},
"item": [{
"href": "%s/Instance_getHateoas" % \
instance.absolute_url(),
"type": "application/vnd.slapos.org.hal+json; " \
"class=slapos.org.instance"
}],
},
}))
class TestSlapOSInstance_getHateoas(testSlapOSMixin):
def _makeInstance(self):
instance = self.portal.software_instance_module\
.template_software_instance.Base_createCloneDocument(batch_mode=1)
instance.edit(
title=self.generateNewSoftwareTitle(),
reference="TESTHS-%s" % self.generateNewId(),
software_type=self.generateNewSoftwareType(),
url_string=self.generateNewSoftwareReleaseUrl(),
instance_xml=self.generateSafeXml(),
sla_xml=self.generateSafeXml(),
connection_xml=self.generateSafeXml(),
)
self.tic()
return instance
def beforeTearDown(self):
transaction.abort()
def test_getHateoasInstance_REQUEST_mandatory(self):
self.assertRaises(
Unauthorized,
self.portal.Instance_getHateoas
)
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd+bar"')
def test_getHateoasInstance_wrong_ACCEPT(self):
instance = self._makeInstance()
fake_request = do_fake_request("GET")
result = instance.Instance_getHateoas(
REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 406)
self.assertEquals(result, "")
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd.slapos.org.hal+json; ' \
'class=slapos.org.instance"')
def test_getHateoasInstance_bad_method(self):
instance = self._makeInstance()
fake_request = do_fake_request("POST")
result = instance.Instance_getHateoas(
REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 405)
self.assertEquals(result, "")
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd.slapos.org.hal+json; ' \
'class=slapos.org.instance"')
def test_getHateoasInstance_not_instance_context(self):
fake_request = do_fake_request("GET")
result = self.portal.Instance_getHateoas(REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 403)
self.assertEquals(result, "")
@simulate('Base_getRequestUrl', '*args, **kwargs',
'return "http://example.org/bar"')
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd.slapos.org.hal+json; ' \
'class=slapos.org.instance"')
def test_getHateoasInstance_result(self):
instance = self._makeInstance()
fake_request = do_fake_request("GET")
result = instance.Instance_getHateoas(
REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 200)
self.assertEquals(fake_request.RESPONSE.getHeader('Content-Type'),
"application/vnd.slapos.org.hal+json; class=slapos.org.instance"
)
self.assertEquals(json.loads(result), json.loads(json.dumps({
'_class': 'slapos.org.instance',
'title': instance.getTitle(),
'slave': False,
'software_type': instance.getSourceReference(),
'parameter': instance.getInstanceXmlAsDict(),
'sla': instance.getSlaXmlAsDict(),
'connection': instance.getConnectionXmlAsDict(),
'status': 'destroyed',
'_links': {
"self": {
"href": "http://example.org/bar",
"type": "application/vnd.slapos.org.hal+json; " \
"class=slapos.org.instance"
},
"http://slapos.org/reg/news": {
"href": "%s/Instance_getHateoasNews" % instance.absolute_url(),
"type": "application/vnd.slapos.org.hal+json; " \
"class=slapos.org.news"
},
"http://slapos.org/reg/release": {
"href": instance.getUrlString(),
},
},
})))
class TestSlapOSInstance_getHateoasNews(testSlapOSMixin):
def _makeInstance(self):
instance = self.portal.software_instance_module\
.template_software_instance.Base_createCloneDocument(batch_mode=1)
instance.edit(
title=self.generateNewSoftwareTitle(),
reference="TESTHS-%s" % self.generateNewId(),
software_type=self.generateNewSoftwareType(),
url_string=self.generateNewSoftwareReleaseUrl(),
instance_xml=self.generateSafeXml(),
sla_xml=self.generateSafeXml(),
connection_xml=self.generateSafeXml(),
)
self.tic()
return instance
def beforeTearDown(self):
transaction.abort()
def test_getHateoasNewsInstance_REQUEST_mandatory(self):
self.assertRaises(
Unauthorized,
self.portal.Instance_getHateoasNews
)
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd+bar"')
def test_getHateoasNewsInstance_wrong_ACCEPT(self):
instance = self._makeInstance()
fake_request = do_fake_request("GET")
result = instance.Instance_getHateoasNews(
REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 406)
self.assertEquals(result, "")
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd.slapos.org.hal+json; ' \
'class=slapos.org.news"')
def test_getHateoasNewsInstance_bad_method(self):
instance = self._makeInstance()
fake_request = do_fake_request("POST")
result = instance.Instance_getHateoasNews(
REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 405)
self.assertEquals(result, "")
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd.slapos.org.hal+json; ' \
'class=slapos.org.news"')
def test_getHateoasNewsInstance_not_instance_context(self):
fake_request = do_fake_request("GET")
result = self.portal.Instance_getHateoasNews(REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 403)
self.assertEquals(result, "")
@simulate('Base_getRequestUrl', '*args, **kwargs',
'return "http://example.org/bar"')
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/vnd.slapos.org.hal+json; ' \
'class=slapos.org.news"')
def test_getHateoasNewsInstance_result(self):
instance = self._makeInstance()
fake_request = do_fake_request("GET")
result = instance.Instance_getHateoasNews(
REQUEST=fake_request)
self.assertEquals(fake_request.RESPONSE.status, 200)
self.assertEquals(fake_request.RESPONSE.getHeader('Content-Type'),
"application/vnd.slapos.org.hal+json; class=slapos.org.news"
)
self.assertEquals(json.loads(result), json.loads(json.dumps({
'_class': 'slapos.org.news',
'news': [{
"user": "SlapOS Master",
"text": "#error no data found for %s" % instance.getReference()
}],
'_links': {
"self": {
"href": "http://example.org/bar",
"type": "application/vnd.slapos.org.hal+json; " \
"class=slapos.org.news"
},
"http://slapos.org/reg/instance": {
"href": "%s/Instance_getHateoas" % instance.absolute_url(),
"type": "application/vnd.slapos.org.hal+json; " \
"class=slapos.org.instance"
},
},
})))
2013 Nexedi S.A.
\ No newline at end of file
slapos_cloud
slapos_slap_tool
\ No newline at end of file
Provide hypermedia access to slapos
\ No newline at end of file
GPL
\ No newline at end of file
1
\ No newline at end of file
slapos_hypermedia
\ No newline at end of file
testSlapOSHypermediaScenario
testSlapOSHypermediaSkins
\ No newline at end of file
slapos_hypermedia
\ No newline at end of file
0.1
\ No newline at end of file
......@@ -238,6 +238,7 @@ class testSlapOSMixin(ERP5TypeTestCase):
'slapos_category',
'slapos_rest_api_tool_portal_type',
'slapos_rest_api',
'slapos_hypermedia',
'slapos_pdm',
'slapos_crm',
'slapos_accounting',
......
......@@ -10,6 +10,7 @@ slapos_bt_list = [
'slapos_pdm',
'slapos_rest_api',
'slapos_slap_tool',
'slapos_hypermedia',
'slapos_web',
'slapos_payzen',
]
......
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