test/prodstub/app/prodstub.py

   1
   2 #  ============LICENSE_START===============================================
   3 #  Copyright (C) 2020 Nordix Foundation. All rights reserved.
   4 #  ========================================================================
   5 #  Licensed under the Apache License, Version 2.0 (the "License");
   6 #  you may not use this file except in compliance with the License.
   7 #  You may obtain a copy of the License at
   8 #
   9 #       http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 #  Unless required by applicable law or agreed to in writing, software
  12 #  distributed under the License is distributed on an "AS IS" BASIS,
  13 #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 #  See the License for the specific language governing permissions and
  15 #  limitations under the License.
  16 #  ============LICENSE_END=================================================
  17 #
  18
  19 from flask import Flask
  20 from flask import request
  21
  22 import requests
  23
  24 import json
  25 from jsonschema import validate
  26
  27 import threading
  28 import time
  29 import datetime
  30 import logging
  31
  32 # Disable all logging of GET on reading counters and status
  33 class AjaxFilter(logging.Filter):
  34     def filter(self, record):
  35         return ("/counter/" not in record.getMessage()) and ("/status" not in record.getMessage())
  36
  37 log = logging.getLogger('werkzeug')
  38 log.addFilter(AjaxFilter())
  39
  40 app = Flask(__name__)
  41
  42 # # list of callback messages
  43 # msg_callbacks={}
  44
  45 # Server info
  46 HOST_IP = "::"
  47 HOST_PORT = 2222
  48
  49 # Request and response constants
  50 CALLBACK_CREATE_URL="/callbacks/job/<string:producer_id>"
  51 CALLBACK_DELETE_URL="/callbacks/job/<string:producer_id>/<string:job_id>"
  52 CALLBACK_SUPERVISION_URL="/callbacks/supervision/<string:producer_id>"
  53
  54 ARM_CREATE_RESPONSE="/arm/create/<string:producer_id>/<string:job_id>"
  55 ARM_DELETE_RESPONSE="/arm/delete/<string:producer_id>/<string:job_id>"
  56 ARM_SUPERVISION_RESPONSE="/arm/supervision/<string:producer_id>"
  57 ARM_TYPE="/arm/type/<string:producer_id>/<string:type_id>"
  58 COUNTER_SUPERVISION="/counter/supervision/<string:producer_id>"
  59 COUNTER_CREATE="/counter/create/<string:producer_id>/<string:job_id>"
  60 COUNTER_DELETE="/counter/delete/<string:producer_id>/<string:job_id>"
  61
  62 JOB_DATA="/jobdata/<string:producer_id>/<string:job_id>"
  63
  64 STATUS="/status"
  65
  66 #Constants
  67 APPL_JSON='application/json'
  68 UNKNOWN_QUERY_PARAMETERS="Unknown query parameter(s)"
  69 RETURNING_CONFIGURED_RESP="returning configured response code"
  70 JOBID_NO_MATCH="job id in stored json does not match request"
  71 PRODUCER_OR_JOB_NOT_FOUND="producer or job not found"
  72 PRODUCER_NOT_FOUND="producer not found"
  73 TYPE_NOT_FOUND="type not found"
  74 TYPE_IN_USE="type is in use in a job"
  75 JOB_NOT_FOUND="job not found"
  76 JOB_DATA_NOT_FOUND="job data not found"
  77 JSON_CORRUPT="json in request is corrupt or missing"
  78
  79 #Producer and job db, including armed responses
  80 db={}
  81 # producer
  82 #  armed response for supervision
  83 #  armed types
  84 #  supervision counter
  85 #  job
  86 #    job json
  87 #    target_type
  88 #    armed response for create
  89 #    armed response for delete
  90 #    create counter
  91 #    delete counter
  92 #    delivering status
  93
  94 # disable warning about unverified https requests
  95 from requests.packages import urllib3
  96
  97 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
  98
  99 # Helper function to populate a callback dict with the basic structure
 100 # if job_id is None then only the producer level is setup and the producer dict is returned
 101 # if job_id is not None, the job level is setup and the job dict is returned (producer must exist)
 102 def setup_callback_dict(producer_id, job_id):
 103
 104     producer_dict=None
 105     if (producer_id in db.keys()):
 106         producer_dict=db[producer_id]
 107     else:
 108         if (job_id is not None):
 109             return None
 110         producer_dict={}
 111         db[producer_id]=producer_dict
 112
 113         producer_dict['supervision_response']=200
 114         producer_dict['supervision_counter']=0
 115         producer_dict['types']=[]
 116
 117     if (job_id is None):
 118         return producer_dict
 119
 120     job_dict=None
 121     if (job_id in producer_dict.keys()):
 122         job_dict=producer_dict[job_id]
 123     else:
 124         job_dict={}
 125         producer_dict[job_id]=job_dict
 126         job_dict['create_response']=201
 127         job_dict['delete_response']=404
 128         job_dict['json']=None
 129         job_dict['create_counter']=0
 130         job_dict['delete_counter']=0
 131         job_dict['delivering']="stopped"
 132         job_dict['delivery_attempts']=0
 133     return job_dict
 134
 135
 136 # Helper function to get an entry from the callback db
 137 # if job_id is None then only the producer dict is returned (or None if producer is not found)
 138 # if job_id is not None, the job is returned (or None if producer/job is not found)
 139 def get_callback_dict(producer_id, job_id):
 140
 141     producer_dict=None
 142     if (producer_id in db.keys()):
 143         producer_dict=db[producer_id]
 144
 145     if (producer_dict is None):
 146         return None
 147
 148     if (job_id is None):
 149         return producer_dict
 150
 151     job_dict=None
 152     if (job_id in producer_dict.keys()):
 153         job_dict=producer_dict[job_id]
 154
 155     return job_dict
 156
 157 # Helper function find if a key/valye exist in the dictionay tree
 158 # True if found
 159 def recursive_search(s_dict, s_key, s_id):
 160     for pkey in s_dict:
 161         if (pkey == s_key) and (s_dict[pkey] == s_id):
 162             return True
 163         if (isinstance(s_dict[pkey], dict)):
 164             recursive_search(s_dict[pkey], s_key, s_id)
 165
 166     return False
 167
 168 # Helper function to find all job dicts
 169 def get_all_jobs():
 170     job_dicts={}
 171     for producer_key in db:
 172         producer_dict = db[producer_key]
 173         for job_key in producer_dict:
 174             job_dict = producer_dict[job_key]
 175             if (isinstance(job_dict, dict)):
 176                 job_dicts[job_key]=job_dict
 177     return job_dicts
 178
 179 # I'm alive function
 180 # response: always 200
 181 @app.route('/',
 182     methods=['GET'])
 183 def index():
 184     return 'OK', 200
 185
 186 # Arm the create callback with a response code
 187 # Omitting the query parameter switch to response back to the standard 200/201 response
 188 # URI and parameters (PUT): /arm/create/<producer_id>/<job_id>[?response=<repsonsecode>]
 189 # Setting
 190 # response: 200 (400 if incorrect query params)
 191 @app.route(ARM_CREATE_RESPONSE,
 192      methods=['PUT'])
 193 def arm_create(producer_id, job_id):
 194
 195     arm_response=request.args.get('response')
 196
 197     if (arm_response is None):
 198         if (len(request.args) != 0):
 199             return UNKNOWN_QUERY_PARAMETERS,400
 200     else:
 201         if (len(request.args) != 1):
 202             return UNKNOWN_QUERY_PARAMETERS,400
 203
 204
 205     print("Arm create received for producer: "+str(producer_id)+" and job: "+str(job_id)+" and response: "+str(arm_response))
 206
 207     job_dict=setup_callback_dict(producer_id, job_id)
 208
 209     if (arm_response is None):    #Reset the response depending if a job exists or not
 210         if (job_dict['json'] is None):
 211             job_dict['create_response']=201
 212         else:
 213             job_dict['create_response']=200
 214     else:
 215         job_dict['create_response']=int(arm_response)
 216
 217     return "",200
 218
 219 # Arm the delete callback with a response code
 220 # Omitting the query parameter switch to response back to the standard 204 response
 221 # URI and parameters (PUT): /arm/delete/<producer_id>/<job-id>[?response=<responsecode>]
 222 # response: 200 (400 if incorrect query params)
 223 @app.route(ARM_DELETE_RESPONSE,
 224      methods=['PUT'])
 225 def arm_delete(producer_id, job_id):
 226
 227     arm_response=request.args.get('response')
 228
 229     if (arm_response is None):
 230         if (len(request.args) != 0):
 231             return UNKNOWN_QUERY_PARAMETERS,400
 232     else:
 233         if (len(request.args) != 1):
 234             return UNKNOWN_QUERY_PARAMETERS,400
 235
 236     print("Arm delete received for producer: "+str(producer_id)+" and job: "+str(job_id)+" and response: "+str(arm_response))
 237
 238     arm_response=request.args.get('response')
 239
 240     job_dict=setup_callback_dict(producer_id, job_id)
 241
 242     if (arm_response is None): #Reset the response depending if a job exists or not
 243         if (job_dict['json'] is None):
 244             job_dict['delete_response']=404
 245         else:
 246             job_dict['delete_response']=204
 247     else:
 248         job_dict['delete_response']=int(arm_response)
 249
 250     return "",200
 251
 252 # Arm the supervision callback with a response code
 253 # Omitting the query parameter switch to response back to the standard 200 response
 254 # URI and parameters (PUT): /arm/supervision/<producer_id>[?response=<responsecode>]
 255 # response: 200 (400 if incorrect query params)
 256 @app.route(ARM_SUPERVISION_RESPONSE,
 257      methods=['PUT'])
 258 def arm_supervision(producer_id):
 259
 260     arm_response=request.args.get('response')
 261
 262     if (arm_response is None):
 263         if (len(request.args) != 0):
 264             return UNKNOWN_QUERY_PARAMETERS,400
 265     else:
 266         if (len(request.args) != 1):
 267             return UNKNOWN_QUERY_PARAMETERS,400
 268
 269     print("Arm supervision received for producer: "+str(producer_id)+" and response: "+str(arm_response))
 270
 271     producer_dict=setup_callback_dict(producer_id, None)
 272     if (arm_response is None):
 273         producer_dict['supervision_response']=200
 274     else:
 275         producer_dict['supervision_response']=int(arm_response)
 276
 277     return "",200
 278
 279 # Arm a producer with a type
 280 # URI and parameters (PUT): /arm/type/<string:producer_id>/<string:type-id>
 281 # response: 200 (404)
 282 @app.route(ARM_TYPE,
 283     methods=['PUT'])
 284 def arm_type(producer_id, type_id):
 285
 286     print("Arm type received for producer: "+str(producer_id)+" and type: "+str(type_id))
 287
 288     producer_dict=get_callback_dict(producer_id, None)
 289
 290     if (producer_dict is None):
 291         return PRODUCER_NOT_FOUND,404
 292
 293     type_list=producer_dict['types']
 294     if (type_id not in type_list):
 295         type_list.append(type_id)
 296
 297     return "",200
 298
 299 # Disarm a producer with a type
 300 # URI and parameters (DELETE): /arm/type/<string:producer_id>/<string:type-id>
 301 # response: 200 (404)
 302 @app.route(ARM_TYPE,
 303     methods=['DELETE'])
 304 def disarm_type(producer_id, type_id):
 305
 306     print("Disarm type received for producer: "+str(producer_id)+" and type: "+str(type_id))
 307
 308     producer_dict=get_callback_dict(producer_id, None)
 309
 310     if (producer_dict is None):
 311         return PRODUCER_NOT_FOUND,404
 312
 313     if (recursive_search(producer_dict, "ei_job_type",type_id) is True):
 314         return "TYPE_IN_USE",400
 315     elif (recursive_search(producer_dict, "ei_type_identity",type_id) is True):
 316         return "TYPE_IN_USE",400
 317     elif (recursive_search(producer_dict, "info_type_identity",type_id) is True):
 318         return "TYPE_IN_USE",400
 319
 320     type_list=producer_dict['types']
 321     type_list.remove(type_id)
 322
 323     return "",200
 324
 325
 326 # Callback for create job
 327 # URI and parameters (POST): /callbacks/job/<producer_id>
 328 # response 201 at create, 200 at update or other configured response code
 329 @app.route(CALLBACK_CREATE_URL,
 330      methods=['POST'])
 331 def callback_create(producer_id):
 332
 333     req_json_dict=None
 334     try:
 335         req_json_dict = json.loads(request.data)
 336         with open('job-schema.json') as f:
 337             schema = json.load(f)
 338             validate(instance=req_json_dict, schema=schema)
 339     except Exception:
 340         return JSON_CORRUPT,400
 341
 342     producer_dict=get_callback_dict(producer_id, None)
 343     if (producer_dict is None):
 344         return PRODUCER_OR_JOB_NOT_FOUND,400
 345     type_list=producer_dict['types']
 346
 347
 348     if 'ei_type_identity' in req_json_dict.keys():
 349         type_key_name='ei_type_identity'
 350         job_key_name='ei_job_identity'
 351     elif 'info_type_identity' in req_json_dict.keys():
 352         type_key_name='info_type_identity'
 353         job_key_name='info_job_identity'
 354     else:
 355         return TYPE_NOT_FOUND, 400
 356
 357     type_id=req_json_dict[type_key_name]
 358     job_id=req_json_dict[job_key_name]
 359
 360     job_dict=get_callback_dict(producer_id, job_id)
 361     if (job_dict is None):
 362         return PRODUCER_OR_JOB_NOT_FOUND,400
 363     return_code=0
 364     return_msg=""
 365     if (req_json_dict[job_key_name] == job_id):
 366         print("Create callback received for producer: "+str(producer_id)+" and job: "+str(job_id))
 367         print(json.loads(request.data))
 368         return_code=job_dict['create_response']
 369         if ((job_dict['create_response'] == 200) or (job_dict['create_response'] == 201)):
 370             job_dict['json']=req_json_dict
 371             job_dict['delivering']="delivering"
 372             if (job_dict['create_response'] == 201): #Set up next response code if create was ok
 373                 job_dict['create_response'] = 200
 374             if (job_dict['delete_response'] == 404):
 375                 job_dict['delete_response'] = 204
 376         else:
 377             if(job_dict['delivering'] == "delivering"):
 378                 job_dict['delivering']="hold"
 379             return_msg=RETURNING_CONFIGURED_RESP
 380
 381         job_dict['create_counter']=job_dict['create_counter']+1
 382     else:
 383         return JOBID_NO_MATCH, 400
 384
 385     return return_msg, return_code
 386
 387 # Callback for delete job
 388 # URI and parameters (DELETE): /callbacks/job/<producer_id>/<job_id>
 389 # response: 204 at delete or other configured response code
 390 @app.route(CALLBACK_DELETE_URL,
 391      methods=['DELETE'])
 392 def callback_delete(producer_id, job_id):
 393
 394     job_dict=get_callback_dict(producer_id, job_id)
 395     if (job_dict is None):
 396         return PRODUCER_OR_JOB_NOT_FOUND,400
 397     return_code=0
 398     return_msg=""
 399     print("Delete callback received for producer: "+str(producer_id)+" and job: "+str(job_id))
 400     return_code=job_dict['delete_response']
 401     if (job_dict['delete_response'] == 204):
 402         job_dict['json']=None
 403         job_dict['delete_response']=404
 404         job_dict['delivering']="stopped"
 405         if (job_dict['create_response'] == 200):
 406             job_dict['create_response'] = 201 # reset create response if delete was ok
 407     else:
 408         return_msg=RETURNING_CONFIGURED_RESP
 409
 410     job_dict['delete_counter']=job_dict['delete_counter']+1
 411
 412     return return_msg, return_code
 413
 414 # Callback for supervision of producer
 415 # URI and parameters (GET): /callbacks/supervision/<producer_id>
 416 # response: 200 or other configured response code
 417 @app.route(CALLBACK_SUPERVISION_URL,
 418      methods=['GET'])
 419 def callback_supervision(producer_id):
 420
 421     print("Supervision callback received for producer: "+str(producer_id))
 422
 423     producer_dict=get_callback_dict(producer_id, None)
 424     if (producer_dict is None):
 425         return PRODUCER_NOT_FOUND,400
 426     return_code=producer_dict['supervision_response']
 427     return_msg=""
 428     if (return_code != 200):
 429         return_msg="returning configured response code"
 430
 431     producer_dict['supervision_counter']=producer_dict['supervision_counter']+1
 432
 433     return return_msg,producer_dict['supervision_response']
 434
 435 # Get the job definition for a job
 436 # URI and parameters (GET): "/jobdata/<string:producer_id>/<string:job_id>"
 437 # response: 200 or 204
 438 @app.route(JOB_DATA,
 439      methods=['GET'])
 440 def get_jobdata(producer_id, job_id):
 441
 442     print("Get job data received for producer: "+str(producer_id)+" and job: "+str(job_id))
 443
 444     job_dict=get_callback_dict(producer_id, job_id)
 445
 446     if (job_dict is None):
 447         return PRODUCER_OR_JOB_NOT_FOUND,400
 448
 449     if (job_dict['json'] is None):
 450         return "",204
 451     else:
 452         return json.dumps(job_dict['json']), 200
 453
 454 # Delete the job definition for a job
 455 # URI and parameters (DELETE): "/jobdata/<string:producer_id>/<string:job_id>"
 456 # response: 204
 457 @app.route(JOB_DATA,
 458      methods=['DELETE'])
 459 def del_jobdata(producer_id, job_id):
 460
 461     print("Delete job data received for producer: "+str(producer_id)+" and job: "+str(job_id))
 462
 463     job_dict=get_callback_dict(producer_id, job_id)
 464
 465     if (job_dict is None):
 466         return PRODUCER_OR_JOB_NOT_FOUND,400
 467
 468     job_dict['json']=None
 469
 470     return "",204
 471
 472
 473 # Start data delivery for a job, action : START or STOP
 474 # URI and parameters (POST): "/jobdata/<string:producer_id>/<string:job_id>?action=action"
 475 # response: 200 or 204
 476 @app.route(JOB_DATA,
 477      methods=['POST'])
 478 def start_jobdata(producer_id, job_id):
 479
 480     action=request.args.get('action')
 481
 482     if (action is None):
 483         return UNKNOWN_QUERY_PARAMETERS,400
 484     else:
 485         if (len(request.args) != 1):
 486             return UNKNOWN_QUERY_PARAMETERS,400
 487         else:
 488             if ((action != "START") and (action != "STOP")):
 489                 return UNKNOWN_QUERY_PARAMETERS,400
 490
 491     print("Job data action received for producer: "+str(producer_id)+" and job: "+str(job_id) + " action: " + action)
 492
 493     job_dict=get_callback_dict(producer_id, job_id)
 494     if (job_dict is None):
 495         return JOB_NOT_FOUND,404
 496
 497     if (job_dict['json'] is None):
 498         return JOB_DATA_NOT_FOUND, 400
 499     else:
 500         if (action == "START"):
 501             job_dict['delivering']="delivering"
 502         else:
 503             job_dict['delivering']="stopped"
 504         return "",200
 505
 506
 507 # Counter for create calls for a job
 508 # URI and parameters (GET): "/counter/create/<string:producer_id>/<string:job_id>"
 509 # response: 200 and counter value
 510 @app.route(COUNTER_CREATE,
 511      methods=['GET'])
 512 def counter_create(producer_id, job_id):
 513     job_dict=get_callback_dict(producer_id, job_id)
 514     if (job_dict is None):
 515         return "-1",200
 516     return str(job_dict['create_counter']),200
 517
 518 # Counter for delete calls for a job
 519 # URI and parameters (GET): "/counter/delete/<string:producer_id>/<string:job_id>"
 520 # response: 200 and counter value
 521 @app.route(COUNTER_DELETE,
 522      methods=['GET'])
 523 def counter_delete(producer_id, job_id):
 524     job_dict=get_callback_dict(producer_id, job_id)
 525     if (job_dict is None):
 526         return "-1",200
 527     return str(job_dict['delete_counter']),200
 528
 529 # Counter for supervision calls for a producer
 530 # URI and parameters (GET): "/counter/supervision/<string:producer_id>"
 531 # response: 200 and counter value
 532 @app.route(COUNTER_SUPERVISION,
 533      methods=['GET'])
 534 def counter_supervision(producer_id):
 535     producer_dict=get_callback_dict(producer_id, None)
 536     if (producer_dict is None):
 537         return "-1",200
 538     return str(producer_dict['supervision_counter']),200
 539
 540 # Get status info
 541 # URI and parameters (GET): "/status"
 542 # -
 543 @app.route(STATUS,
 544     methods=['GET'])
 545 def status():
 546     global db
 547     return json.dumps(db),200
 548
 549
 550 # Reset db
 551 @app.route('/reset',
 552     methods=['GET', 'POST', 'PUT'])
 553 def reset():
 554     global db
 555     db={}
 556     return "",200
 557
 558
 559 def datadelivery() :
 560     while True:
 561         try:
 562             job_dicts=get_all_jobs()
 563             for key in job_dicts:
 564                 job=job_dicts[key]
 565                 if (job['delivering'] == "delivering" and job['json'] != None):
 566                     url=job['json']['target_uri']
 567                     if (str(url).find("localhost:") == -1):   #Dont deliver to localhost...
 568                         data={}
 569                         data["date"]=str(datetime.datetime.now())
 570                         data["job"]=""+key
 571                         data["sequence_no"]=""+str(job['delivery_attempts'])
 572                         data["value"]=str(100)
 573                         print("Sending to "+url+" payload:"+json.dumps(data))
 574
 575                         requests.post(url, json=data, verify=False, timeout=2) #NOSONAR
 576                         job['delivery_attempts'] += 1
 577         except Exception as err:
 578             print("Error during data delivery: "+ str(err))
 579         time.sleep(1)
 580
 581
 582 ### Main function ###
 583
 584 print("Starting data delivery thread")
 585 thread = threading.Thread(target=datadelivery, args=())
 586 thread.daemon = True
 587 thread.start()
 588
 589 if __name__ == "__main__":
 590     app.run(port=HOST_PORT, host=HOST_IP)