# ============LICENSE_START===============================================
-# Copyright (C) 2020 Nordix Foundation. All rights reserved.
+# Copyright (C) 2021 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.
# ============LICENSE_END=================================================
#
-# This test case test the STD_2.0.0 version of the simulator
+# This test case tests the STD_2.0.0 version of the simulator.
import json
+import pytest
import time
+import multiprocessing
+from unittest_setup import SERVER_URL, HOST_IP, PORT_NUMBER, setup_env, get_testdata_dir, client
+from unittest_setup import run_flask_app
-#Version of simulator
-INTERFACE_VERSION="STD_2.0.0"
+# Setup env and import paths
-from unittest_setup import SERVER_URL, HOST_IP, PORT_NUMBER, setup_env, get_testdata_dir, client
+# Version of simulator
+INTERFACE_VERSION="STD_2.0.0"
-#Setup env and import paths
setup_env(INTERFACE_VERSION)
-
from compare_json import compare
+from models.enforceStatus import EnforceStatus
+
+def test_enforce_reason(client):
+ """
+ Test that we can set a valid enforce status and reason, and that we reject invalid cases.
+ """
+ enforceStatus = EnforceStatus()
+
+ enforceStatus.enforce_status = 'NOT_ENFORCED'
+ enforceStatus.enforce_reason = 'SCOPE_NOT_APPLICABLE'
+ enforce_dict = enforceStatus.to_dict()
+ assert enforce_dict['enforceStatus'] == 'NOT_ENFORCED'
+ assert enforce_dict['enforceReason'] == 'SCOPE_NOT_APPLICABLE'
+
+ enforceStatus.enforce_status = 'ENFORCED'
+ enforceStatus.enforce_reason = 'STATEMENT_NOT_APPLICABLE'
+ enforce_dict = enforceStatus.to_dict()
+ assert enforce_dict['enforceStatus'] == 'ENFORCED'
+ assert enforce_dict['enforceReason'] == 'STATEMENT_NOT_APPLICABLE'
+
+ enforceStatus.enforce_reason = 'OTHER_REASON'
+ enforce_dict = enforceStatus.to_dict()
+ assert enforce_dict['enforceReason'] == 'OTHER_REASON'
+
+ enforce_status = enforceStatus.enforce_status
+ assert str(enforce_status) == 'ENFORCED'
+
+ enforce_reason = enforceStatus.enforce_reason
+ assert str(enforce_reason) == 'OTHER_REASON'
+
+ with pytest.raises(ValueError):
+ enforceStatus.enforce_status = 'ERROR'
+
+ with pytest.raises(ValueError):
+ enforceStatus.enforce_reason = 'ERROR'
+
+
def test_apis(client):
testdata=get_testdata_dir()
res=compare(json_payload, result)
assert res == True
- # API: Duplicate policy instance pi2 of type: STD_1
+ # # API: Duplicate policy instance pi2 of type: STD_1 - and delete it
+
with open(testdata+'pi1_updated.json') as json_file:
- data_response = {"title": "Duplicate, the policy json already exists.", "status": 400, "instance": "pi2"}
json_payload=json.load(json_file)
response=client.put(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2', headers=header, data=json.dumps(json_payload))
- assert response.status_code == 400
+ assert response.status_code == 201
result=json.loads(response.data)
- res=compare(data_response, result)
+ res=compare(json_payload, result)
assert res == True
+ data_response = b""
+ response=client.delete(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2')
+ assert response.status_code == 204
+ assert data_response == response.data
+
# API: Get policy instances, shall contain pi1
data_response = ["pi1"]
assert res == True
# API: API: Get policy status
- data_response = {"enforceStatus" : "", "enforceReason" : ""}
+ data_response = {"enforceStatus" : "NOT_ENFORCED", "enforceReason" : "OTHER_REASON"}
response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi1/status')
assert response.status_code == 200
result=json.loads(response.data)
assert res == True
# API: API: Get policy status
- data_response = {"enforceStatus" : "", "enforceReason" : ""}
+ data_response = {"enforceStatus" : "NOT_ENFORCED", "enforceReason" : "OTHER_REASON"}
response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2/status')
assert response.status_code == 200
result=json.loads(response.data)
assert res == True
# Set status for policy instance pi2
- response=client.put(SERVER_URL+'status?policyid=pi2&status=OK')
+ response=client.put(SERVER_URL+'status?policyid=pi2&status=NOT_ENFORCED&reason=STATEMENT_NOT_APPLICABLE')
assert response.status_code == 200
# API: API: Get policy status
- data_response = {"enforceStatus" : "OK"}
+ data_response = {"enforceStatus" : "NOT_ENFORCED", "enforceReason" : "STATEMENT_NOT_APPLICABLE"}
response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2/status')
assert response.status_code == 200
result=json.loads(response.data)
assert res == True
# Set status for policy instance pi2
- response=client.put(SERVER_URL+'status?policyid=pi2&status=NOTOK&reason=notok_reason')
+ response=client.put(SERVER_URL+'status?policyid=pi2&status=NOT_ENFORCED&reason=OTHER_REASON')
assert response.status_code == 200
# API: API: Get policy status
- data_response = {"enforceStatus" : "NOTOK", "enforceReason" : "notok_reason"}
+ data_response = {"enforceStatus" : "NOT_ENFORCED", "enforceReason" : "OTHER_REASON"}
response=client.get(SERVER_URL+'A1-P/v2/policytypes/STD_1/policies/pi2/status')
assert response.status_code == 200
result=json.loads(response.data)
res=compare(data_response, result)
assert res == True
-
- # #Found no way to test these functions
- # #'sendstatus' will send a http request that will fail
- # #since no server will receive the call
- # #These function is instead tested when running the bash script in the 'test' dir
- # # # Send status for pi2
- # # response=client.post(SERVER_URL+'sendstatus?policyid=pi2')
- # # assert response.status_code == 200
- # # result=json.loads(response.data)
- # # res=compare(data_get_status, result)
- # # assert res == True
-
- # # # Send status, shall fail
- # # response=client.post(SERVER_URL+'sendstatus')
- # # assert response.status_code == 400
-
# Get counter: data_delivery
response=client.get(SERVER_URL+'counter/datadelivery')
assert response.status_code == 200
# Get counter: types
response=client.get(SERVER_URL+'counter/num_types')
assert response.status_code == 200
- assert response.data == b"1"
\ No newline at end of file
+ assert response.data == b"1"
+
+def test_notificationDestination(client):
+ test_data = get_testdata_dir() + 'pi2.json'
+ # Header for json payload
+ header = { "Content-Type" : "application/json" }
+
+ # === API: Update policy instance pi2 of type: 2 ==="
+ with open(test_data) as json_file:
+ payload = json.load(json_file)
+ response = client.put(SERVER_URL+"A1-P/v2/policytypes/STD_1/policies/pi2?notificationDestination=http://localhost:8086/statustest", headers=header, data=json.dumps(payload))
+ assert response.status_code == 200
+ result = json.loads(response.data)
+ assert compare(payload, result) == True
+
+def test_sendstatus(client):
+ # Create a new thread to run the Flask app in parallel on a different port so that we can call the callback.
+ proc = multiprocessing.Process(target=run_flask_app, args=())
+ proc.start()
+
+ test_data = get_testdata_dir() + 'pi2.json'
+ header = { "Content-Type" : "application/json" }
+
+ # === Send status for pi2===
+ with open(test_data) as json_file:
+ payload = json.load(json_file)
+ response = client.post(SERVER_URL+'sendstatus?policyid=pi2', headers=header, data=json.dumps(payload))
+
+ assert response.status_code == 204
+ result = response.data
+ assert result == b""
+
+ # Send status, negative test with missing parameter
+ response = client.post(SERVER_URL+'sendstatus', headers=header, data="")
+ assert response.status_code == 400
+
+ # Send status pi9, negative test for policy id not found
+ response = client.post(SERVER_URL+'sendstatus?policyid=pi9', headers=header, data="")
+ assert response.status_code == 404
+
+ proc.terminate()
Code Review / sim / a1-interface.git / blobdiff