[slapos_hypermedia] Add support for other types of users, like Computer Partition.

parent c45943c4
##############################################################################
#
# Copyright (c) 2002-2014 Nexedi SA and Contributors. All Rights Reserved.
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# guarantees and support are strongly adviced to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
#
##############################################################################
from Products.SlapOS.SlapOSMachineAuthenticationPlugin import getUserByLogin
def getUserDocument(self, username):
portal = self.getPortalObject()
user_document = getUserByLogin(portal, str(username))
if len(user_document) != 1:
return
return user_document[0]
...@@ -52,6 +52,7 @@ ...@@ -52,6 +52,7 @@
<key> <string>_body</string> </key> <key> <string>_body</string> </key>
<value> <string>from zExceptions import Unauthorized\n <value> <string>from zExceptions import Unauthorized\n
from AccessControl import getSecurityManager\n from AccessControl import getSecurityManager\n
\n
if REQUEST is None:\n if REQUEST is None:\n
raise Unauthorized\n raise Unauthorized\n
\n \n
...@@ -86,13 +87,20 @@ result_dict = {\n ...@@ -86,13 +87,20 @@ result_dict = {\n
\n \n
portal = context.getPortalObject()\n portal = context.getPortalObject()\n
\n \n
# XXX Handle also other kind of users: instance, computer, master\n
person = portal.ERP5Site_getAuthenticatedMemberPersonValue()\n person = portal.ERP5Site_getAuthenticatedMemberPersonValue()\n
if person is not None:\n if person is not None:\n
result_dict[\'_links\'][\'action_object_jump\'] = {\n result_dict[\'_links\'][\'action_object_jump\'] = {\n
\'href\': \'%s/ERP5Document_getHateoas\' % person.absolute_url(),\n \'href\': \'%s/ERP5Document_getHateoas\' % person.absolute_url(),\n
\'title\': \'Person\'\n \'title\': \'Person\'\n
}\n }\n
else:\n
user = str(portal.portal_membership.getAuthenticatedMember())\n
if user != "Anonymous User":\n
user_document = context.ERP5Site_getUserDocument(user)\n
result_dict[\'_links\'][\'action_object_jump\'] = {\n
\'href\': \'%s/ERP5Document_getHateoas\' % user_document.absolute_url(),\n
\'title\': user_document.getPortalType()\n
}\n
\n \n
response.setHeader(\'Content-Type\', mime_type)\n response.setHeader(\'Content-Type\', mime_type)\n
return json.dumps(result_dict, indent=2)\n return json.dumps(result_dict, indent=2)\n
......
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="ExternalMethod" module="Products.ExternalMethod.ExternalMethod"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>_function</string> </key>
<value> <string>getUserDocument</string> </value>
</item>
<item>
<key> <string>_module</string> </key>
<value> <string>SlapOSHypermedia</string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>ERP5Site_getUserDocument</string> </value>
</item>
<item>
<key> <string>title</string> </key>
<value> <string></string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
...@@ -49,8 +49,7 @@ class TestSlapOSHypermediaPersonScenario(testSlapOSMixin): ...@@ -49,8 +49,7 @@ class TestSlapOSHypermediaPersonScenario(testSlapOSMixin):
def getNewHttpConnection(api_netloc): def getNewHttpConnection(api_netloc):
if api_scheme == 'https': if api_scheme == 'https':
connection = httplib.HTTPSConnection(api_netloc) raise Exception('Please connect directly to the Zope server')
else:
connection = httplib.HTTPConnection(api_netloc) connection = httplib.HTTPConnection(api_netloc)
return connection return connection
...@@ -220,6 +219,32 @@ class TestSlapOSHypermediaPersonScenario(testSlapOSMixin): ...@@ -220,6 +219,32 @@ class TestSlapOSHypermediaPersonScenario(testSlapOSMixin):
self.assertEquals(response.getheader('Content-Type'), content_type) self.assertEquals(response.getheader('Content-Type'), content_type)
instance_hal = json.loads(response.read()) instance_hal = json.loads(response.read())
#####################################################
# Fetch instance informations
#####################################################
request_link_dict = hateoasGetLinkFromLinks(
instance_hal['_links']['action_object_slap'],
'getHateoasInformation'
)
self.assertNotEqual(request_link_dict, None)
connection = getNewHttpConnection(api_netloc)
connection.request(
method=request_link_dict.get('method', 'GET'),
url=request_link_dict['href'],
headers={
'Authorization': authorization,
'Accept': content_type,
},
body="",
)
response = connection.getresponse()
self.assertEquals(response.status, 200)
self.assertEquals(response.getheader('Content-Type'), content_type)
self.tic()
##################################################### #####################################################
# Get instance news # Get instance news
##################################################### #####################################################
...@@ -348,3 +373,224 @@ class TestSlapOSHypermediaPersonScenario(testSlapOSMixin): ...@@ -348,3 +373,224 @@ class TestSlapOSHypermediaPersonScenario(testSlapOSMixin):
response = connection.getresponse() response = connection.getresponse()
self.assertEquals(response.getheader('Content-Type'), content_type) self.assertEquals(response.getheader('Content-Type'), content_type)
software_hal = json.loads(response.read()) software_hal = json.loads(response.read())
class TestSlapOSHypermediaInstanceScenario(testSlapOSMixin):
def generateNewId(self):
return "%s" % self.portal.portal_ids.generateNewId(
id_group=('slapos_core_test'))
def generateNewSoftwareReleaseUrl(self):
return 'http://example.org/test%s.cfg' % self.generateNewId()
def generateNewSoftwareType(self):
return 'Type %s' % self.generateNewId()
def generateNewSoftwareTitle(self):
return 'Title %s' % self.generateNewId()
def test(self):
self._makeTree()
instance = self.software_instance
remote_user = instance.getReference()
content_type = "application/hal+json"
# 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())
def getNewHttpConnection(api_netloc):
if api_scheme == 'https':
raise Exception('Please connect directly to the Zope server')
connection = httplib.HTTPConnection(api_netloc)
return connection
#####################################################
# Access the master home page hal
#####################################################
connection = getNewHttpConnection(api_netloc)
connection.request(
method='GET',
url='%s/web_site_module/hateoas/Base_getHateoasMaster' % \
self.portal.absolute_url(),
headers={
'REMOTE_USER': remote_user,
'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 instance hal
#####################################################
user_link_dict = home_page_hal['_links']['action_object_jump']
self.assertNotEqual(user_link_dict, None)
connection = getNewHttpConnection(api_netloc)
connection.request(
method=user_link_dict.get('method', 'GET'),
url=user_link_dict['href'],
headers={
'REMOTE_USER': remote_user,
'Accept': content_type,
},
body="",
)
response = connection.getresponse()
self.assertEquals(response.status, 200)
self.assertEquals(response.getheader('Content-Type'), content_type)
instance_hal = json.loads(response.read())
#####################################################
# Fetch instance informations
#####################################################
request_link_dict = hateoasGetLinkFromLinks(
instance_hal['_links']['action_object_slap'],
'getHateoasInformation'
)
self.assertNotEqual(request_link_dict, None)
connection = getNewHttpConnection(api_netloc)
connection.request(
method=user_link_dict.get('method', 'GET'),
url=user_link_dict['href'],
headers={
'REMOTE_USER': remote_user,
'Accept': content_type,
},
body="",
)
response = connection.getresponse()
self.assertEquals(response.status, 200)
self.assertEquals(response.getheader('Content-Type'), content_type)
self.tic()
#####################################################
# Get instance news
#####################################################
request_link_dict = hateoasGetLinkFromLinks(
instance_hal['_links']['action_object_slap'],
'getHateoasNews'
)
self.assertNotEqual(request_link_dict, None)
connection = getNewHttpConnection(api_netloc)
connection.request(
method=user_link_dict.get('method', 'GET'),
url=user_link_dict['href'],
headers={
'REMOTE_USER': remote_user,
'Accept': content_type,
},
body="",
)
response = connection.getresponse()
self.assertEquals(response.status, 200)
self.assertEquals(response.getheader('Content-Type'), content_type)
self.tic()
#####################################################
# Get hosting subscription of instance
#####################################################
# XXX can be simpler and doesn't need getHateoasRelatedHostingSubscription script
request_link_dict = hateoasGetLinkFromLinks(
instance_hal['_links']['action_object_slap'],
'getHateoasRelatedHostingSubscription'
)
self.assertNotEqual(request_link_dict, None)
connection = getNewHttpConnection(api_netloc)
connection.request(
method=request_link_dict.get('method', 'GET'),
url=request_link_dict['href'],
headers={
'REMOTE_USER': remote_user,
'Accept': content_type,
},
body="",
)
response = connection.getresponse()
self.assertEquals(response.status, 200)
self.assertEquals(response.getheader('Content-Type'), content_type)
subscription_hal = json.loads(response.read())
self.tic()
request_link_dict = hateoasGetLinkFromLinks(
subscription_hal['_links']['action_object_jump'],
'Hosting Subscription'
)
self.assertNotEqual(request_link_dict, None)
connection = getNewHttpConnection(api_netloc)
connection.request(
method=request_link_dict.get('method', 'GET'),
url=request_link_dict['href'],
headers={
'REMOTE_USER': remote_user,
'Accept': content_type,
},
body="",
)
response = connection.getresponse()
self.assertEquals(response.status, 200)
self.assertEquals(response.getheader('Content-Type'), content_type)
subscription_hal = json.loads(response.read())
self.tic()
#####################################################
# Get hosting subscription's instance list
#####################################################
user_link_dict = hateoasGetLinkFromLinks(
subscription_hal['_links']['action_object_slap'],
'getHateoasInstanceList'
)
self.assertNotEqual(user_link_dict, None)
connection = getNewHttpConnection(api_netloc)
connection.request(
method=user_link_dict.get('method', 'GET'),
url=user_link_dict['href'],
headers={
'REMOTE_USER': remote_user,
'Accept': content_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
#####################################################
subscription_link_dict = instance_collection_hal['_links']\
['content'][0]
self.assertNotEqual(subscription_link_dict, None)
connection = getNewHttpConnection(api_netloc)
connection.request(
method=subscription_link_dict.get('method', 'GET'),
url=subscription_link_dict['href'],
headers={
'REMOTE_USER': remote_user,
'Accept': content_type,
},
body="",
)
response = connection.getresponse()
self.assertEquals(response.status, 200)
self.assertEquals(response.getheader('Content-Type'), content_type)
instance_hal = json.loads(response.read())
...@@ -361,6 +361,33 @@ class TestSlapOSBase_getHateoasMaster(TestSlapOSHypermediaMixin): ...@@ -361,6 +361,33 @@ class TestSlapOSBase_getHateoasMaster(TestSlapOSHypermediaMixin):
}, },
}, indent=2)) }, indent=2))
@simulate('Base_getRequestUrl', '*args, **kwargs',
'return "http://example.org/bar"')
@simulate('Base_getRequestHeader', '*args, **kwargs',
'return "application/hal+json"')
@changeSkin('Hal')
def test_getHateoasMaster_instance_result(self):
self._makeTree()
self.login(self.software_instance.getReference())
self.changeSkin('Hal')
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/hal+json"
)
self.assertEquals(result, json.dumps({
'_links': {
"self": {
"href": "http://example.org/bar"
},
"action_object_jump": {
"href": "%s/ERP5Document_getHateoas" % self.software_instance.absolute_url(),
"title": "Software Instance"
},
},
}, indent=2))
class TestSlapOSPerson_requestHateoasHostingSubscription(TestSlapOSHypermediaMixin): class TestSlapOSPerson_requestHateoasHostingSubscription(TestSlapOSHypermediaMixin):
@changeSkin('Hal') @changeSkin('Hal')
......
SlapOSHypermedia
\ No newline at end of file
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