Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_types_put(self):
method = 'test_types_put'
url = '/api/assets/types'
self.logger.info('[%s] URL: %s', method, url)
json_dict = {'session_id': 'test', 'object': self.prepare_new_asset_type()}
new_asset_type_body = jsonpickle.encode(json_dict)
self.logger.info('JSON data: %s', new_asset_type_body)
rv = self.app.delete('/api/assets/types/name/%s?session_id=test' % quote(self.prepare_new_asset_type().theName))
rv = self.app.post(url, content_type='application/json', data=new_asset_type_body)
self.logger.debug('[%s] Response data: %s', method, rv.data)
json_resp = jsonpickle.decode(rv.data)
self.assertIsNotNone(json_resp, 'No results after deserialization')
type_id = json_resp.get('asset_type_id', None)
self.assertIsNotNone(type_id, 'No asset type ID returned')
self.assertGreater(type_id, 0, 'Invalid asset type ID returned [%d]' % type_id)
self.logger.info('[%s] Asset type ID: %d', method, type_id)
type_to_update = self.prepare_new_asset_type()
type_to_update.theName = 'Edited test asset type'
type_to_update.theId = type_id
json_dict = {'session_id': 'test', 'object': type_to_update}
upd_type_body = jsonpickle.encode(json_dict)
rv = self.app.put('/api/assets/types/name/%s?session_id=test' % quote(self.prepare_new_asset_type().theName), data=upd_type_body, content_type='application/json')
self.assertIsNotNone(rv.data, 'No response')
json_resp = jsonpickle.decode(rv.data)
self.assertIsNotNone(json_resp)
self.assertIsInstance(json_resp, dict)
def test_types_get(self):
method = 'test_types_get'
rv = self.app.get('/api/assets/types?session_id=test')
assets = jsonpickle.decode(rv.data)
self.assertIsNotNone(assets, 'No results after deserialization')
self.assertIsInstance(assets, list, 'The result is not a dictionary as expected')
self.assertGreater(len(assets), 0, 'No assets in the dictionary')
self.logger.info('[%s] Asset types found: %d', method, len(assets))
asset_type = assets[0]
self.logger.info('[%s] First asset types: %s [%d]\n', method, asset_type['theName'], asset_type['theId'])
def ParallelSimulation(modelname,packname,tstart,tstop,tolerance,stepsize,interval,events,dirname,config,resultpath,pickleobj,simulatorname,subdir,logfile):
##run the simuations in parallel using the multiprocessing module##
simulator=jsonpickle.decode(pickleobj)
try:
try:
import pythoncom
pythoncom.CoInitialize() # Initialize the COM library on the current thread
haveCOM = True
except:
pass
canLoadAllPackages = True
sp = packname[0].rsplit('.', 1)
if len(sp) > 1:
if not sp[1] in simulator.modelExtension:
canLoadAllPackages = False
else:
canLoadAllPackages = False
def message_handler(self, msg):
build_req = jsonpickle.decode(msg.body)
if not isinstance(build_req, BuildRequest):
self.message_chan.basic_ack(msg.delivery_tag)
return
if self.process:
logging.debug("Detected a running process")
self.state_table[self.state](msg, build_req)
#Conformamos la dirección:
url = "http://%s/" % modules.get_hostname(module=sbd)
#Añadimos el metodo al que queremos conectarnos.
url+="comprobarAccesoUsuario"
#Extraemos lo datos de la petición al endpoints y empaquetamos un dict.
datos = {
"username": formatTextInput(request.username),
"password": formatTextInput(request.password),
}
#Petición al microservicio:
result = urlfetch.fetch(url=url, payload=urllib.urlencode(datos), method=urlfetch.POST)
json = jsonpickle.decode(result.content)
if json=='Usuario no encontrado':
raise endpoints.NotFoundException('Usuario no encontrado')
else:
mensajeSalida=salidaLogin(idUser=str(json['idUsuario']), nombre=str(json['nombre']), rol=str(json['rol']))
#Info de seguimiento
if v:
print nombreMicroservicio
print ' Return: '+str(mensajeSalida)+'\n'
return mensajeSalida
def __init__(self):
super().__init__(self.docker_image_name, self.docker_image_tag)
with open(getJsonName(__file__, "jupyter_sleuth")) as f:
self.data = jsonpickle.decode(f.read())
f.close()
self.initVolumes()
self.inputConnections = ConnectionDict(self.inputConnectionsStore)
self.drawGUI()
def get_scenes(self, p_body):
l_msg = jsonpickle.decode(p_body)
LOG.debug('Got Scenes {}'.format(PrettyFormatAny.form(l_msg, 'Scenes', 190)))
def restore(self, data):
_, meta = self.pp.restore_pandas(data)
left = decode(meta['left'])
right = decode(meta['right'])
closed = str(meta['closed'])
obj = pd.Interval(left, right, closed=closed)
return obj
def restore(flat):
"""
Restore more complex data that Python's json can't handle (e.g. Numpy arrays).
Copied from sacred.serializer for performance reasons.
"""
return jsonpickle.decode(json.dumps(flat), keys=True)