Skip to content

S
slapos.core
Commit 794fca99 authored by Rafael Monnerat's avatar Rafael Monnerat 👻
Browse files
Options

cleanup master: Remove old rest API 

  Remove old rest API as it will be replaced by hal implementation
parent 78407bb9
Show whitespace changes
Inline Side-by-side
Showing with 0 additions and 3684 deletions
master/bt5/slapos_configurator/TestTemplateItem/portal_components/test.erp5.testSlapOSConfigurator.py
View file @ 794fca99
......@@ -196,8 +196,6 @@ class TestSlapOSConfigurator(testSlapOSMixin):
'slapos_cloud',
'slapos_slap_tool',
'slapos_category',
'slapos_rest_api_tool_portal_type',
'slapos_rest_api',
'slapos_pdm',
'slapos_crm',
'slapos_accounting',
......
master/bt5/slapos_erp5/TestTemplateItem/portal_components/test.erp5.testSlapOSERP5SkinSelection.py
View file @ 794fca99
......@@ -132,7 +132,6 @@ slapos_core
slapos_crm
slapos_payzen
slapos_pdm
slapos_rest_api
slapos_simulation
slapos_slap_tool
slapos_web
......@@ -244,7 +243,6 @@ slapos_core
slapos_crm
slapos_payzen
slapos_pdm
slapos_rest_api
slapos_simulation
slapos_slap_tool
slapos_web
......@@ -351,7 +349,6 @@ slapos_core
slapos_crm
slapos_payzen
slapos_pdm
slapos_rest_api
slapos_simulation
slapos_slap_tool
slapos_web
......@@ -459,7 +456,6 @@ slapos_core
slapos_crm
slapos_payzen
slapos_pdm
slapos_rest_api
slapos_simulation
slapos_slap_tool
slapos_web
......@@ -564,7 +560,6 @@ slapos_core
slapos_crm
slapos_payzen
slapos_pdm
slapos_rest_api
slapos_simulation
slapos_slap_tool
slapos_web
......@@ -670,7 +665,6 @@ slapos_core
slapos_crm
slapos_payzen
slapos_pdm
slapos_rest_api
slapos_simulation
slapos_slap_tool
slapos_web
......@@ -775,7 +769,6 @@ slapos_core
slapos_crm
slapos_payzen
slapos_pdm
slapos_rest_api
slapos_simulation
slapos_slap_tool
slapos_web
......@@ -881,7 +874,6 @@ slapos_core
slapos_crm
slapos_payzen
slapos_pdm
slapos_rest_api
slapos_simulation
slapos_slap_tool
slapos_web
......@@ -986,7 +978,6 @@ slapos_core
slapos_crm
slapos_payzen
slapos_pdm
slapos_rest_api
slapos_simulation
slapos_slap_tool
slapos_web
......@@ -1093,7 +1084,6 @@ slapos_core
slapos_crm
slapos_payzen
slapos_pdm
slapos_rest_api
slapos_simulation
slapos_slap_tool
slapos_web
......@@ -1201,7 +1191,6 @@ slapos_core
slapos_crm
slapos_payzen
slapos_pdm
slapos_rest_api
slapos_simulation
slapos_slap_tool
slapos_web
......
master/bt5/slapos_erp5/bt/dependency_list
View file @ 794fca99
......@@ -13,7 +13,6 @@ slapos_cache
slapos_category
slapos_cloud
slapos_ecoallocation
slapos_rest_api
slapos_slap_tool
slapos_crm
slapos_pdm
......
master/bt5/slapos_rest_api/ActionTemplateItem/portal_types/System%20Preference/slapos_rest_api_preference.xml deleted 100644 → 0
View file @ 78407bb9
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="ActionInformation" module="Products.CMFCore.ActionInformation"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>action</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAI=</string> </persistent>
</value>
</item>
<item>
<key> <string>categories</string> </key>
<value>
<tuple>
<string>action_type/object_view</string>
</tuple>
</value>
</item>
<item>
<key> <string>category</string> </key>
<value> <string>object_view</string> </value>
</item>
<item>
<key> <string>condition</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>description</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>icon</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>slapos_rest_api_preference</string> </value>
</item>
<item>
<key> <string>permissions</string> </key>
<value>
<tuple>
<string>View</string>
</tuple>
</value>
</item>
<item>
<key> <string>portal_type</string> </key>
<value> <string>Action Information</string> </value>
</item>
<item>
<key> <string>priority</string> </key>
<value> <float>11.0</float> </value>
</item>
<item>
<key> <string>title</string> </key>
<value> <string>SlapOS Rest API</string> </value>
</item>
<item>
<key> <string>visible</string> </key>
<value> <int>1</int> </value>
</item>
</dictionary>
</pickle>
</record>
<record id="2" aka="AAAAAAAAAAI=">
<pickle>
<global name="Expression" module="Products.CMFCore.Expression"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>text</string> </key>
<value> <string>string:${object_url}/SystemPreference_viewSlapOSRestAPI</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
master/bt5/slapos_rest_api/DocumentTemplateItem/portal_components/SlapOSRestAPIV1.py deleted 100644 → 0
View file @ 78407bb9
# -*- coding: utf-8 -*-
##############################################################################
#
# Copyright (c) 2012 Nexedi SA and Contributors. All Rights Reserved.
# Łukasz Nowak <luke@nexedi.com>
# Romain Courteaud <romain@nexedi.com>
#
# 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 advised 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., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################
from Acquisition import Implicit
from AccessControl import ClassSecurityInfo, getSecurityManager, Unauthorized
from Products.SlapOS.SlapOSMachineAuthenticationPlugin import getUserByLogin
from Products.ERP5Type import Permissions
from ComputedAttribute import ComputedAttribute
from zLOG import LOG, ERROR
from lxml import etree
import json
import transaction
from App.Common import rfc1123_date
from DateTime import DateTime
import re
class WrongRequest(Exception):
pass
def etreeXml(d):
r = etree.Element('instance')
for k, v in d.iteritems():
v = str(v)
etree.SubElement(r, "parameter", attrib={'id': k}).text = v
return etree.tostring(r, pretty_print=True, xml_declaration=True,
encoding='utf-8')
def jsonify(d):
return json.dumps(d, indent=2)
def requireHeader(header_dict):
def outer(fn):
def wrapperRequireHeader(self, *args, **kwargs):
problem_dict = {}
for header, value in header_dict.iteritems():
send_header = self.REQUEST.getHeader(header)
if send_header is None or not re.match(value, send_header):
problem_dict[header] = 'Header with value %r is required.' % value
if not problem_dict:
return fn(self, *args, **kwargs)
else:
self.REQUEST.response.setStatus(400)
self.REQUEST.response.setBody(jsonify(problem_dict))
return self.REQUEST.response
wrapperRequireHeader.__doc__ = fn.__doc__
return wrapperRequireHeader
return outer
def supportModifiedSince(document_url_id):
def outer(fn):
def wrapperSupportModifiedSince(self, *args, **kwargs):
modified_since = self.REQUEST.getHeader('If-Modified-Since')
if modified_since is not None:
# RFC 2616 If-Modified-Since support
try:
modified_since = DateTime(self.REQUEST.getHeader('If-Modified-Since'))
except Exception:
# client sent wrong header, shall be ignored
pass
else:
if modified_since <= DateTime():
# client send date before current time, shall continue and
# compare with second precision, as client by default shall set
# If-Modified-Since to last known Last-Modified value
document = self.restrictedTraverse(getattr(self, document_url_id))
document_date = document.getModificationDate() or \
document.bobobase_modification_time()
if int(document_date.timeTime()) <= int(modified_since.timeTime()):
# document was not modified since
self.REQUEST.response.setStatus(304)
return self.REQUEST.response
return fn(self, *args, **kwargs)
wrapperSupportModifiedSince.__doc__ = fn.__doc__
return wrapperSupportModifiedSince
return outer
def encode_utf8(s):
return s.encode('utf-8')
def requireParameter(json_dict, optional_key_list=None):
if optional_key_list is None:
optional_key_list = []
def outer(fn):
def wrapperRequireJson(self, *args, **kwargs):
self.jbody = {}
error_dict = {}
for key, type_ in json_dict.iteritems():
if key not in self.REQUEST:
if key not in optional_key_list:
error_dict[key] = 'Missing.'
else:
value = self.REQUEST[key]
method = None
if type(type_) in (tuple, list):
type_, method = type_
if type_ == unicode:
value = '"%s"' % value
try:
value = json.loads(value)
except Exception:
error_dict[key] = 'Malformed value.'
else:
if not isinstance(value, type_):
error_dict[key] = '%s is not %s.' % (type(value
).__name__, type_.__name__)
if method is None:
self.jbody[key] = value
else:
try:
self.jbody[key] = method(value)
except Exception:
error_dict[key] = 'Malformed value.'
if error_dict:
self.REQUEST.response.setStatus(400)
self.REQUEST.response.setBody(jsonify(error_dict))
return self.REQUEST.response
return fn(self, *args, **kwargs)
wrapperRequireJson.__doc__ = fn.__doc__
return wrapperRequireJson
return outer
def requireJson(json_dict, optional_key_list=None):
if optional_key_list is None:
optional_key_list = []
def outer(fn):
def wrapperRequireJson(self, *args, **kwargs):
self.REQUEST.stdin.seek(0)
try:
self.jbody = json.load(self.REQUEST.stdin)
except Exception:
self.REQUEST.response.setStatus(400)
self.REQUEST.response.setBody(jsonify(
{'error': 'Data is not json object.'}))
return self.REQUEST.response
else:
error_dict = {}
for key, type_ in json_dict.iteritems():
if key not in self.jbody:
if key not in optional_key_list:
error_dict[key] = 'Missing.'
elif key in self.jbody:
method = None
if type(type_) in (tuple, list):
type_, method = type_
if not isinstance(self.jbody[key], type_):
error_dict[key] = '%s is not %s.' % (type(self.jbody[key]
).__name__, type_.__name__)
if method is not None:
try:
self.jbody[key] = method(self.jbody[key])
except Exception:
error_dict[key] = 'Malformed value.'
if error_dict:
self.REQUEST.response.setStatus(400)
self.REQUEST.response.setBody(jsonify(error_dict))
return self.REQUEST.response
return fn(self, *args, **kwargs)
wrapperRequireJson.__doc__ = fn.__doc__
return wrapperRequireJson
return outer
def responseSupport(anonymous=False):
def outer(fn):
def wrapperResponseSupport(self, *args, **kwargs):
self.REQUEST.response.setHeader('Content-Type', 'application/json')
request_headers = self.REQUEST.getHeader('Access-Control-Request-Headers')
if request_headers:
self.REQUEST.response.setHeader('Access-Control-Allow-Headers',
request_headers)
self.REQUEST.response.setHeader('Access-Control-Allow-Origin', '*')
self.REQUEST.response.setHeader('Access-Control-Allow-Methods', 'DELETE, PUT, POST, '
'GET, OPTIONS')
if not anonymous:
if getSecurityManager().getUser().getId() is None:
if self.REQUEST.get('USER_CREATION_IN_PROGRESS') is not None:
# inform that user is not ready yet
self.REQUEST.response.setStatus(202)
self.REQUEST.response.setBody(jsonify(
{'status':'User under creation.'}))
else:
# force login
self.REQUEST.response.setStatus(401)
self.REQUEST.response.setHeader('WWW-Authenticate', 'Bearer realm="%s"'%
self.getAPIRoot())
self.REQUEST.response.setHeader('Location', self.getPortalObject()\
.portal_preferences.getPreferredRestApiTokenServerUrl())
return self.REQUEST.response
else:
user_name = self.getPortalObject().portal_membership\
.getAuthenticatedMember()
user_document = getUserByLogin(self.getPortalObject(),
str(user_name))
if len(user_document) != 1:
transaction.abort()
LOG('SlapOSRestApiV1', ERROR,
'Currenty logged in user %r wrong document list %r.'%
(user_name, user_document))
self.REQUEST.response.setStatus(500)
self.REQUEST.response.setBody(jsonify({'error':
'There is system issue, please try again later.'}))
return self.REQUEST.response
self.user_url = user_document[0].getRelativeUrl()
return fn(self, *args, **kwargs)
wrapperResponseSupport.__doc__ = fn.__doc__
return wrapperResponseSupport
return outer
def extractDocument(portal_type):
if not isinstance(portal_type, (list, tuple)):
portal_type = [portal_type]
def outer(fn):
def wrapperExtractDocument(self, *args, **kwargs):
if not self.REQUEST['traverse_subpath']:
self.REQUEST.response.setStatus(404)
return self.REQUEST.response
path = self.REQUEST['traverse_subpath'][:2]
try:
document = self.getPortalObject().restrictedTraverse(path)
if getattr(document, 'getPortalType', None) is None or \
document.getPortalType() not in portal_type:
raise WrongRequest('%r is neiter of %s' % (path, ', '.join(
portal_type)))
self.document_url = document.getRelativeUrl()
except WrongRequest:
LOG('SlapOSRestApiV1', ERROR,
'Problem while trying to find document:', error=True)
self.REQUEST.response.setStatus(404)
except (Unauthorized, KeyError):
self.REQUEST.response.setStatus(404)
except Exception:
LOG('SlapOSRestApiV1', ERROR,
'Problem while trying to find instance:', error=True)
self.REQUEST.response.setStatus(500)
self.REQUEST.response.setBody(jsonify({'error':
'There is system issue, please try again later.'}))
else:
self.REQUEST['traverse_subpath'] = self.REQUEST['traverse_subpath'][2:]
return fn(self, *args, **kwargs)
return self.REQUEST.response
wrapperExtractDocument.__doc__ = fn.__doc__
return wrapperExtractDocument
return outer
class GenericPublisher(Implicit):
@responseSupport(True)
def OPTIONS(self, *args, **kwargs):
"""HTTP OPTIONS implementation"""
self.REQUEST.response.setStatus(204)
return self.REQUEST.response
def __before_publishing_traverse__(self, self2, request):
path = request['TraversalRequestNameStack']
subpath = path[:]
path[:] = []
subpath.reverse()
request.set('traverse_subpath', subpath)
class InstancePublisher(GenericPublisher):
"""Instance publisher"""
@responseSupport()
@requireHeader({'Content-Type': '^application/json.*'})
@requireJson(dict(
title=(unicode, encode_utf8),
connection=dict
), ['title', 'connection'])
@extractDocument(['Software Instance', 'Slave Instance'])
def PUT(self):
"""Instance PUT support"""
d = {}
try:
self.REQUEST.response.setStatus(204)
software_instance = self.restrictedTraverse(self.document_url)
if 'title' in self.jbody and \
software_instance.getTitle() != self.jbody['title']:
software_instance.setTitle(self.jbody['title'])
d['title'] = 'Modified.'
self.REQUEST.response.setStatus(200)
if 'connection' in self.jbody:
xml = etreeXml(self.jbody['connection'])
if xml != software_instance.getConnectionXml():
software_instance.setConnectionXml(xml)
d['connection'] = 'Modified.'
self.REQUEST.response.setStatus(200)
except Exception:
transaction.abort()
LOG('SlapOSRestApiV1', ERROR,
'Problem while modifying:', error=True)
self.REQUEST.response.setStatus(500)
self.REQUEST.response.setBody(jsonify({'error':
'There is system issue, please try again later.'}))
else:
if d:
self.REQUEST.response.setBody(jsonify(d))
return self.REQUEST.response
@requireHeader({'Content-Type': '^application/json.*'})
@requireJson(dict(log=unicode))
@extractDocument(['Software Instance', 'Slave Instance'])
def __bang(self):
try:
self.restrictedTraverse(self.document_url
).bang(bang_tree=True, comment=self.jbody['log'])
except Exception:
LOG('SlapOSRestApiV1', ERROR,
'Problem while trying to generate instance dict:', error=True)
self.REQUEST.response.setStatus(500)
self.REQUEST.response.setBody(jsonify({'error':
'There is system issue, please try again later.'}))
else:
self.REQUEST.response.setStatus(204)
return self.REQUEST.response
@requireHeader({'Content-Type': '^application/json.*'})
@requireJson(dict(
slave=bool,
software_release=(unicode, encode_utf8),
title=(unicode, encode_utf8),
software_type=(unicode, encode_utf8),
parameter=(dict, etreeXml),
sla=(dict, etreeXml),
status=(unicode, encode_utf8),
))
def __request(self):
request_dict = {}
for k_j, k_i in (
('software_release', 'software_release'),
('title', 'software_title'),
('software_type', 'software_type'),
('parameter', 'instance_xml'),
('sla', 'sla_xml'),
('slave', 'shared'),
('status', 'state')
):
request_dict[k_i] = self.jbody[k_j]
if request_dict['state'] not in ['started', 'stopped', 'destroyed']:
self.REQUEST.response.setStatus(400)
self.REQUEST.response.setBody(jsonify(
{'status': 'Status shall be one of: started, stopped, destroyed.'}))
return self.REQUEST.response
try:
self.restrictedTraverse(self.user_url
).requestSoftwareInstance(**request_dict)
except Exception:
transaction.abort()
LOG('SlapOSRestApiV1', ERROR,
'Problem with person.requestSoftwareInstance:', error=True)
self.REQUEST.response.setStatus(500)
self.REQUEST.response.setBody(jsonify({'error':
'There is system issue, please try again later.'}))
return self.REQUEST.response
self.REQUEST.response.setStatus(202)
self.REQUEST.response.setBody(jsonify({'status':'processing'}))
return self.REQUEST.response
@requireParameter(dict(
slave=bool,
software_release=(unicode, encode_utf8),
software_type=(unicode, encode_utf8),
sla=dict,
))
def __allocable(self):
try:
user = self.restrictedTraverse(self.user_url)
user_portal_type = user.getPortalType()
if user_portal_type == 'Person':
pass
elif user_portal_type == 'Software Instance':
hosting_subscription = user.getSpecialiseValue(
portal_type="Hosting Subscription")
user = hosting_subscription.getDestinationSectionValue(
portal_type="Person")
else:
raise NotImplementedError, "Can not get Person document"
result = user.Person_restrictMethodAsShadowUser(
shadow_document=user,
callable_object=user.Person_findPartition,
argument_list=[
self.jbody['software_release'],
self.jbody['software_type'],
('Software Instance', 'Slave Instance')[int(self.jbody['slave'])],
self.jbody['sla']],
argument_dict={
'test_mode': True})
except Exception:
transaction.abort()
LOG('SlapOSRestApiV1', ERROR,
'Problem with person.allocable:', error=True)
self.REQUEST.response.setStatus(500)
self.REQUEST.response.setBody(jsonify({'error':
'There is system issue, please try again later.'}))
return self.REQUEST.response
self.REQUEST.response.setStatus(200)
self.REQUEST.response.setHeader('Cache-Control',
'no-cache, no-store')
self.REQUEST.response.setBody(jsonify({'result': result}))
return self.REQUEST.response
@extractDocument(['Software Instance', 'Slave Instance'])
@supportModifiedSince('document_url')
def __instance_info(self):
certificate = False
software_instance = self.restrictedTraverse(self.document_url)
if self.REQUEST['traverse_subpath'] and self.REQUEST[
'traverse_subpath'][-1] == 'certificate':
certificate = True
try:
if certificate:
d = {
"ssl_key": software_instance.getSslKey(),
"ssl_certificate": software_instance.getSslCertificate()
}
else:
d = {
"title": software_instance.getTitle(),
"status": software_instance.getSlapState(),
"software_release": software_instance.getUrlString(),
"software_type": software_instance.getSourceReference(),
"slave": software_instance.getPortalType() == 'Slave Instance',
"connection": software_instance.getConnectionXmlAsDict(),
"parameter": software_instance.getInstanceXmlAsDict(),
"sla": software_instance.getSlaXmlAsDict(),
"children_list": [self.getAPIRoot() + '/' + q.getRelativeUrl() \
for q in software_instance.getPredecessorValueList()],
"partition": { # not ready yet
"public_ip": [],
"private_ip": [],
"tap_interface": "",
}
}
except Exception:
LOG('SlapOSRestApiV1', ERROR,
'Problem while trying to generate instance dict:', error=True)
self.REQUEST.response.setStatus(500)
self.REQUEST.response.setBody(jsonify({'error':
'There is system issue, please try again later.'}))
else:
self.REQUEST.response.setStatus(200)
# Note: Last-Modified will result in resending certificate many times,
# because each modification of instance will result in new Last-Modified
# TODO: Separate certificate from instance and use correct
# Last-Modified header of subobject containing certificate
self.REQUEST.response.setHeader('Last-Modified',
rfc1123_date(software_instance.getModificationDate()))
# Say that content is publicly cacheable. It is only required in order to
# *force* storing content on clients' disk in case of using HTTPS
self.REQUEST.response.setHeader('Cache-Control', 'must-revalidate')
self.REQUEST.response.setBody(jsonify(d))
return self.REQUEST.response
def __instance_list(self):
kw = dict(
portal_type=('Software Instance', 'Slave Instance'),
)
d = {"list": []}
a = d['list'].append
for si in self.getPortalObject().portal_catalog(**kw):
a('/'.join([self.getAPIRoot(), 'instance', si.getRelativeUrl()]))
try:
d['list'][0]
except IndexError:
# no results, so nothing to return
self.REQUEST.response.setStatus(204)
else:
self.REQUEST.response.setStatus(200)
self.REQUEST.response.setBody(jsonify(d))
return self.REQUEST.response
@responseSupport()
def __call__(self):
"""Instance GET/POST support"""
if self.REQUEST['REQUEST_METHOD'] == 'POST':
if self.REQUEST['traverse_subpath'] and \
self.REQUEST['traverse_subpath'][-1] == 'bang':
self.__bang()
else:
self.__request()
elif self.REQUEST['REQUEST_METHOD'] == 'GET':
if self.REQUEST['traverse_subpath']:
if self.REQUEST['traverse_subpath'][-1] == 'request':
self.__allocable()
else:
self.__instance_info()
else:
self.__instance_list()
class ComputerPublisher(GenericPublisher):
@responseSupport()
@requireHeader({'Content-Type': '^application/json.*'})
@extractDocument('Computer')
@requireJson(dict(
partition=list,
software=list
), ['partition', 'software'])
def PUT(self):
"""Computer PUT support"""
computer = self.restrictedTraverse(self.document_url)
error_dict = {}
def getErrorDict(list_, key_list, prefix):
no = 0
for dict_ in list_:
error_list = []
if not isinstance(dict_, dict):
error_list.append('Not a dict.')
else:
for k in key_list:
if k not in dict_:
error_list.append('Missing key "%s".' % k)
elif not isinstance(dict_[k], unicode):
error_list.append('Key "%s" is not unicode.' % k)
elif k == 'status' and dict_[k] not in ['installed',
'uninstalled', 'error']:
error_list.append('Status "%s" is incorrect.' % dict_[k])
if len(error_list) > 0:
error_dict['%s_%s' % (prefix, no)] = error_list
no += 1
return error_dict
error_dict = {}
transmitted = False
if 'partition' in self.jbody:
error_dict.update(getErrorDict(self.jbody['partition'],
('title', 'public_ip', 'private_ip', 'tap_interface'), 'partition'))
transmitted = True
if 'software' in self.jbody:
error_dict.update(getErrorDict(self.jbody['software'],
('software_release', 'status', 'log'), 'software'))
transmitted = True
# XXX: Support status as enum.
if error_dict:
self.REQUEST.response.setStatus(400)
self.REQUEST.response.setBody(jsonify(error_dict))
return self.REQUEST.response
if transmitted:
try:
computer.Computer_updateFromJson(self.jbody)
except Exception:
transaction.abort()
LOG('SlapOSRestApiV1', ERROR,
'Problem while trying to update computer:', error=True)
self.REQUEST.response.setStatus(500)
self.REQUEST.response.setBody(jsonify({'error':
'There is system issue, please try again later.'}))
return self.REQUEST.response
self.REQUEST.response.setStatus(204)
return self.REQUEST.response
class StatusPublisher(GenericPublisher):
@responseSupport()
def __call__(self):
"""Log GET support"""
if self.REQUEST['REQUEST_METHOD'] == 'POST':
self.REQUEST.response.setStatus(404)
return self.REQUEST.response
elif self.REQUEST['REQUEST_METHOD'] == 'GET':
if self.REQUEST['traverse_subpath']:
self.__status_info()
else:
self.__status_list()
def __status_list(self):
portal = self.getPortalObject()
open_friend = portal.restrictedTraverse(
"portal_categories/allocation_scope/open/friend", None).getUid()
open_personal = portal.restrictedTraverse(
"portal_categories/allocation_scope/open/personal", None).getUid()
open_public = portal.restrictedTraverse(
"portal_categories/allocation_scope/open/public", None).getUid()
kw = dict(
validation_state="validated",
portal_type=['Computer', 'Software Installation', 'Software Instance']
)
d = {"list": []}
a = d['list'].append
for si in self.getPortalObject().portal_catalog(**kw):
if (si.getPortalType() == "Software Instance" or \
si.getPortalType() == "Software Installation") and \
si.getSlapState() not in ['start_requested','stop_requested']:
continue
if si.getPortalType() == "Computer" and \
si.getAllocationScopeUid() not in [open_friend, open_personal, open_public]:
continue
a('/'.join([self.getAPIRoot(), 'status', si.getRelativeUrl()]))
try:
d['list'][0]
except IndexError:
# no results, so nothing to return
self.REQUEST.response.setStatus(204)
else:
self.REQUEST.response.setStatus(200)
self.REQUEST.response.setHeader('Cache-Control',
'max-age=300, private')
self.REQUEST.response.setBody(jsonify(d))
return self.REQUEST.response
# XXX Use a decated document to keep the history
@extractDocument(['Computer', 'Software Installation', 'Software Instance'])
# @supportModifiedSince('document_url')
def __status_info(self):
certificate = False
document = self.restrictedTraverse(self.document_url)
try:
memcached_dict = self.getPortalObject().portal_memcached.getMemcachedDict(
key_prefix='slap_tool',
plugin_path='portal_memcached/default_memcached_plugin')
try:
d = memcached_dict[document.getReference()]
except KeyError:
d = {
"user": "SlapOS Master",
'created_at': '%s' % rfc1123_date(DateTime()),
"text": "#error no data found for %s" % document.getReference()
}
else:
d = json.loads(d)
except Exception:
LOG('SlapOSRestApiV1', ERROR,
'Problem while trying to generate status information:', error=True)
self.REQUEST.response.setStatus(500)
self.REQUEST.response.setBody(jsonify({'error':
'There is system issue, please try again later.'}))
else:
d['@document'] = self.document_url
self.REQUEST.response.setStatus(200)
self.REQUEST.response.setHeader('Cache-Control',
'max-age=300, private')
self.REQUEST.response.setBody(jsonify(d))
return self.REQUEST.response
class SlapOSRestAPIV1(Implicit):
security = ClassSecurityInfo()
security.declareObjectProtected(Permissions.AccessContentsInformation)
security.declarePublic('instance')
security.declarePublic('getAPIRoot')
def getAPIRoot(self):
"""Returns the root of API"""
return self.absolute_url() + '/v1'
@ComputedAttribute
def instance(self):
"""Instance publisher"""
return InstancePublisher().__of__(self)
security.declarePublic('computer')
@ComputedAttribute
def computer(self):
"""Computer publisher"""
return ComputerPublisher().__of__(self)
security.declarePublic('log')
@ComputedAttribute
def status(self):
"""Status publisher"""
return StatusPublisher().__of__(self)
@responseSupport(True)
def OPTIONS(self, *args, **kwargs):
"""HTTP OPTIONS implementation"""
self.REQUEST.response.setStatus(204)
return self.REQUEST.response
security.declarePublic('__call__')
@responseSupport(True)
def __call__(self):
"""Possible API discovery"""
self.REQUEST.response.setStatus(400)
return self.REQUEST.response
master/bt5/slapos_rest_api/DocumentTemplateItem/portal_components/SlapOSRestAPIV1.xml deleted 100644 → 0
View file @ 78407bb9
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="Document Component" module="erp5.portal_type"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>_recorded_property_dict</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAI=</string> </persistent>
</value>
</item>
<item>
<key> <string>default_reference</string> </key>
<value> <string>SlapOSRestAPIV1</string> </value>
</item>
<item>
<key> <string>description</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>SlapOSRestAPIV1</string> </value>
</item>
<item>
<key> <string>portal_type</string> </key>
<value> <string>Document Component</string> </value>
</item>
<item>
<key> <string>sid</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>version</string> </key>
<value> <string>erp5</string> </value>
</item>
<item>
<key> <string>workflow_history</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAM=</string> </persistent>
</value>
</item>
</dictionary>
</pickle>
</record>
<record id="2" aka="AAAAAAAAAAI=">
<pickle>
<global name="PersistentMapping" module="Persistence.mapping"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>data</string> </key>
<value>
<dictionary/>
</value>
</item>
</dictionary>
</pickle>
</record>
<record id="3" aka="AAAAAAAAAAM=">
<pickle>
<global name="PersistentMapping" module="Persistence.mapping"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>data</string> </key>
<value>
<dictionary>
<item>
<key> <string>component_validation_workflow</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAQ=</string> </persistent>
</value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</pickle>
</record>
<record id="4" aka="AAAAAAAAAAQ=">
<pickle>
<global name="WorkflowHistoryList" module="Products.ERP5Type.patches.WorkflowTool"/>
</pickle>
<pickle>
<tuple>
<none/>
<list>
<dictionary>
<item>
<key> <string>action</string> </key>
<value> <string>validate</string> </value>
</item>
<item>
<key> <string>validation_state</string> </key>
<value> <string>validated</string> </value>
</item>
</dictionary>
</list>
</tuple>
</pickle>
</record>
</ZopeData>
master/bt5/slapos_rest_api/PropertySheetTemplateItem/portal_property_sheets/SlapOSRestAPISystemPreference.xml deleted 100644 → 0
View file @ 78407bb9
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="Property Sheet" module="erp5.portal_type"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>_count</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAI=</string> </persistent>
</value>
</item>
<item>
<key> <string>_mt_index</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAM=</string> </persistent>
</value>
</item>
<item>
<key> <string>_tree</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAQ=</string> </persistent>
</value>
</item>
<item>
<key> <string>description</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>SlapOSRestAPISystemPreference</string> </value>
</item>
<item>
<key> <string>portal_type</string> </key>
<value> <string>Property Sheet</string> </value>
</item>
</dictionary>
</pickle>
</record>
<record id="2" aka="AAAAAAAAAAI=">
<pickle>
<global name="Length" module="BTrees.Length"/>
</pickle>
<pickle> <int>0</int> </pickle>
</record>
<record id="3" aka="AAAAAAAAAAM=">
<pickle>
<global name="OOBTree" module="BTrees.OOBTree"/>
</pickle>
<pickle>
<none/>
</pickle>
</record>
<record id="4" aka="AAAAAAAAAAQ=">
<pickle>
<global name="OOBTree" module="BTrees.OOBTree"/>
</pickle>
<pickle>
<none/>
</pickle>
</record>
</ZopeData>
master/bt5/slapos_rest_api/PropertySheetTemplateItem/portal_property_sheets/SlapOSRestAPISystemPreference/preferred_rest_api_token_server_url_property.xml deleted 100644 → 0
View file @ 78407bb9
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="Standard Property" module="erp5.portal_type"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>_local_properties</string> </key>
<value>
<tuple>
<dictionary>
<item>
<key> <string>id</string> </key>
<value> <string>mode</string> </value>
</item>
<item>
<key> <string>type</string> </key>
<value> <string>string</string> </value>
</item>
</dictionary>
</tuple>
</value>
</item>
<item>
<key> <string>categories</string> </key>
<value>
<tuple>
<string>elementary_type/string</string>
</tuple>
</value>
</item>
<item>
<key> <string>description</string> </key>
<value> <string>Stores token server URL</string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>preferred_rest_api_token_server_url_property</string> </value>
</item>
<item>
<key> <string>mode</string> </key>
<value> <string>w</string> </value>
</item>
<item>
<key> <string>portal_type</string> </key>
<value> <string>Standard Property</string> </value>
</item>
<item>
<key> <string>preference</string> </key>
<value> <int>1</int> </value>
</item>
<item>
<key> <string>property_default</string> </key>
<value> <string>python: \'\'</string> </value>
</item>
<item>
<key> <string>read_permission</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>write_permission</string> </key>
<value> <string>Manage properties</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
master/bt5/slapos_rest_api/SkinTemplateItem/portal_skins/slapos_rest_api.xml deleted 100644 → 0
View file @ 78407bb9
<?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_rest_api</string> </value>
</item>
<item>
<key> <string>title</string> </key>
<value> <string></string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
master/bt5/slapos_rest_api/SkinTemplateItem/portal_skins/slapos_rest_api/SystemPreference_viewSlapOSRestAPI.xml deleted 100644 → 0
View file @ 78407bb9
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="ERP5Form" module="Products.ERP5Form.Form"/>
</pickle>
<pickle>
<dictionary>
<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/>
</value>
</item>
</dictionary>
</state>
</object>
</value>
</item>
<item>
<key> <string>_objects</string> </key>
<value>
<tuple/>
</value>
</item>
<item>
<key> <string>action</string> </key>
<value> <string>Base_edit</string> </value>
</item>
<item>
<key> <string>description</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>edit_order</string> </key>
<value>
<list/>
</value>
</item>
<item>
<key> <string>encoding</string> </key>
<value> <string>UTF-8</string> </value>
</item>
<item>
<key> <string>enctype</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>group_list</string> </key>
<value>
<list>
<string>left</string>
<string>right</string>
<string>center</string>
<string>bottom</string>
<string>hidden</string>
</list>
</value>
</item>
<item>
<key> <string>groups</string> </key>
<value>
<dictionary>
<item>
<key> <string>bottom</string> </key>
<value>
<list/>
</value>
</item>
<item>
<key> <string>center</string> </key>
<value>
<list/>
</value>
</item>
<item>
<key> <string>hidden</string> </key>
<value>
<list/>
</value>
</item>
<item>
<key> <string>left</string> </key>
<value>
<list>
<string>my_preferred_rest_api_token_server_url</string>
</list>
</value>
</item>
<item>
<key> <string>right</string> </key>
<value>
<list/>
</value>
</item>
</dictionary>
</value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>SystemPreference_viewSlapOSRestAPI</string> </value>
</item>
<item>
<key> <string>method</string> </key>
<value> <string>POST</string> </value>
</item>
<item>
<key> <string>name</string> </key>
<value> <string>SystemPreference_viewSlapOSRestAPI</string> </value>
</item>
<item>
<key> <string>pt</string> </key>
<value> <string>form_view</string> </value>
</item>
<item>
<key> <string>row_length</string> </key>
<value> <int>4</int> </value>
</item>
<item>
<key> <string>stored_encoding</string> </key>
<value> <string>UTF-8</string> </value>
</item>
<item>
<key> <string>title</string> </key>
<value> <string>SlapOS Rest API</string> </value>
</item>
<item>
<key> <string>unicode_mode</string> </key>
<value> <int>0</int> </value>
</item>
<item>
<key> <string>update_action</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>update_action_title</string> </key>
<value> <string></string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
master/bt5/slapos_rest_api/SkinTemplateItem/portal_skins/slapos_rest_api/SystemPreference_viewSlapOSRestAPI/my_preferred_rest_api_token_server_url.xml deleted 100644 → 0
View file @ 78407bb9
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="ProxyField" module="Products.ERP5Form.ProxyField"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>delegated_list</string> </key>
<value>
<list>
<string>items</string>
<string>title</string>
</list>
</value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>my_preferred_rest_api_token_server_url</string> </value>
</item>
<item>
<key> <string>message_values</string> </key>
<value>
<dictionary>
<item>
<key> <string>external_validator_failed</string> </key>
<value> <string>The input failed the external validator.</string> </value>
</item>
</dictionary>
</value>
</item>
<item>
<key> <string>overrides</string> </key>
<value>
<dictionary>
<item>
<key> <string>field_id</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>form_id</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>target</string> </key>
<value> <string></string> </value>
</item>
</dictionary>
</value>
</item>
<item>
<key> <string>tales</string> </key>
<value>
<dictionary>
<item>
<key> <string>field_id</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>form_id</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>items</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAI=</string> </persistent>
</value>
</item>
<item>
<key> <string>target</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>title</string> </key>
<value> <string></string> </value>
</item>
</dictionary>
</value>
</item>
<item>
<key> <string>values</string> </key>
<value>
<dictionary>
<item>
<key> <string>field_id</string> </key>
<value> <string>my_reference</string> </value>
</item>
<item>
<key> <string>form_id</string> </key>
<value> <string>Base_viewFieldLibrary</string> </value>
</item>
<item>
<key> <string>items</string> </key>
<value>
<list/>
</value>
</item>
<item>
<key> <string>target</string> </key>
<value> <string>Click to edit the target</string> </value>
</item>
<item>
<key> <string>title</string> </key>
<value> <string>Preferred Token Server URL</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</pickle>
</record>
<record id="2" aka="AAAAAAAAAAI=">
<pickle>
<global name="TALESMethod" module="Products.Formulator.TALESField"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>_text</string> </key>
<value> <string>python: [(\'\', \'\')] + [(x.getTitle(), x.getRelativeUrl()) for x in here.portal_catalog(portal_type=\'Service\', sort_on=((\'title\', \'ASC\'),),checked_permission=\'View\')]</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
master/bt5/slapos_rest_api/TestTemplateItem/portal_components/test.erp5.testSlapOSRestAPIV1.py deleted 100644 → 0
View file @ 78407bb9
# Copyright (c) 2002-2012 Nexedi SA and Contributors. All Rights Reserved.
from Products.SlapOS.tests.testSlapOSMixin import \
testSlapOSMixin
from Products.ERP5Type.Base import WorkflowMethod
import transaction
import httplib
import urllib
import urlparse
import json
import tempfile
import os
from App.Common import rfc1123_date
from DateTime import DateTime
import time
from unittest import skip
def _getMemcachedDict(self):
return self.getPortal().portal_memcached.getMemcachedDict(
key_prefix='slap_tool',
plugin_path='portal_memcached/default_memcached_plugin')
def _logAccess(self, user_reference, context_reference, text):
memcached_dict = self._getMemcachedDict()
value = json.dumps({
'user': '%s' % user_reference,
'created_at': '%s' % rfc1123_date(DateTime()),
'text': '%s' % text,
})
memcached_dict[context_reference] = value
class Simulator:
def __init__(self, outfile, method):
self.outfile = outfile
self.method = method
def __call__(self, *args, **kwargs):
"""Simulation Method"""
old = open(self.outfile, 'r').read()
if old:
l = eval(old)
else:
l = []
l.append({'recmethod': self.method,
'recargs': args,
'reckwargs': kwargs})
open(self.outfile, 'w').write(repr(l))
class RaisingSimulator:
def __init__(self, exception):
self.exception = exception
def __call__(self, *args, **kwargs):
"""Simulation Method"""
raise self.exception
class CustomHeaderHTTPConnection(httplib.HTTPConnection):
def __init__(self, custom_header, *args, **kwargs):
self._custom_header = custom_header
httplib.HTTPConnection.__init__(self, *args, **kwargs)
def request(self, *args, **kwargs):
headers = kwargs.get('headers', {})
headers.update(self._custom_header)
kwargs['headers'] = headers
return httplib.HTTPConnection.request(self, *args, **kwargs)
def SlapOSRestAPIV1MixinBase_afterSetUp(self):
self.test_random_id = self.generateNewId()
self.access_control_allow_headers = 'some, funny, headers, ' \
'always, expected, %s' % self.test_random_id
self.document_list = []
self.portal = self.getPortalObject()
self.api_url = self.portal.portal_slapos_rest_api.v1.getAPIRoot()
self.api_scheme, self.api_netloc, self.api_path, self.api_query, \
self.api_fragment = urlparse.urlsplit(self.api_url)
self.connection = CustomHeaderHTTPConnection(host=self.api_netloc,
custom_header={
'Access-Control-Request-Headers': self.access_control_allow_headers,
'Content-Type': 'application/json',
})
class SlapOSRestAPIV1MixinBase(testSlapOSMixin):
def generateNewId(self):
return str(self.getPortalObject().portal_ids.generateNewId(
id_group=('slapos_rest_api_v1_test')))
def cloneByPath(self, path):
return self.portal.restrictedTraverse(path).Base_createCloneDocument(
batch_mode=1)
def assertCacheControlHeader(self):
self.assertEqual('must-revalidate',
self.response.getheader('Cache-Control'))
def _afterSetUp(self):
super(SlapOSRestAPIV1MixinBase, self).afterSetUp()
def afterSetUp(self):
self._afterSetUp()
SlapOSRestAPIV1MixinBase_afterSetUp(self)
def beforeTearDown(self):
pass
def prepareResponse(self):
self.response = self.connection.getresponse()
self.response_data = self.response.read()
def assertBasicResponse(self):
self.assertEqual(self.response.getheader('access-control-allow-origin'),
'*')
self.assertEqual(self.response.getheader('access-control-allow-methods'),
'DELETE, PUT, POST, GET, OPTIONS')
self.assertEqual(self.response.getheader('access-control-allow-headers'),
self.access_control_allow_headers)
def assertResponseCode(self, code):
self.assertEqual(self.response.status, code,
'%s was expected, but got %s with response:\n%s' %
(code, self.response.status, self.response_data))
def assertResponseJson(self):
self.assertEqual(self.response.getheader('Content-Type'), 'application/json')
self.json_response = json.loads(self.response_data)
def assertResponseNoContentType(self):
self.assertEqual(self.response.getheader('Content-Type'), None)
def SlapOSRestAPIV1Mixin_afterSetUp(self):
SlapOSRestAPIV1MixinBase_afterSetUp(self)
self.person_request_simulator = tempfile.mkstemp()[1]
self.customer, self.customer_reference = self.createPerson()
self.customer.requestSoftwareInstance = Simulator(self.person_request_simulator,
'requestSoftwareInstance')
transaction.commit()
def SlapOSRestAPIV1Mixin_beforeTearDown(self):
if os.path.exists(self.person_request_simulator):
os.unlink(self.person_request_simulator)
class SlapOSRestAPIV1Mixin(SlapOSRestAPIV1MixinBase):
def createPerson(self):
customer = self.cloneByPath('person_module/template_member')
customer_reference = 'P' + self.generateNewId()
customer.edit(
reference=customer_reference,
default_email_url_string=customer_reference+'@example.com')
customer.validate()
for assignment in customer.contentValues(portal_type='Assignment'):
assignment.open()
customer.manage_setLocalRoles(customer.getReference(),
['Associate'])
transaction.commit()
customer.recursiveImmediateReindexObject()
transaction.commit()
return customer, customer_reference
def afterSetUp(self):
self._afterSetUp()
SlapOSRestAPIV1Mixin_afterSetUp(self)
def beforeTearDown(self):
SlapOSRestAPIV1Mixin_beforeTearDown(self)
def assertPersonRequestSimulatorEmpty(self):
self.assertEqual(open(self.person_request_simulator).read(), '')
def assertPersonRequestSimulator(self, args, kwargs):
stored = eval(open(self.person_request_simulator).read())
# do the same translation magic as in tool
kwargs = kwargs.copy()
for k_j, k_i in (
('software_release', 'software_release'),
('title', 'software_title'),
('software_type', 'software_type'),
('parameter', 'instance_xml'),
('sla', 'sla_xml'),
('slave', 'shared'),
('status', 'state')
):
kwargs[k_i] = kwargs.pop(k_j)
self.assertEqual(stored,
[{'recargs': args, 'reckwargs': kwargs,
'recmethod': 'requestSoftwareInstance'}])
reckwargs = stored[0]['reckwargs']
self.assertEqual(
set([
type(reckwargs['software_title']), type(reckwargs['software_release']),
type(reckwargs['software_type']), type(reckwargs['state']),
type(reckwargs['instance_xml']), type(reckwargs['sla_xml'])
]),
set([str])
)
@skip('Undecided.')
class TestInstanceRequest(SlapOSRestAPIV1Mixin):
def test_not_logged_in(self):
self.connection.request(method='POST',
url='/'.join([self.api_path, 'instance']))
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(401)
self.assertTrue(self.response.getheader('Location') is not None)
auth = self.response.getheader('WWW-Authenticate')
self.assertTrue(auth is not None)
self.assertTrue('Bearer realm="' in auth)
self.assertPersonRequestSimulatorEmpty()
def test_no_json(self):
self.connection.request(method='POST',
url='/'.join([self.api_path, 'instance']),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(400)
self.assertResponseJson()
self.assertEqual({'error': "Data is not json object."}, self.json_response)
self.assertPersonRequestSimulatorEmpty()
def test_bad_json(self):
self.connection.request(method='POST',
url='/'.join([self.api_path, 'instance']),
body='This is not JSON',
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(400)
self.assertResponseJson()
self.assertEqual({'error': "Data is not json object."}, self.json_response)
self.assertPersonRequestSimulatorEmpty()
def test_empty_json(self):
self.connection.request(method='POST',
url='/'.join([self.api_path, 'instance']),
body='{}',
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(400)
self.assertResponseJson()
self.assertEqual({
"status": "Missing.",
"slave": "Missing.",
"title": "Missing.",
"software_release": "Missing.",
"software_type": "Missing.",
"parameter": "Missing.",
"sla": "Missing."},
self.json_response)
self.assertPersonRequestSimulatorEmpty()
def test_status_slave_missing_json(self):
self.connection.request(method='POST',
url='/'.join([self.api_path, 'instance']),
body="""
{
"title": "My unique instance",
"software_release": "http://example.com/example.cfg",
"software_type": "type_provided_by_the_software",
"parameter": {
"Custom1": "one string",
"Custom2": "one float",
"Custom3": [
"abc",
"def"
]
},
"sla": {
"computer_id": "COMP-0"
}
}""",
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(400)
self.assertResponseJson()
self.assertEqual({
"status": "Missing.",
"slave": "Missing."
},
self.json_response)
self.assertPersonRequestSimulatorEmpty()
def test_slave_not_bool(self):
kwargs = {
'parameter': {
'Custom1': 'one string',
'Custom2': 'one float',
'Custom3': ['abc', 'def']},
'title': 'My unique instance',
'software_release': 'http://example.com/example.cfg',
'status': 'started',
'sla': {
'computer_id': 'COMP-0'},
'software_type': 'type_provided_by_the_software',
'slave': "True"}
self.connection.request(method='POST',
url='/'.join([self.api_path, 'instance']),
body=json.dumps(kwargs),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(400)
self.assertResponseJson()
self.assertEqual({
"slave": "unicode is not bool.",
},
self.json_response)
self.assertPersonRequestSimulatorEmpty()
def test_incorrect_status(self):
kwargs = {
'parameter': {
'Custom1': 'one string',
'Custom2': 'one float',
'Custom3': ['abc', 'def']},
'title': 'My unique instance',
'software_release': 'http://example.com/example.cfg',
'status': 'badstatus',
'sla': {
'computer_id': 'COMP-0'},
'software_type': 'type_provided_by_the_software',
'slave': True}
self.connection.request(method='POST',
url='/'.join([self.api_path, 'instance']),
body=json.dumps(kwargs),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(400)
self.assertResponseJson()
self.assertEqual({
"status": "Status shall be one of: started, stopped, destroyed.",
},
self.json_response)
self.assertPersonRequestSimulatorEmpty()
def test_correct(self):
kwargs = {
'parameter': {
'Custom1': 'one string',
'Custom2': 'one float',
'Custom3': ['abc', 'def']},
'title': 'My unique instance',
'software_release': 'http://example.com/example.cfg',
'status': 'started',
'sla': {
'computer_id': 'COMP-0'},
'software_type': 'type_provided_by_the_software',
'slave': True}
self.connection.request(method='POST',
url='/'.join([self.api_path, 'instance']),
body=json.dumps(kwargs),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(202)
self.assertResponseJson()
kwargs['parameter'] = '<?xml version=\'1.0\' encoding=\'utf-8\'?>\n<i'\
'nstance>\n <parameter id="Custom1">one string</parameter>\n <paramet'\
'er id="Custom2">one float</parameter>\n <parameter id="Custom3">[u\'a'\
'bc\', u\'def\']</parameter>\n</instance>\n'
kwargs['sla'] = '<?xml version=\'1.0\' encoding=\'utf-8\'?>\n<instanc'\
'e>\n <parameter id="computer_id">COMP-0</parameter>\n</instance>\n'
self.assertPersonRequestSimulator((), kwargs)
self.assertEqual({
"status": "processing",
},
self.json_response)
def test_additional_key_json(self):
kw_request = {
'parameter': {
'Custom1': 'one string',
'Custom2': 'one float',
'Custom3': ['abc', 'def']},
'title': 'My unique instance',
'software_release': 'http://example.com/example.cfg',
'status': 'started',
'sla': {
'computer_id': 'COMP-0'},
'software_type': 'type_provided_by_the_software',
'slave': True}
kwargs = kw_request.copy()
kwargs.update(**{'wrong_key': 'Be ignored'})
self.connection.request(method='POST',
url='/'.join([self.api_path, 'instance']),
body=json.dumps(kwargs),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(202)
self.assertResponseJson()
kw_request['parameter'] = '<?xml version=\'1.0\' encoding=\'utf-8\'?>\n<i'\
'nstance>\n <parameter id="Custom1">one string</parameter>\n <paramet'\
'er id="Custom2">one float</parameter>\n <parameter id="Custom3">[u\'a'\
'bc\', u\'def\']</parameter>\n</instance>\n'
kw_request['sla'] = '<?xml version=\'1.0\' encoding=\'utf-8\'?>\n<instanc'\
'e>\n <parameter id="computer_id">COMP-0</parameter>\n</instance>\n'
self.assertPersonRequestSimulator((), kw_request)
self.assertEqual({
"status": "processing",
},
self.json_response)
def test_correct_server_side_raise(self):
self.customer.requestSoftwareInstance = \
RaisingSimulator(AttributeError)
transaction.commit()
kwargs = {
'parameter': {
'Custom1': 'one string',
'Custom2': 'one float',
'Custom3': ['abc', 'def']},
'title': 'My unique instance',
'software_release': 'http://example.com/example.cfg',
'status': 'started',
'sla': {
'computer_id': 'COMP-0'},
'software_type': 'type_provided_by_the_software',
'slave': True}
self.connection.request(method='POST',
url='/'.join([self.api_path, 'instance']),
body=json.dumps(kwargs),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(500)
self.assertResponseJson()
self.assertEqual({
"error": "There is system issue, please try again later.",
},
self.json_response)
self.assertPersonRequestSimulatorEmpty()
def test_content_negotiation_headers(self):
self.connection = CustomHeaderHTTPConnection(host=self.api_netloc,
custom_header={
'Access-Control-Request-Headers': self.access_control_allow_headers
})
kwargs = {
'parameter': {
'Custom1': 'one string',
'Custom2': 'one float',
'Custom3': ['abc', 'def']},
'title': 'My unique instance',
'software_release': 'http://example.com/example.cfg',
'status': 'started',
'sla': {
'computer_id': 'COMP-0'},
'software_type': 'type_provided_by_the_software',
'slave': True}
self.connection.request(method='POST',
url='/'.join([self.api_path, 'instance']),
body=json.dumps(kwargs),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(400)
self.assertResponseJson()
self.assertEqual({
'Content-Type': "Header with value '^application/json.*' is required."},
self.json_response)
self.assertPersonRequestSimulatorEmpty()
# now check with incorrect headers
self.connection.request(method='POST',
url='/'.join([self.api_path, 'instance']),
body=json.dumps(kwargs),
headers={'REMOTE_USER': self.customer_reference,
'Content-Type': 'please/complain',
'Accept': 'be/silent'})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(400)
self.assertResponseJson()
self.assertEqual({
'Content-Type': "Header with value '^application/json.*' is required."},
self.json_response)
self.assertPersonRequestSimulatorEmpty()
# and with correct ones are set by default
@skip('Undecided.')
class TestInstanceOPTIONS(SlapOSRestAPIV1Mixin):
def test_OPTIONS_not_logged_in(self):
self.connection = CustomHeaderHTTPConnection(host=self.api_netloc,
custom_header={
'Access-Control-Request-Headers': self.access_control_allow_headers
})
self.connection.request(method='OPTIONS',
url='/'.join([self.api_path, 'instance']))
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(204)
self.assertResponseNoContentType()
self.assertPersonRequestSimulatorEmpty()
def SlapOSRestAPIV1InstanceMixin_afterSetUp(self):
SlapOSRestAPIV1Mixin_afterSetUp(self)
self.software_instance = self.createSoftwareInstance(self.customer)
class SlapOSRestAPIV1InstanceMixin(SlapOSRestAPIV1Mixin):
def afterSetUp(self):
self._afterSetUp()
SlapOSRestAPIV1InstanceMixin_afterSetUp(self)
def assertLastModifiedHeader(self):
calculated = rfc1123_date(self.software_instance.getModificationDate())
self.assertEqual(calculated, self.response.getheader('Last-Modified'))
def createSoftwareInstance(self, person):
software_instance = self.cloneByPath(
'software_instance_module/template_software_instance')
hosting_subscription = self.cloneByPath(
'hosting_subscription_module/template_hosting_subscription')
software_instance.edit(
reference='SI' + self.test_random_id,
ssl_key='SSL Key',
ssl_certificate='SSL Certificate',
url_string='http://url.of.software.release/'
)
software_instance.validate()
hosting_subscription.edit(
reference='HS' + self.test_random_id,
predecessor=software_instance.getRelativeUrl(),
destination_section=person.getRelativeUrl()
)
hosting_subscription.validate()
hosting_subscription.manage_setLocalRoles(person.getReference(),
['Assignee'])
software_instance.manage_setLocalRoles(person.getReference(),
['Assignee'])
transaction.commit()
hosting_subscription.recursiveImmediateReindexObject()
software_instance.recursiveImmediateReindexObject()
transaction.commit()
return software_instance
# needed to avoid calling interaction and being able to destroy XML
@WorkflowMethod.disable
def _destroySoftwareInstanceTextContentXml(self, software_instance):
software_instance.setTextContent('This is bad XML')
transaction.commit()
software_instance.recursiveImmediateReindexObject()
transaction.commit()
@skip('Undecided.')
class TestInstanceGET(SlapOSRestAPIV1InstanceMixin):
def test_non_existing(self):
non_existing = 'software_instance_module/' + self.generateNewId()
try:
self.portal.restrictedTraverse(non_existing)
except KeyError:
pass
else:
raise AssertionError('It was impossible to test')
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance',
non_existing]),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(404)
def test_something_else(self):
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance',
self.software_instance.getPredecessorRelatedValue().getRelativeUrl()]),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(404)
def test_bad_xml(self):
self._destroySoftwareInstanceTextContentXml(self.software_instance)
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance',
self.software_instance.getRelativeUrl()]),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(500)
self.assertResponseJson()
self.assertEqual({
"error": "There is system issue, please try again later."},
self.json_response)
def test(self):
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance',
self.software_instance.getRelativeUrl()]),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(200)
self.assertLastModifiedHeader()
self.assertCacheControlHeader()
self.assertResponseJson()
self.assertEqual({
"status": "draft",
"connection": {
"parameter1": "valueof1",
"parameter2": "https://niut:pass@example.com:4567/arfarf/oink?m=1#4.5"},
"partition": {
"public_ip": [],
"tap_interface": "",
"private_ip": []},
"slave": False,
"children_list": [],
"title": "Template Software Instance",
"software_type": "RootSoftwareInstance",
"parameter": {
"parameter1": "valueof1",
"parameter2": "valueof2"},
"software_release": "http://url.of.software.release/",
"sla": {"computer_guid": "SOMECOMP"}},
self.json_response)
def test_if_modified_since_equal(self):
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance',
self.software_instance.getRelativeUrl()]),
headers={'REMOTE_USER': self.customer_reference,
'If-Modified-Since': rfc1123_date(self.software_instance\
.getModificationDate())})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(304)
def test_if_modified_since_after(self):
# wait three seconds before continuing in order to not hit time precision
# issue, as test needs to provide date with second precision after
# last modification of software instance and *before* now()
time.sleep(3)
# add 2 seconds, as used rfc1123_date will ceil the second in response and
# one more second will be required in order to be *after* the modification time
if_modified = self.software_instance.getModificationDate().timeTime() + 2
# check the test: is calculated time *before* now?
self.assertTrue(int(if_modified) < int(DateTime().timeTime()))
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance',
self.software_instance.getRelativeUrl()]),
headers={'REMOTE_USER': self.customer_reference,
'If-Modified-Since': rfc1123_date(DateTime(if_modified))})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(304)
def test_if_modified_since_before(self):
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance',
self.software_instance.getRelativeUrl()]),
headers={'REMOTE_USER': self.customer_reference,
'If-Modified-Since': rfc1123_date(self.software_instance\
.getModificationDate() - 1)})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(200)
self.assertLastModifiedHeader()
self.assertCacheControlHeader()
self.assertResponseJson()
self.assertEqual({
"status": "draft",
"connection": {
"parameter1": "valueof1",
"parameter2": "https://niut:pass@example.com:4567/arfarf/oink?m=1#4.5"},
"partition": {
"public_ip": [],
"tap_interface": "",
"private_ip": []},
"slave": False,
"children_list": [],
"title": "Template Software Instance",
"software_type": "RootSoftwareInstance",
"parameter": {
"parameter1": "valueof1",
"parameter2": "valueof2"},
"software_release": "http://url.of.software.release/",
"sla": {"computer_guid": "SOMECOMP"}},
self.json_response)
def test_if_modified_since_date_not_date(self):
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance',
self.software_instance.getRelativeUrl()]),
headers={'REMOTE_USER': self.customer_reference,
'If-Modified-Since': 'This Is Not A date'})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(200)
self.assertLastModifiedHeader()
self.assertCacheControlHeader()
self.assertResponseJson()
self.assertEqual({
"status": "draft",
"connection": {
"parameter1": "valueof1",
"parameter2": "https://niut:pass@example.com:4567/arfarf/oink?m=1#4.5"},
"partition": {
"public_ip": [],
"tap_interface": "",
"private_ip": []},
"slave": False,
"children_list": [],
"title": "Template Software Instance",
"software_type": "RootSoftwareInstance",
"parameter": {
"parameter1": "valueof1",
"parameter2": "valueof2"},
"software_release": "http://url.of.software.release/",
"sla": {"computer_guid": "SOMECOMP"}},
self.json_response)
def test_if_modified_since_date_future(self):
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance',
self.software_instance.getRelativeUrl()]),
headers={'REMOTE_USER': self.customer_reference,
'If-Modified-Since': rfc1123_date(DateTime() + 1)})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(200)
self.assertLastModifiedHeader()
self.assertCacheControlHeader()
self.assertResponseJson()
self.assertEqual({
"status": "draft",
"connection": {
"parameter1": "valueof1",
"parameter2": "https://niut:pass@example.com:4567/arfarf/oink?m=1#4.5"},
"partition": {
"public_ip": [],
"tap_interface": "",
"private_ip": []},
"slave": False,
"children_list": [],
"title": "Template Software Instance",
"software_type": "RootSoftwareInstance",
"parameter": {
"parameter1": "valueof1",
"parameter2": "valueof2"},
"software_release": "http://url.of.software.release/",
"sla": {"computer_guid": "SOMECOMP"}},
self.json_response)
def test_other_one(self):
person, person_reference = self.createPerson()
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance',
self.software_instance.getRelativeUrl()]),
headers={'REMOTE_USER': person_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(404)
@skip('Undecided.')
class TestInstanceGETcertificate(SlapOSRestAPIV1InstanceMixin):
def test(self):
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance',
self.software_instance.getRelativeUrl(), 'certificate']),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(200)
self.assertLastModifiedHeader()
self.assertCacheControlHeader()
self.assertResponseJson()
self.assertEqual({
"ssl_key": "SSL Key",
"ssl_certificate": "SSL Certificate"
},
self.json_response)
def test_bad_xml(self):
self._destroySoftwareInstanceTextContentXml(self.software_instance)
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance',
self.software_instance.getRelativeUrl(), 'certificate']),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(200)
self.assertResponseJson()
self.assertEqual({
"ssl_key": "SSL Key",
"ssl_certificate": "SSL Certificate"
},
self.json_response)
def test_non_existing(self):
non_existing = 'software_instance_module/' + self.generateNewId()
try:
self.portal.restrictedTraverse(non_existing)
except KeyError:
pass
else:
raise AssertionError('It was impossible to test')
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance',
non_existing, 'certificate']),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(404)
def test_something_else(self):
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance',
self.software_instance.getPredecessorRelatedValue().getRelativeUrl(),
'certificate']),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(404)
def test_other_one(self):
person, person_reference = self.createPerson()
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance',
self.software_instance.getRelativeUrl(), 'certificate']),
headers={'REMOTE_USER': person_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(404)
class TestInstanceAllocableGET(SlapOSRestAPIV1InstanceMixin):
def test_not_logged_in(self):
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance', 'request']))
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(401)
self.assertTrue(self.response.getheader('Location') is not None)
auth = self.response.getheader('WWW-Authenticate')
self.assertTrue(auth is not None)
self.assertTrue('Bearer realm="' in auth)
def test_empty_parameter(self):
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance', 'request']),
body='{}',
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(400)
self.assertResponseJson()
self.assertEqual({
"slave": "Missing.",
"software_release": "Missing.",
"software_type": "Missing.",
"sla": "Missing."},
self.json_response)
def test_bad_sla_json(self):
kwargs = {
'software_release': 'http://example.com/example.cfg',
'sla': 'This is not JSON',
'software_type': 'type_provided_by_the_software',
'slave': 'true'}
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance', 'request']) + \
'?%s' % urllib.urlencode(kwargs),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(400)
self.assertResponseJson()
self.assertEqual({'sla': "Malformed value."}, self.json_response)
def test_slave_not_bool(self):
kwargs = {
'software_release': 'http://example.com/example.cfg',
'sla': json.dumps({
'computer_id': 'COMP-0'}),
'software_type': 'type_provided_by_the_software',
'slave': 'this is not a JSON boolean'}
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance', 'request']) + \
'?%s' % urllib.urlencode(kwargs),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(400)
self.assertResponseJson()
self.assertEqual({
"slave": "Malformed value.",
},
self.json_response)
def test_correct(self):
kwargs = {
'software_release': 'http://example.com/example.cfg',
'sla': json.dumps({
'computer_id': 'COMP-0'}),
'software_type': 'type_provided_by_the_software',
'slave': 'true'}
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance', 'request']) + \
'?%s' % urllib.urlencode(kwargs),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(200)
self.assertResponseJson()
def test_additional_key_json(self):
kw_request = {
'software_release': 'http://example.com/example.cfg',
'sla': json.dumps({
'computer_id': 'COMP-0'}),
'software_type': 'type_provided_by_the_software',
'slave': 'true'}
kwargs = kw_request.copy()
kwargs.update(**{'wrong_key': 'Be ignored'})
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance', 'request']) + \
'?%s' % urllib.urlencode(kwargs),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(200)
self.assertResponseJson()
# def test_correct_server_side_raise(self):
# self.customer.requestSoftwareInstance = \
# RaisingSimulator(AttributeError)
# transaction.commit()
# kwargs = {
# 'parameter': {
# 'Custom1': 'one string',
# 'Custom2': 'one float',
# 'Custom3': ['abc', 'def']},
# 'title': 'My unique instance',
# 'software_release': 'http://example.com/example.cfg',
# 'status': 'started',
# 'sla': {
# 'computer_id': 'COMP-0'},
# 'software_type': 'type_provided_by_the_software',
# 'slave': True}
# self.connection.request(method='GET',
# url='/'.join([self.api_path, 'instance', 'request']),
# body=json.dumps(kwargs),
# headers={'REMOTE_USER': self.customer_reference})
# self.prepareResponse()
# self.assertBasicResponse()
# self.assertResponseCode(500)
# self.assertResponseJson()
# self.assertEqual({
# "error": "There is system issue, please try again later.",
# },
# self.json_response)
# def test_content_negotiation_headers(self):
# self.connection = CustomHeaderHTTPConnection(host=self.api_netloc,
# custom_header={
# 'Access-Control-Request-Headers': self.access_control_allow_headers
# })
# kwargs = {
# 'parameter': {
# 'Custom1': 'one string',
# 'Custom2': 'one float',
# 'Custom3': ['abc', 'def']},
# 'title': 'My unique instance',
# 'software_release': 'http://example.com/example.cfg',
# 'status': 'started',
# 'sla': {
# 'computer_id': 'COMP-0'},
# 'software_type': 'type_provided_by_the_software',
# 'slave': True}
# self.connection.request(method='GET',
# url='/'.join([self.api_path, 'instance', 'request']),
# body=json.dumps(kwargs),
# headers={'REMOTE_USER': self.customer_reference})
# self.prepareResponse()
# self.assertBasicResponse()
# self.assertResponseCode(400)
# self.assertResponseJson()
# self.assertEqual({
# 'Content-Type': "Header with value '^application/json.*' is required."},
# self.json_response)
#
# # now check with incorrect headers
# self.connection.request(method='GET',
# url='/'.join([self.api_path, 'instance', 'request']),
# body=json.dumps(kwargs),
# headers={'REMOTE_USER': self.customer_reference,
# 'Content-Type': 'please/complain',
# 'Accept': 'be/silent'})
# self.prepareResponse()
# self.assertBasicResponse()
# self.assertResponseCode(400)
# self.assertResponseJson()
# self.assertEqual({
# 'Content-Type': "Header with value '^application/json.*' is required."},
# self.json_response)
# # and with correct ones are set by default
def SlapOSRestAPIV1BangMixin_afterSetUp(self):
SlapOSRestAPIV1BangMixin_afterSetUp(self)
self.instance_bang_simulator = tempfile.mkstemp()[1]
self.software_instance.bang = Simulator(
self.instance_bang_simulator, 'bang')
transaction.commit()
def SlapOSRestAPIV1BangMixin_beforeTearDown(self):
SlapOSRestAPIV1BangMixin_beforeTearDown()
if os.path.exists(self.instance_bang_simulator):
os.unlink(self.instance_bang_simulator)
class SlapOSRestAPIV1BangMixin(SlapOSRestAPIV1InstanceMixin):
def afterSetUp(self):
self._afterSetUp()
SlapOSRestAPIV1BangMixin_afterSetUp(self)
def beforeTearDown(self):
SlapOSRestAPIV1BangMixin_beforeTearDown(self)
def assertInstanceBangSimulatorEmpty(self):
self.assertEqual(open(self.instance_bang_simulator).read(), '')
def assertInstanceBangSimulator(self, args, kwargs):
stored = eval(open(self.instance_bang_simulator).read())
# do the same translation magic as in workflow
kwargs['comment'] = kwargs.pop('log')
self.assertEqual(stored,
[{'recargs': args, 'reckwargs': kwargs,
'recmethod': 'bang'}])
@skip('Undecided.')
class TestInstancePOSTbang(SlapOSRestAPIV1BangMixin):
def test(self):
kwargs = {'log': 'This is cool log!', 'bang_tree': True}
self.connection.request(method='POST',
url='/'.join([self.api_path, 'instance',
self.software_instance.getRelativeUrl(), 'bang']),
body=json.dumps(kwargs),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(204)
self.assertInstanceBangSimulator((), kwargs)
def test_server_side_raise(self):
self.software_instance.bang = RaisingSimulator(
AttributeError)
transaction.commit()
kwargs = {'log': 'This is cool log!'}
self.connection.request(method='POST',
url='/'.join([self.api_path, 'instance',
self.software_instance.getRelativeUrl(), 'bang']),
body=json.dumps(kwargs),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(500)
self.assertResponseJson()
self.assertEqual({
"error": "There is system issue, please try again later.",
},
self.json_response)
self.assertInstanceBangSimulatorEmpty()
def test_bad_xml(self):
self._destroySoftwareInstanceTextContentXml(self.software_instance)
kwargs = {'log': 'This is cool log!', 'bang_tree': True}
self.connection.request(method='POST',
url='/'.join([self.api_path, 'instance',
self.software_instance.getRelativeUrl(), 'bang']),
body=json.dumps(kwargs),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(204)
self.assertInstanceBangSimulator((), kwargs)
def test_non_existing(self):
non_existing = 'software_instance_module/' + self.generateNewId()
try:
self.portal.restrictedTraverse(non_existing)
except KeyError:
pass
else:
raise AssertionError('It was impossible to test')
kwargs = {'log': 'This is cool log!'}
self.connection.request(method='POST',
url='/'.join([self.api_path, 'instance',
non_existing, 'bang']),
body=json.dumps(kwargs),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(404)
self.assertInstanceBangSimulatorEmpty()
def test_something_else(self):
kwargs = {'log': 'This is cool log!'}
self.connection.request(method='POST',
url='/'.join([self.api_path, 'instance',
self.software_instance.getPredecessorRelatedValue().getRelativeUrl(),
'bang']), body=json.dumps(kwargs),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(404)
self.assertInstanceBangSimulatorEmpty()
def test_other_one(self):
kwargs = {'log': 'This is cool log!'}
person, person_reference = self.createPerson()
self.connection.request(method='POST',
url='/'.join([self.api_path, 'instance',
self.software_instance.getPredecessorRelatedValue().getRelativeUrl(),
'bang']), body=json.dumps(kwargs),
headers={'REMOTE_USER': person_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(404)
self.assertInstanceBangSimulatorEmpty()
def test_no_json(self):
self.connection.request(method='POST',
url='/'.join([self.api_path, 'instance',
self.software_instance.getRelativeUrl(), 'bang']),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(400)
self.assertResponseJson()
self.assertEqual({'error': "Data is not json object."}, self.json_response)
self.assertInstanceBangSimulatorEmpty()
def test_bad_json(self):
kwargs = {'wrong_key': 'This is cool log!'}
self.connection.request(method='POST',
url='/'.join([self.api_path, 'instance',
self.software_instance.getPredecessorRelatedValue().getRelativeUrl(),
'bang']), body=json.dumps(kwargs),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(400)
self.assertResponseJson()
self.assertEqual({'log': 'Missing.'}, self.json_response)
self.assertInstanceBangSimulatorEmpty()
def test_empty_json(self):
kwargs = {}
self.connection.request(method='POST',
url='/'.join([self.api_path, 'instance',
self.software_instance.getPredecessorRelatedValue().getRelativeUrl(),
'bang']), body=json.dumps(kwargs),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(400)
self.assertResponseJson()
self.assertEqual({'log': 'Missing.'}, self.json_response)
self.assertInstanceBangSimulatorEmpty()
def test_additional_key_json(self):
kw_log = {'log': 'This is cool log!', 'bang_tree': True}
kwargs = kw_log.copy()
kwargs.update(**{'wrong_key': 'Be ignored'})
self.connection.request(method='POST',
url='/'.join([self.api_path, 'instance',
self.software_instance.getRelativeUrl(),
'bang']), body=json.dumps(kwargs),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(204)
self.assertInstanceBangSimulator((), kw_log)
def test_log_not_string(self):
kwargs = {'log': True}
self.connection.request(method='POST',
url='/'.join([self.api_path, 'instance',
self.software_instance.getPredecessorRelatedValue().getRelativeUrl(),
'bang']), body=json.dumps(kwargs),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(400)
self.assertResponseJson()
self.assertEqual({'log': 'bool is not unicode.'}, self.json_response)
self.assertInstanceBangSimulatorEmpty()
@skip('Undecided.')
class TestInstancePUT(SlapOSRestAPIV1InstanceMixin):
def afterSetUp(self):
super(TestInstancePUT, self).afterSetUp()
self.instance_put_simulator = tempfile.mkstemp()[1]
self.software_instance.setTitle = Simulator(self.instance_put_simulator,
'setTitle')
self.software_instance.setConnectionXml = Simulator(self.instance_put_simulator,
'setConnectionXml')
transaction.commit()
def beforeTearDown(self):
super(TestInstancePUT, self).beforeTearDown()
if os.path.exists(self.instance_put_simulator):
os.unlink(self.instance_put_simulator)
def assertInstancePUTSimulator(self, l):
stored = eval(open(self.instance_put_simulator).read())
self.assertEqual(stored, l)
self.assertEqual(
set([type(q) for q in l[0]['recargs']]),
set([str])
)
def assertInstancePUTSimulatorEmpty(self):
self.assertEqual('', open(self.instance_put_simulator).read())
def test(self):
d = {
'title': 'New Title' + self.test_random_id,
'connection': {'url': 'http://new' + self.test_random_id}
}
self.connection.request(method='PUT',
url='/'.join([self.api_path, 'instance',
self.software_instance.getRelativeUrl()]),
body=json.dumps(d),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(200)
self.assertResponseJson()
self.assertEqual({
"title": "Modified.",
"connection": "Modified."
},
self.json_response)
self.assertInstancePUTSimulator([{'recargs': (d['title'],), 'reckwargs': {},
'recmethod': 'setTitle'},
{'recargs': ('<?xml version=\'1.0\' encoding=\'utf-8\'?>\n<instance>\n '\
'<parameter id="url">http://new%s</parameter>\n</instance>\n'%
self.test_random_id,), 'reckwargs': {}, 'recmethod': 'setConnectionXml'}])
def test_same_title(self):
d = {
'title': self.software_instance.getTitle(),
'connection': {'url': 'http://new' + self.test_random_id}
}
self.connection.request(method='PUT',
url='/'.join([self.api_path, 'instance',
self.software_instance.getRelativeUrl()]),
body=json.dumps(d),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(200)
self.assertResponseJson()
self.assertEqual({"connection": "Modified."}, self.json_response)
self.assertInstancePUTSimulator([
{'recargs': ('<?xml version=\'1.0\' encoding=\'utf-8\'?>\n<instance>\n '\
'<parameter id="url">http://new%s</parameter>\n</instance>\n'%
self.test_random_id,), 'reckwargs': {}, 'recmethod': 'setConnectionXml'}])
def test_same_connection(self):
d = {
'title': 'New Title 2' + self.test_random_id,
'connection': self.software_instance.getConnectionXmlAsDict()
}
self.connection.request(method='PUT',
url='/'.join([self.api_path, 'instance',
self.software_instance.getRelativeUrl()]),
body=json.dumps(d),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(200)
self.assertResponseJson()
self.assertEqual({
"title": "Modified.",
},
self.json_response)
self.assertInstancePUTSimulator([{'recargs': (d['title'],),
'reckwargs': {}, 'recmethod': 'setTitle'}])
def test_same_title_connection(self):
d = {
'title': self.software_instance.getTitle(),
'connection': self.software_instance.getConnectionXmlAsDict()
}
self.connection.request(method='PUT',
url='/'.join([self.api_path, 'instance',
self.software_instance.getRelativeUrl()]),
body=json.dumps(d),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(204)
self.assertInstancePUTSimulatorEmpty()
def test_not_logged_in(self):
self.connection.request(method='PUT',
url='/'.join([self.api_path, 'instance',
self.software_instance.getRelativeUrl()]))
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(401)
self.assertTrue(self.response.getheader('Location') is not None)
auth = self.response.getheader('WWW-Authenticate')
self.assertTrue(auth is not None)
self.assertTrue('Bearer realm="' in auth)
self.assertInstancePUTSimulatorEmpty()
def test_no_json(self):
self.connection.request(method='PUT',
url='/'.join([self.api_path, 'instance',
self.software_instance.getRelativeUrl()]),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(400)
self.assertResponseJson()
self.assertEqual({'error': "Data is not json object."}, self.json_response)
self.assertInstancePUTSimulatorEmpty()
def test_bad_json(self):
self.connection.request(method='PUT',
url='/'.join([self.api_path, 'instance',
self.software_instance.getRelativeUrl()]),
body='This is not JSON',
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(400)
self.assertResponseJson()
self.assertEqual({'error': "Data is not json object."}, self.json_response)
self.assertInstancePUTSimulatorEmpty()
def test_empty_json(self):
self.connection.request(method='PUT',
url='/'.join([self.api_path, 'instance',
self.software_instance.getRelativeUrl()]),
body='{}',
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(204)
self.assertInstancePUTSimulatorEmpty()
def test_future_compat(self):
self.connection.request(method='PUT',
url='/'.join([self.api_path, 'instance',
self.software_instance.getRelativeUrl()]),
body=json.dumps({'ignore':'me'}),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(204)
self.assertInstancePUTSimulatorEmpty()
@skip('Undecided.')
class TestInstanceGETlist(SlapOSRestAPIV1InstanceMixin):
def assertLastModifiedHeader(self):
calculated = rfc1123_date(self.portal.software_instance_module\
.bobobase_modification_time())
self.assertEqual(calculated, self.response.getheader('Last-Modified'))
def test_no_cache(self):
# version of test which ignores cache to expose possible other errors
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance']),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(200)
self.assertResponseJson()
self.assertEqual({
"list": ['/'.join([self.api_url, 'instance',
self.software_instance.getRelativeUrl()])]
},
self.json_response)
def test(self):
self.test_no_cache()
self.assertLastModifiedHeader()
self.assertCacheControlHeader()
def test_if_modified_since_equal(self):
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance']),
headers={'REMOTE_USER': self.customer_reference,
'If-Modified-Since': rfc1123_date(self.portal.software_instance_module\
.bobobase_modification_time())})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(304)
def test_if_modified_since_after(self):
# wait three seconds before continuing in order to not hit time precision
# issue, as test needs to provide date with second precision after
# last modification of software instance and *before* now()
time.sleep(3)
# add 2 seconds, as used rfc1123_date will ceil the second in response and
# one more second will be required in order to be *after* the modification time
if_modified = self.portal.software_instance_module\
.bobobase_modification_time().timeTime() + 2
# check the test: is calculated time *before* now?
self.assertTrue(int(if_modified) < int(DateTime().timeTime()))
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance']),
headers={'REMOTE_USER': self.customer_reference,
'If-Modified-Since': rfc1123_date(self.portal.software_instance_module\
.bobobase_modification_time())})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(304)
def test_if_modified_since_before(self):
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance']),
headers={'REMOTE_USER': self.customer_reference,
'If-Modified-Since': rfc1123_date(self.portal.software_instance_module\
.bobobase_modification_time() - 1)})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(200)
self.assertResponseJson()
self.assertLastModifiedHeader()
self.assertCacheControlHeader()
self.assertEqual({
"list": ['/'.join([self.api_url, 'instance',
self.software_instance.getRelativeUrl()])]
},
self.json_response)
def test_if_modified_since_date_not_date(self):
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance']),
headers={'REMOTE_USER': self.customer_reference,
'If-Modified-Since': 'This Is Not A date'})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(200)
self.assertResponseJson()
self.assertLastModifiedHeader()
self.assertCacheControlHeader()
self.assertEqual({
"list": ['/'.join([self.api_url, 'instance',
self.software_instance.getRelativeUrl()])]
},
self.json_response)
def test_if_modified_since_date_future(self):
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance']),
headers={'REMOTE_USER': self.customer_reference,
'If-Modified-Since': rfc1123_date(DateTime() + 1)})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(200)
self.assertResponseJson()
self.assertLastModifiedHeader()
self.assertCacheControlHeader()
self.assertEqual({
"list": ['/'.join([self.api_url, 'instance',
self.software_instance.getRelativeUrl()])]
},
self.json_response)
def test_another_one(self):
person, person_reference = self.createPerson()
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance']),
headers={'REMOTE_USER': person_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(204)
def test_empty(self):
self.portal.portal_catalog.unindexObject(
uid=self.software_instance.getUid())
self.software_instance.getParentValue().deleteContent(
self.software_instance.getId())
transaction.commit()
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance']),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(204)
def test_not_logged_in(self):
self.connection.request(method='GET',
url='/'.join([self.api_path, 'instance']))
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(401)
self.assertTrue(self.response.getheader('Location') is not None)
auth = self.response.getheader('WWW-Authenticate')
self.assertTrue(auth is not None)
self.assertTrue('Bearer realm="' in auth)
self.assertPersonRequestSimulatorEmpty()
@skip('Undecided.')
class TestComputerPUT(SlapOSRestAPIV1MixinBase):
def createComputer(self):
computer = self.cloneByPath(
'computer_module/template_computer')
computer.edit(reference='C' + self.test_random_id)
computer.validate()
computer.manage_setLocalRoles(computer.getReference(),
['Assignor'])
transaction.commit()
computer.recursiveImmediateReindexObject()
transaction.commit()
return computer
def afterSetUp(self):
super(TestComputerPUT, self).afterSetUp()
self.computer = self.createComputer()
self.computer_reference = self.computer.getReference()
self.computer_put_simulator = tempfile.mkstemp()[1]
self.computer.Computer_updateFromJson = Simulator(self.computer_put_simulator,
'Computer_updateFromJson')
transaction.commit()
def beforeTearDown(self):
super(TestComputerPUT, self).beforeTearDown()
if os.path.exists(self.computer_put_simulator):
os.unlink(self.computer_put_simulator)
def assertComputerPUTSimulator(self, l):
stored = eval(open(self.computer_put_simulator).read())
self.assertEqual(stored, l)
# check that method is called with strings, not unicodes
for l_ in l[0]['recargs'][0].itervalues():
for el in l_:
self.assertEqual(
set([type(q) for q in el.itervalues()]),
set([str])
)
def assertComputerPUTSimulatorEmpty(self):
self.assertEqual('', open(self.computer_put_simulator).read())
def test(self):
d = {
"partition": [
{
"title": "part0",
"public_ip": "::0",
"private_ip": "127.0.0.0",
"tap_interface": "tap0"
}
],
"software": [
{
"software_release": "software_release",
"status": "uninstalled",
"log": "Installation log"
}
]
}
self.connection.request(method='PUT',
url='/'.join([self.api_path, 'computer',
self.computer.getRelativeUrl()]),
body=json.dumps(d),
headers={'REMOTE_USER': self.computer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(204)
self.assertComputerPUTSimulator([
{'recmethod': 'Computer_updateFromJson',
'recargs': ({
'partition':
[{
'public_ip': '::0',
'tap_interface': 'tap0',
'private_ip': '127.0.0.0',
'title': 'part0'}],
'software':
[{
'status': 'uninstalled',
'software_release': 'software_release',
'log': 'Installation log'}]},),
'reckwargs': {}}])
def test_not_logged_in(self):
self.connection.request(method='PUT',
url='/'.join([self.api_path, 'computer',
self.computer.getRelativeUrl()]))
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(401)
self.assertTrue(self.response.getheader('Location') is not None)
auth = self.response.getheader('WWW-Authenticate')
self.assertTrue(auth is not None)
self.assertTrue('Bearer realm="' in auth)
self.assertComputerPUTSimulatorEmpty()
def test_no_json(self):
self.connection.request(method='PUT',
url='/'.join([self.api_path, 'computer',
self.computer.getRelativeUrl()]),
headers={'REMOTE_USER': self.computer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(400)
self.assertResponseJson()
self.assertEqual({'error': "Data is not json object."}, self.json_response)
self.assertComputerPUTSimulatorEmpty()
def test_bad_json(self):
self.connection.request(method='PUT',
url='/'.join([self.api_path, 'computer',
self.computer.getRelativeUrl()]),
body='This is not JSON',
headers={'REMOTE_USER': self.computer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(400)
self.assertResponseJson()
self.assertEqual({'error': "Data is not json object."}, self.json_response)
self.assertComputerPUTSimulatorEmpty()
def test_empty_json(self):
self.connection.request(method='PUT',
url='/'.join([self.api_path, 'computer',
self.computer.getRelativeUrl()]),
body='{}',
headers={'REMOTE_USER': self.computer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(204)
self.assertComputerPUTSimulatorEmpty()
def test_future_compat(self):
self.connection.request(method='PUT',
url='/'.join([self.api_path, 'computer',
self.computer.getRelativeUrl()]),
body=json.dumps({'ignore':'me'}),
headers={'REMOTE_USER': self.computer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(204)
self.assertComputerPUTSimulatorEmpty()
def test_software_release_status(self):
d = {
"partition": [
{
"title": "part0",
"public_ip": "::0",
"private_ip": "127.0.0.0",
"tap_interface": "tap0"
}
],
"software": [
{
"software_release": "software_release",
"status": "wrong status",
"log": "Installation log"
}
]
}
self.connection.request(method='PUT',
url='/'.join([self.api_path, 'computer',
self.computer.getRelativeUrl()]),
body=json.dumps(d),
headers={'REMOTE_USER': self.computer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(400)
self.assertResponseJson()
self.assertEqual({'software_0': ['Status "wrong status" is incorrect.']},
self.json_response)
self.assertComputerPUTSimulatorEmpty()
def test_only_partition(self):
d = {
"partition": [
{
"title": "part0",
"public_ip": "::0",
"private_ip": "127.0.0.0",
"tap_interface": "tap0"
}
]
}
self.connection.request(method='PUT',
url='/'.join([self.api_path, 'computer',
self.computer.getRelativeUrl()]),
body=json.dumps(d),
headers={'REMOTE_USER': self.computer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(204)
self.assertComputerPUTSimulator([
{'recmethod': 'Computer_updateFromJson',
'recargs': ({
'partition':
[{
'public_ip': '::0',
'tap_interface': 'tap0',
'private_ip': '127.0.0.0',
'title': 'part0'}],
},),
'reckwargs': {}}])
def test_only_software(self):
d = {
"software": [
{
"software_release": "software_release",
"status": "uninstalled",
"log": "Installation log"
}
]
}
self.connection.request(method='PUT',
url='/'.join([self.api_path, 'computer',
self.computer.getRelativeUrl()]),
body=json.dumps(d),
headers={'REMOTE_USER': self.computer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(204)
self.assertComputerPUTSimulator([
{'recmethod': 'Computer_updateFromJson',
'recargs': ({
'software':
[{
'status': 'uninstalled',
'software_release': 'software_release',
'log': 'Installation log'}]},),
'reckwargs': {}}])
def test_partition_object_incorrect(self):
d = {
"partition": [
{
"title": "part0",
"public_ip": "::0",
"private_ip": "127.0.0.0"
}
],
"software": [
{
"software_release": "software_release",
"status": "uninstalled",
"log": "Installation log"
}
]
}
self.connection.request(method='PUT',
url='/'.join([self.api_path, 'computer',
self.computer.getRelativeUrl()]),
body=json.dumps(d),
headers={'REMOTE_USER': self.computer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(400)
self.assertResponseJson()
self.assertEqual({'partition_0': ['Missing key "tap_interface".']},
self.json_response)
self.assertComputerPUTSimulatorEmpty()
def test_software_object_incorrect(self):
d = {
"partition": [
{
"title": "part0",
"public_ip": "::0",
"private_ip": "127.0.0.0",
"tap_interface": "tap0"
}
],
"software": [
{
"software_release": "software_release",
"status": "uninstalled",
}
]
}
self.connection.request(method='PUT',
url='/'.join([self.api_path, 'computer',
self.computer.getRelativeUrl()]),
body=json.dumps(d),
headers={'REMOTE_USER': self.computer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(400)
self.assertResponseJson()
self.assertEqual({'software_0': ['Missing key "log".']},
self.json_response)
self.assertComputerPUTSimulatorEmpty()
class TestStatusGET(SlapOSRestAPIV1InstanceMixin):
def afterSetUp(self):
self._afterSetUp()
SlapOSRestAPIV1Mixin_afterSetUp(self)
def createComputer(self):
computer = self.cloneByPath(
'computer_module/template_computer')
computer.edit(reference='C' + self.test_random_id)
computer.validate()
computer.manage_setLocalRoles(self.customer_reference,
['Assignee'])
transaction.commit()
computer.recursiveImmediateReindexObject()
transaction.commit()
return computer
def assertCacheControlHeader(self):
self.assertEqual('max-age=300, private',
self.response.getheader('Cache-Control'))
def test_non_existing_status(self):
non_existing = 'system_event_module/' + self.generateNewId()
try:
self.portal.restrictedTraverse(non_existing)
except KeyError:
pass
else:
raise AssertionError('It was impossible to test')
self.connection.request(method='GET',
url='/'.join([self.api_path, 'status',
non_existing]),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(404)
def test_something_else(self):
self.connection.request(method='GET',
url='/'.join([self.api_path, 'status',
self.customer.getRelativeUrl()]),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(404)
def _storeJson(self, key, json):
memcached_dict = self.getPortalObject().portal_memcached.\
getMemcachedDict(
key_prefix='slap_tool',
plugin_path='portal_memcached/default_memcached_plugin')
memcached_dict[key] = json
def test_on_computer(self):
self.computer = self.createComputer()
reference = self.computer.getReference()
value = json.dumps({'foo': reference})
self._storeJson(reference, value)
transaction.commit()
self.connection.request(method='GET',
url='/'.join([self.api_path, 'status',
self.computer.getRelativeUrl()]),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(200)
self.assertCacheControlHeader()
self.assertResponseJson()
value = json.loads(value)
value['@document'] = self.computer.getRelativeUrl()
self.assertEqual(value, self.json_response)
def test_on_instance(self):
self.software_instance = self.createSoftwareInstance(self.customer)
reference = self.software_instance.getReference()
value = json.dumps({'bar': reference})
self._storeJson(reference, value)
transaction.commit()
self.connection.request(method='GET',
url='/'.join([self.api_path, 'status',
self.software_instance.getRelativeUrl()]),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(200)
self.assertCacheControlHeader()
self.assertResponseJson()
value = json.loads(value)
value['@document'] = self.software_instance.getRelativeUrl()
self.assertEqual(value, self.json_response)
def test_no_data_in_memcached(self):
self.computer = self.createComputer()
self.connection.request(method='GET',
url='/'.join([self.api_path, 'status',
self.computer.getRelativeUrl()]),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(200)
self.assertCacheControlHeader()
self.assertResponseJson()
self.assertEquals(self.json_response['user'], 'SlapOS Master')
def test_search_no_status(self):
self.connection.request(method='GET',
url='/'.join([self.api_path, 'status']),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(204)
def test_search_existing_start_instance(self):
self.software_instance = self.createSoftwareInstance(self.customer)
self.software_instance.workflow_history[\
'instance_slap_interface_workflow'].append({
'comment':'Directly in Started state',
'error_message': '',
'actor': 'ERP5TypeTestCase',
'slap_state': 'start_requested',
'time': DateTime(),
'action': 'foo_transition'})
transaction.commit()
self.software_instance.recursiveImmediateReindexObject()
transaction.commit()
self.connection.request(method='GET',
url='/'.join([self.api_path, 'status']),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(200)
self.assertCacheControlHeader()
self.assertResponseJson()
self.assertEqual({
'list': ['/'.join([self.api_url, 'status',
self.software_instance.getRelativeUrl()])]
}, self.json_response)
def test_search_existing_stop_instance(self):
self.software_instance = self.createSoftwareInstance(self.customer)
self.software_instance.workflow_history[\
'instance_slap_interface_workflow'].append({
'comment':'Directly in stop state',
'error_message': '',
'actor': 'ERP5TypeTestCase',
'slap_state': 'stop_requested',
'time': DateTime(),
'action': 'foo_transition'})
transaction.commit()
self.software_instance.recursiveImmediateReindexObject()
transaction.commit()
self.connection.request(method='GET',
url='/'.join([self.api_path, 'status']),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(200)
self.assertCacheControlHeader()
self.assertResponseJson()
self.assertEqual({
'list': ['/'.join([self.api_url, 'status',
self.software_instance.getRelativeUrl()])]
}, self.json_response)
def test_check_no_destroyed_instance(self):
self.software_instance = self.createSoftwareInstance(self.customer)
self.software_instance.workflow_history[\
'instance_slap_interface_workflow'].append({
'comment':'Directly in destroyed state',
'error_message': '',
'actor': 'ERP5TypeTestCase',
'slap_state': 'destroy_requested',
'time': DateTime(),
'action': 'foo_transition'})
transaction.commit()
self.software_instance.recursiveImmediateReindexObject()
transaction.commit()
self.connection.request(method='GET',
url='/'.join([self.api_path, 'status']),
headers={'REMOTE_USER': self.customer_reference})
self.prepareResponse()
self.assertBasicResponse()
self.assertResponseCode(204)
master/bt5/slapos_rest_api/TestTemplateItem/portal_components/test.erp5.testSlapOSRestAPIV1.xml deleted 100644 → 0
View file @ 78407bb9
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="Test Component" module="erp5.portal_type"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>default_reference</string> </key>
<value> <string>testSlapOSRestAPIV1</string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>test.erp5.testSlapOSRestAPIV1</string> </value>
</item>
<item>
<key> <string>portal_type</string> </key>
<value> <string>Test Component</string> </value>
</item>
<item>
<key> <string>sid</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>text_content_error_message</string> </key>
<value>
<tuple>
<string>E:1013, 2: No value for argument \'self\' in function call (no-value-for-parameter)</string>
</tuple>
</value>
</item>
<item>
<key> <string>text_content_warning_message</string> </key>
<value>
<tuple>
<string>W: 41, 10: Use of eval (eval-used)</string>
<string>W:175, 13: Use of eval (eval-used)</string>
<string>W:753, 4: Unused variable \'person\' (unused-variable)</string>
<string>W:824, 4: Unused variable \'person\' (unused-variable)</string>
<string>W:1029, 13: Use of eval (eval-used)</string>
<string>W:1116, 4: Unused variable \'person\' (unused-variable)</string>
<string>W:1212, 13: Use of eval (eval-used)</string>
<string>W:1473, 4: Unused variable \'person\' (unused-variable)</string>
<string>W:1536, 13: Use of eval (eval-used)</string>
<string>W:1853, 28: Redefining name \'json\' from outer scope (line 9) (redefined-outer-name)</string>
</tuple>
</value>
</item>
<item>
<key> <string>version</string> </key>
<value> <string>erp5</string> </value>
</item>
<item>
<key> <string>workflow_history</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAI=</string> </persistent>
</value>
</item>
</dictionary>
</pickle>
</record>
<record id="2" aka="AAAAAAAAAAI=">
<pickle>
<global name="PersistentMapping" module="Persistence.mapping"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>data</string> </key>
<value>
<dictionary>
<item>
<key> <string>component_validation_workflow</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAM=</string> </persistent>
</value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</pickle>
</record>
<record id="3" aka="AAAAAAAAAAM=">
<pickle>
<global name="WorkflowHistoryList" module="Products.ERP5Type.patches.WorkflowTool"/>
</pickle>
<pickle>
<tuple>
<none/>
<list>
<dictionary>
<item>
<key> <string>action</string> </key>
<value> <string>validate</string> </value>
</item>
<item>
<key> <string>validation_state</string> </key>
<value> <string>validated</string> </value>
</item>
</dictionary>
</list>
</tuple>
</pickle>
</record>
</ZopeData>
master/bt5/slapos_rest_api/ToolTemplateItem/portal_slapos_rest_api.xml deleted 100644 → 0
View file @ 78407bb9
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="SlapOS Rest API Tool" module="erp5.portal_type"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>_Access_contents_information_Permission</string> </key>
<value>
<tuple>
<string>Member</string>
<string>Manager</string>
</tuple>
</value>
</item>
<item>
<key> <string>_Add_portal_content_Permission</string> </key>
<value>
<tuple>
<string>Manager</string>
</tuple>
</value>
</item>
<item>
<key> <string>_View_Permission</string> </key>
<value>
<tuple>
<string>Manager</string>
</tuple>
</value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>portal_slapos_rest_api</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
master/bt5/slapos_rest_api/bt/copyright_list deleted 100644 → 0
View file @ 78407bb9
2012 Nexedi S.A.
\ No newline at end of file
master/bt5/slapos_rest_api/bt/dependency_list deleted 100644 → 0
View file @ 78407bb9
slapos_cloud
slapos_rest_api_tool_portal_type
\ No newline at end of file
master/bt5/slapos_rest_api/bt/description deleted 100644 → 0
View file @ 78407bb9
Tool and shared code for all versions of API.
\ No newline at end of file
master/bt5/slapos_rest_api/bt/template_action_path_list deleted 100644 → 0
View file @ 78407bb9
System Preference | slapos_rest_api_preference
\ No newline at end of file
master/bt5/slapos_rest_api/bt/template_property_sheet_id_list deleted 100644 → 0
View file @ 78407bb9
SlapOSRestAPISystemPreference
\ No newline at end of file
master/bt5/slapos_rest_api/bt/template_skin_id_list deleted 100644 → 0
View file @ 78407bb9
slapos_rest_api
\ No newline at end of file
master/bt5/slapos_rest_api/bt/template_test_id_list deleted 100644 → 0
View file @ 78407bb9
test.erp5.testSlapOSRestAPIV1
\ No newline at end of file
master/bt5/slapos_rest_api/bt/template_tool_id_list deleted 100644 → 0
View file @ 78407bb9
portal_slapos_rest_api
\ No newline at end of file
master/bt5/slapos_rest_api/bt/title deleted 100644 → 0
View file @ 78407bb9
slapos_rest_api
\ No newline at end of file
master/bt5/slapos_rest_api_tool_portal_type/PortalTypeTemplateItem/portal_types/SlapOS%20Rest%20API%20Tool.xml deleted 100644 → 0
View file @ 78407bb9
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="Base Type" module="erp5.portal_type"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>content_icon</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>description</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>SlapOS Rest API Tool</string> </value>
</item>
<item>
<key> <string>init_script</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>permission</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>portal_type</string> </key>
<value> <string>Base Type</string> </value>
</item>
<item>
<key> <string>type_class</string> </key>
<value> <string>SlapOSRestAPITool</string> </value>
</item>
<item>
<key> <string>type_interface</string> </key>
<value>
<tuple/>
</value>
</item>
<item>
<key> <string>type_mixin</string> </key>
<value>
<tuple/>
</value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
master/bt5/slapos_rest_api_tool_portal_type/bt/description deleted 100644 → 0
View file @ 78407bb9
Workaround in order to have portal type in different Business Template then tool itself.
\ No newline at end of file
master/bt5/slapos_rest_api_tool_portal_type/bt/title deleted 100644 → 0
View file @ 78407bb9
slapos_rest_api_tool_portal_type
\ No newline at end of file
master/product/SlapOS/Tool/SlapOSRestAPITool.py deleted 100644 → 0
View file @ 78407bb9
# -*- coding: utf-8 -*-
##############################################################################
#
# Copyright (c) 2012 Nexedi SA and Contributors. All Rights Reserved.
# Łukasz Nowak <luke@nexedi.com>
# Romain Courteaud <romain@nexedi.com>
#
# 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 advised 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., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################
from Products.ERP5Type.Tool.BaseTool import BaseTool
from AccessControl import ClassSecurityInfo
from Products.ERP5Type.Globals import InitializeClass
from Products.ERP5Type import Permissions
from ComputedAttribute import ComputedAttribute
class SlapOSRestAPITool(BaseTool):
"""SlapOS REST API Tool
This is container for multiple versions of API.
"""
id = 'portal_slapos_rest_api'
meta_type = 'ERP5 SlapOS Rest API Tool'
portal_type = 'SlapOS Rest API Tool'
security = ClassSecurityInfo()
security.declareObjectProtected(Permissions.AccessContentsInformation)
allowed_types = ()
security.declarePublic('v1')
@ComputedAttribute
def v1(self):
"""API hooked"""
# XXX: It could be done better (more dynamic, configurable from UI)
from erp5.component.document.SlapOSRestAPIV1 import SlapOSRestAPIV1
return SlapOSRestAPIV1().__of__(self)
security.declarePrivate('manage_afterAdd')
def manage_afterAdd(self, item, container) :
"""Init permissions right after creation.
Permissions in slap tool are simple:
o Each member can access the tool.
o Only manager can view and create.
o Anonymous can not access
"""
item.manage_permission(Permissions.AddPortalContent,
['Manager'])
item.manage_permission(Permissions.AccessContentsInformation,
['Member', 'Manager'])
item.manage_permission(Permissions.View,
['Manager',])
BaseTool.inheritedAttribute('manage_afterAdd')(self, item, container)
InitializeClass(SlapOSRestAPITool)
master/tests/__init__.py
View file @ 794fca99
......@@ -11,7 +11,6 @@ slapos_bt_list = [
'slapos_cloud',
'slapos_erp5',
'slapos_pdm',
'slapos_rest_api',
'slapos_slap_tool',
'slapos_web',
'slapos_crm',
......
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