--- /dev/null
+# ============LICENSE_START===============================================
+# Copyright (C) 2020 Nordix Foundation. All rights reserved.
+# ========================================================================
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ============LICENSE_END=================================================
+#
+
+# This test case test the STD_1.1.3 version of the simulator
+
+from unittest import TestCase
+import sys
+import os
+import threading
+import requests
+import json
+
+#Constants for the test case
+INTERFACE_VERSION="STD_1.1.3"
+PORT_NUMBER=2224
+HOST_IP="127.0.0.1"
+SERVER_URL="http://127.0.0.1:"+str(PORT_NUMBER)+"/"
+
+cwd=os.getcwd()+"/"
+# Env TESTS_BASE_PATH is set when executed via tox.ini
+# If basic test is executed from cmd line, that env var is not needed
+if 'TESTS_BASE_PATH' in os.environ:
+ cwd=os.environ['TESTS_BASE_PATH']+"/"+INTERFACE_VERSION+"/"
+TESTDATA=cwd+"/../../test/"+INTERFACE_VERSION+"/jsonfiles/"
+
+#Env var to setup api version and host logging
+os.environ['APIPATH'] = cwd+"/../../api/"+INTERFACE_VERSION
+os.environ['REMOTE_HOSTS_LOGGING'] = "ON"
+
+# Paths need to run the sim, including needed source file dirs
+sys.path.append(os.path.abspath(cwd+'../../src/common'))
+sys.path.append(os.path.abspath(cwd+'../../test/common'))
+sys.path.append(os.path.abspath(cwd+'../../src/'+INTERFACE_VERSION))
+os.chdir(cwd+"../../src/"+INTERFACE_VERSION)
+
+import main
+from compare_json import compare
+
+class TestCase1(TestCase):
+ from main import app
+ server = None
+
+ def startServer(self):
+ self.app.run(port=PORT_NUMBER, host=HOST_IP, threaded=True)
+
+ def setUp(self):
+ self.server=threading.Thread(target=self.startServer, args=())
+ self.server.daemon = True
+ self.server.start()
+
+ # executed after each test
+ def tearDown(self):
+ self.server.killed=True
+
+ #from main import app
+ def test_apis(self):
+
+ # header for json payload
+ header = {
+ "Content-Type" : "application/json"
+ }
+
+ # Simulator hello world
+ response=requests.get(SERVER_URL)
+ self.assertEqual(response.status_code, 200)
+
+ # Check used and implemented interfaces
+ response=requests.get(SERVER_URL+'container_interfaces')
+ self.assertEqual(response.status_code, 200)
+ self.assertEqual(response.text, "Current interface: STD_1.1.3 All supported A1 interface yamls in this container: ['OSC_2.1.0', 'STD_1.1.3']")
+
+ # Reset simulator instances
+ response=requests.post(SERVER_URL+'deleteinstances')
+ self.assertEqual(response.status_code, 200)
+
+ # Reset simulator, all
+ response=requests.post(SERVER_URL+'deleteall')
+ self.assertEqual(response.status_code, 200)
+
+ # API: Get policy instances, shall be empty
+ data_policy_get = [ ]
+ response=requests.get(SERVER_URL+'A1-P/v1/policies')
+ self.assertEqual(response.status_code, 200)
+ result=json.loads(response.text)
+ res=compare(data_policy_get, result)
+ self.assertEqual(res, True)
+
+ #API: Create policy instance pi1
+ data_pi1 = {
+ "scope": {
+ "ueId": "ue1",
+ "groupId": "group1",
+ "sliceId": "slice1",
+ "qosId": "qos1",
+ "cellId": "cell1"
+ },
+ "statement": {
+ "priorityLevel": 5
+ }
+ }
+ response=requests.put(SERVER_URL+'A1-P/v1/policies/pi1', headers=header, data=json.dumps(data_pi1))
+ self.assertEqual(response.status_code, 201)
+ result=json.loads(response.text)
+ res=compare(data_pi1, result)
+ self.assertEqual(res, True)
+
+ #API: Update policy instance pi1
+ data_pi1_updated = {
+ "scope": {
+ "ueId": "ue1",
+ "groupId": "group1",
+ "sliceId": "slice1",
+ "qosId": "qos1",
+ "cellId": "cell1"
+ },
+ "statement": {
+ "priorityLevel": 6
+ }
+ }
+ response=requests.put(SERVER_URL+'A1-P/v1/policies/pi1', headers=header, data=json.dumps(data_pi1_updated))
+ self.assertEqual(response.status_code, 200)
+ result=json.loads(response.text)
+ res=compare(data_pi1_updated, result)
+ self.assertEqual(res, True)
+
+ #API: Create policy instance pi9, bad json
+ response=requests.put(SERVER_URL+'A1-P/v1/policies/pi9', headers=header, data="test")
+ self.assertEqual(response.status_code, 400)
+
+ # API: Get policy instances, shall contain pi1
+ data_policy_get = [ "pi1" ]
+ response=requests.get(SERVER_URL+'A1-P/v1/policies')
+ self.assertEqual(response.status_code, 200)
+ result=json.loads(response.text)
+ res=compare(data_policy_get, result)
+ self.assertEqual(res, True)
+
+ # API: Create policy instance pi2 (copy of pi1). Shall fail
+ data_create_errror_pi1 = {
+ "title" : "The policy json already exists.",
+ "status" : 400,
+ "instance" : "pi2"
+ }
+ response=requests.put(SERVER_URL+'A1-P/v1/policies/pi2', headers=header, data=json.dumps(data_pi1_updated))
+ self.assertEqual(response.status_code, 400)
+ result=json.loads(response.text)
+ res=compare(data_create_errror_pi1, result)
+ self.assertEqual(res, True)
+
+ # Set force response code 409. ==="
+ response=requests.post(SERVER_URL+'forceresponse?code=409')
+ self.assertEqual(response.status_code, 200)
+
+ # API: Get policy instances, shall fail
+ data_get_errror = {
+ "title" : "Conflict",
+ "status" : 409,
+ "detail" : "Request could not be processed in the current state of the resource"
+ }
+ response=requests.get(SERVER_URL+'A1-P/v1/policies')
+ self.assertEqual(response.status_code, 409)
+ result=json.loads(response.text)
+ res=compare(data_get_errror, result)
+ self.assertEqual(res, True)
+
+ # Reset force response
+ response=requests.post(SERVER_URL+'forceresponse')
+ self.assertEqual(response.status_code, 200)
+ self.assertEqual(response.text, "Force response code: None set for one single A1 response")
+
+ ###
+ ### Repeating the above two test for code coverage
+ ###
+ # Set force response code 400
+ response=requests.post(SERVER_URL+'forceresponse?code=400')
+ self.assertEqual(response.status_code, 200)
+
+ #API: Create policy instance pi3, shall fail
+ data_pi3 = {
+ "scope": {
+ "ueId": "ue3"
+ },
+ "statement": {
+ "priorityLevel": 5
+ }
+ }
+ data_get_errror = {
+ "title" : "Bad request",
+ "status" : 400,
+ "detail" : "Object in payload not properly formulated or not related to the method"
+ }
+ response=requests.put(SERVER_URL+'A1-P/v1/policies/pi3', headers=header, data=json.dumps(data_pi3))
+ self.assertEqual(response.status_code, 400)
+ result=json.loads(response.text)
+ res=compare(data_get_errror, result)
+ self.assertEqual(res, True)
+
+ # Set force response code 404
+ response=requests.post(SERVER_URL+'forceresponse?code=404')
+ self.assertEqual(response.status_code, 200)
+
+ # API: Get policy instance pi1, shall fail =="
+ data_get_errror = {
+ "title" : "Not found",
+ "status" : 404,
+ "detail" : "No resource found at the URI"
+ }
+ response=requests.get(SERVER_URL+'A1-P/v1/policies/pi1')
+ self.assertEqual(response.status_code, 404)
+ result=json.loads(response.text)
+ res=compare(data_get_errror, result)
+ self.assertEqual(res, True)
+
+ # Set force response code 405
+ response=requests.post(SERVER_URL+'forceresponse?code=405')
+ self.assertEqual(response.status_code, 200)
+
+ # API: Delete policy instances pi1, shall fail =="
+ data_get_errror = {
+ "title" : "Method not allowed",
+ "status" : 405,
+ "detail" : "Method not allowed for the URI"
+ }
+ response=requests.delete(SERVER_URL+'A1-P/v1/policies/pi1')
+ self.assertEqual(response.status_code, 405)
+ result=json.loads(response.text)
+ res=compare(data_get_errror, result)
+ self.assertEqual(res, True)
+
+ # Set force response code 429
+ response=requests.post(SERVER_URL+'forceresponse?code=429')
+ self.assertEqual(response.status_code, 200)
+
+ # API: Get policy status pi3, shall fail =="
+ data_get_errror = {
+ "title" : "Too many requests",
+ "status" : 429,
+ "detail" : "Too many requests have been sent in a given amount of time"
+ }
+ response=requests.get(SERVER_URL+'A1-P/v1/policies/pi3/status')
+ self.assertEqual(response.status_code, 429)
+ result=json.loads(response.text)
+ res=compare(data_get_errror, result)
+ self.assertEqual(res, True)
+
+ # Set force response code 507
+ response=requests.post(SERVER_URL+'forceresponse?code=507')
+ self.assertEqual(response.status_code, 200)
+
+ # API: Get policy instances, shall fail
+ data_get_errror = {
+ "title" : "Insufficient storage",
+ "status" : 507,
+ "detail" : "The method could not be performed on the resource because the provider is unable to store the representation needed to successfully complete the request"
+ }
+ response=requests.get(SERVER_URL+'A1-P/v1/policies')
+ self.assertEqual(response.status_code, 507)
+ result=json.loads(response.text)
+ res=compare(data_get_errror, result)
+ self.assertEqual(res, True)
+
+ # Set force response code 503. ==="
+ response=requests.post(SERVER_URL+'forceresponse?code=503')
+ self.assertEqual(response.status_code, 200)
+
+ # API: Get policy instances, shall fail
+ data_get_errror = {
+ "title" : "Service unavailable",
+ "status" : 503,
+ "detail" : "The provider is currently unable to handle the request due to a temporary overload"
+ }
+ response=requests.get(SERVER_URL+'A1-P/v1/policies')
+ self.assertEqual(response.status_code, 503)
+ result=json.loads(response.text)
+ res=compare(data_get_errror, result)
+ self.assertEqual(res, True)
+
+ # Set force response code 555. ==="
+ response=requests.post(SERVER_URL+'forceresponse?code=555')
+ self.assertEqual(response.status_code, 200)
+
+ # API: Get policy instances, shall fail
+ data_get_errror = {
+ "title" : "Unknown",
+ "status" : "555",
+ "detail" : "Not implemented response code"
+ }
+ response=requests.get(SERVER_URL+'A1-P/v1/policies')
+ self.assertEqual(response.status_code, 555)
+ result=json.loads(response.text)
+ res=compare(data_get_errror, result)
+ self.assertEqual(res, True)
+
+ ###
+ ### End of repeated test
+ ###
+
+
+ # API: Get policy status
+ data_policy_status = {
+ "enforceStatus" : "UNDEFINED"
+ }
+ response=requests.get(SERVER_URL+'A1-P/v1/policies/pi1/status')
+ self.assertEqual(response.status_code, 200)
+ result=json.loads(response.text)
+ res=compare(data_policy_status, result)
+ self.assertEqual(res, True)
+
+ # API: Create policy instance pi2
+ data_pi2 = {
+ "scope": {
+ "ueId": "ue2",
+ "groupId": "group2",
+ "sliceId": "slice2",
+ "qosId": "qos2",
+ "cellId": "cell2"
+ },
+ "statement": {
+ "priorityLevel": 10
+ }
+ }
+ response=requests.put(SERVER_URL+'A1-P/v1/policies/pi2', headers=header, data=json.dumps(data_pi2))
+ self.assertEqual(response.status_code, 201)
+ result=json.loads(response.text)
+ res=compare(data_pi2, result)
+ self.assertEqual(res, True)
+
+
+ # API: Update policy instance pi2
+ # Reuse same policy data
+ response=requests.put(SERVER_URL+'A1-P/v1/policies/pi2?notificationDestination=http://localhost:2224/statustest', headers=header, data=json.dumps(data_pi2))
+ self.assertEqual(response.status_code, 200)
+ result=json.loads(response.text)
+ res=compare(data_pi2, result)
+ self.assertEqual(res, True)
+
+ # API: Get policy instances, shall contain pi1 and pi2
+ data_policy_get = [ "pi1", "pi2" ]
+ response=requests.get(SERVER_URL+'A1-P/v1/policies')
+ self.assertEqual(response.status_code, 200)
+ result=json.loads(response.text)
+ res=compare(data_policy_get, result)
+ self.assertEqual(res, True)
+
+ # Set force delay 10
+ response=requests.post(SERVER_URL+'forcedelay?delay=10')
+ self.assertEqual(response.status_code, 200)
+ self.assertEqual(response.text, "Force delay: 10 sec set for all A1 responses")
+
+ # API: Get policy instances, shall contain pi1 and pi2 and delayed 10 sec
+ data_policy_get = [ "pi1", "pi2" ]
+ response=requests.get(SERVER_URL+'A1-P/v1/policies')
+ self.assertEqual(response.status_code, 200)
+ result=json.loads(response.text)
+ res=compare(data_policy_get, result)
+ self.assertEqual(res, True)
+
+ # Reset force delay
+ response=requests.post(SERVER_URL+'forcedelay')
+ self.assertEqual(response.status_code, 200)
+ self.assertEqual(response.text, "Force delay: None sec set for all A1 responses")
+
+ # API: GET policy instance pi1
+ response=requests.get(SERVER_URL+'A1-P/v1/policies/pi1')
+ self.assertEqual(response.status_code, 200)
+ result=json.loads(response.text)
+ res=compare(data_pi1_updated, result)
+ self.assertEqual(res, True)
+
+ # API: GET policy instance pi2
+ response=requests.get(SERVER_URL+'A1-P/v1/policies/pi2')
+ self.assertEqual(response.status_code, 200)
+ result=json.loads(response.text)
+ res=compare(data_pi2, result)
+ self.assertEqual(res, True)
+
+ # API: GET policy instance pi9, shall fail
+ response=requests.get(SERVER_URL+'A1-P/v1/policies/pi9')
+ self.assertEqual(response.status_code, 404)
+
+ # API: DELETE policy instance pi1
+ response=requests.delete(SERVER_URL+'A1-P/v1/policies/pi1')
+ self.assertEqual(response.status_code, 204)
+
+ # API: DELETE policy instance pi9, shall fail
+ response=requests.delete(SERVER_URL+'A1-P/v1/policies/pi9')
+ self.assertEqual(response.status_code, 404)
+
+ # API: Get policy status for pi1, shall fail
+ data_get_errror = {
+ "title" : "The policy identity does not exist.",
+ "status" : 404,
+ "detail" : "There is no existing policy instance with the identity: pi1",
+ "instance" : "pi1"
+ }
+ response=requests.get(SERVER_URL+'A1-P/v1/policies/pi1/status')
+ self.assertEqual(response.status_code, 404)
+ result=json.loads(response.text)
+ res=compare(data_get_errror, result)
+ self.assertEqual(res, True)
+
+ # Set status for policy instance pi2
+ response=requests.put(SERVER_URL+'status?policyid=pi2&status=OK')
+ self.assertEqual(response.status_code, 200)
+
+ # API: Get policy status for pi2
+ data_get_status = {
+ "enforceStatus" : "OK"
+ }
+ response=requests.get(SERVER_URL+'A1-P/v1/policies/pi2/status')
+ self.assertEqual(response.status_code, 200)
+ result=json.loads(response.text)
+ res=compare(data_get_status, result)
+ self.assertEqual(res, True)
+
+ # Set status for policy instance pi2
+ response=requests.put(SERVER_URL+'status?policyid=pi2&status=NOTOK&reason=notok_reason')
+ self.assertEqual(response.status_code, 200)
+
+ # API: Get policy status for pi2
+ data_get_status = {
+ "enforceStatus" : "NOTOK",
+ "enforceReason" : "notok_reason"
+ }
+ response=requests.get(SERVER_URL+'A1-P/v1/policies/pi2/status')
+ self.assertEqual(response.status_code, 200)
+ result=json.loads(response.text)
+ res=compare(data_get_status, result)
+ self.assertEqual(res, True)
+
+ # Send status for pi2
+ response=requests.post(SERVER_URL+'sendstatus?policyid=pi2')
+ self.assertEqual(response.status_code, 200)
+ result=json.loads(response.text)
+ res=compare(data_get_status, result)
+ self.assertEqual(res, True)
+
+ # Send status, shall fail
+ response=requests.post(SERVER_URL+'sendstatus')
+ self.assertEqual(response.status_code, 400)
+
+ # Send status pi9, shall fail
+ response=requests.post(SERVER_URL+'sendstatus?policyid=pi9')
+ self.assertEqual(response.status_code, 404)
+
+ # Set status for policy instance , shall fail
+ response=requests.put(SERVER_URL+'status')
+ self.assertEqual(response.status_code, 400)
+
+ # Set status for policy instance pi9, shall fail
+ response=requests.put(SERVER_URL+'status?policyid=pi9')
+ self.assertEqual(response.status_code, 404)
+
+ # Set status for policy instance pi2, shall fail
+ response=requests.put(SERVER_URL+'status?policyid=pi2')
+ self.assertEqual(response.status_code, 400)
+
+
+ # Get counter: intstance
+ response=requests.get(SERVER_URL+'counter/num_instances')
+ self.assertEqual(response.status_code, 200)
+ self.assertEqual(response.text, "1")
+
+ # Get counter: types (shall be 0)
+ response=requests.get(SERVER_URL+'counter/num_types')
+ self.assertEqual(response.status_code, 200)
+ self.assertEqual(response.text, "0")
+
+ # Get counter: interface
+ response=requests.get(SERVER_URL+'counter/interface')
+ self.assertEqual(response.status_code, 200)
+ self.assertEqual(response.text, "STD_1.1.3")
+
+ # Get counter: remote hosts
+ response=requests.get(SERVER_URL+'counter/remote_hosts')
+ self.assertEqual(response.status_code, 200)
+
+ # Get counter: test, shall fail
+ response=requests.get(SERVER_URL+'counter/test')
+ self.assertEqual(response.status_code, 404)
Code Review / sim / a1-interface.git / blobdiff