aboutsummaryrefslogtreecommitdiffstats
path: root/moonv4/moon_interface/tests/apitests/utils
diff options
context:
space:
mode:
authorWuKong <rebirthmonkey@gmail.com>2017-04-22 13:25:07 +0200
committerWuKong <rebirthmonkey@gmail.com>2017-04-22 13:25:07 +0200
commitd182202fc6001983541504ed323d68479086317e (patch)
tree11d4c10cdd3e995f519c3e0e324968fdaf175114 /moonv4/moon_interface/tests/apitests/utils
parent83c1c966baf73329fab8ddcfad19ad7fe0c41c2a (diff)
add moonv4
Change-Id: I247af788d0b0fb961fbc85416486b241eb1d807c Signed-off-by: WuKong <rebirthmonkey@gmail.com>
Diffstat (limited to 'moonv4/moon_interface/tests/apitests/utils')
-rw-r--r--moonv4/moon_interface/tests/apitests/utils/__init__.py0
-rw-r--r--moonv4/moon_interface/tests/apitests/utils/models.py260
-rw-r--r--moonv4/moon_interface/tests/apitests/utils/pdp.py145
-rw-r--r--moonv4/moon_interface/tests/apitests/utils/policies.py581
4 files changed, 986 insertions, 0 deletions
diff --git a/moonv4/moon_interface/tests/apitests/utils/__init__.py b/moonv4/moon_interface/tests/apitests/utils/__init__.py
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/moonv4/moon_interface/tests/apitests/utils/__init__.py
diff --git a/moonv4/moon_interface/tests/apitests/utils/models.py b/moonv4/moon_interface/tests/apitests/utils/models.py
new file mode 100644
index 00000000..1ba6e440
--- /dev/null
+++ b/moonv4/moon_interface/tests/apitests/utils/models.py
@@ -0,0 +1,260 @@
+import requests
+import copy
+
+URL = "http://172.18.0.11:38001{}"
+HEADERS = {"content-type": "application/json"}
+
+model_template = {
+ "name": "test_model",
+ "description": "test",
+ "meta_rules": []
+}
+
+category_template = {
+ "name": "name of the category",
+ "description": "description of the category"
+}
+
+meta_rule_template = {
+ "name": "test_meta_rule",
+ "subject_categories": [],
+ "object_categories": [],
+ "action_categories": []
+}
+
+
+def check_model(model_id=None):
+ req = requests.get(URL.format("/models"))
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ assert "models" in result
+ if model_id:
+ assert result["models"]
+ assert model_id in result['models']
+ assert "name" in result['models'][model_id]
+ assert model_template["name"] == result['models'][model_id]["name"]
+ return result['models']
+
+
+def add_model(name=None):
+ if name:
+ model_template['name'] = name
+ req = requests.post(URL.format("/models"), json=model_template, headers=HEADERS)
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ model_id = list(result['models'].keys())[0]
+ if "result" in result:
+ assert result["result"]
+ assert "name" in result['models'][model_id]
+ assert model_template["name"] == result['models'][model_id]["name"]
+ return model_id
+
+
+def delete_model(model_id):
+ req = requests.delete(URL.format("/models/{}".format(model_id)))
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ assert "result" in result
+ assert result["result"]
+
+
+def add_subject_category(name="subject_cat_1"):
+ category_template["name"] = name
+ req = requests.post(URL.format("/subject_categories"), json=category_template, headers=HEADERS)
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ assert "subject_categories" in result
+ category_id = list(result['subject_categories'].keys())[0]
+ if "result" in result:
+ assert result["result"]
+ assert "name" in result['subject_categories'][category_id]
+ assert category_template["name"] == result['subject_categories'][category_id]["name"]
+ return category_id
+
+
+def check_subject_category(category_id):
+ req = requests.get(URL.format("/subject_categories"))
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ assert "subject_categories" in result
+ if "result" in result:
+ assert result["result"]
+ assert category_id in result['subject_categories']
+ assert "name" in result['subject_categories'][category_id]
+ assert category_template["name"] == result['subject_categories'][category_id]["name"]
+
+
+def delete_subject_category(category_id):
+ req = requests.delete(URL.format("/subject_categories/{}".format(category_id)))
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ if "result" in result:
+ assert result["result"]
+
+
+def add_object_category(name="object_cat_1"):
+ category_template["name"] = name
+ req = requests.post(URL.format("/object_categories"), json=category_template, headers=HEADERS)
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ assert "object_categories" in result
+ category_id = list(result['object_categories'].keys())[0]
+ if "result" in result:
+ assert result["result"]
+ assert "name" in result['object_categories'][category_id]
+ assert category_template["name"] == result['object_categories'][category_id]["name"]
+ return category_id
+
+
+def check_object_category(category_id):
+ req = requests.get(URL.format("/object_categories"))
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ assert "object_categories" in result
+ if "result" in result:
+ assert result["result"]
+ assert category_id in result['object_categories']
+ assert "name" in result['object_categories'][category_id]
+ assert category_template["name"] == result['object_categories'][category_id]["name"]
+
+
+def delete_object_category(category_id):
+ req = requests.delete(URL.format("/object_categories/{}".format(category_id)))
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ if "result" in result:
+ assert result["result"]
+
+
+def add_action_category(name="action_cat_1"):
+ category_template["name"] = name
+ req = requests.post(URL.format("/action_categories"), json=category_template, headers=HEADERS)
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ assert "action_categories" in result
+ category_id = list(result['action_categories'].keys())[0]
+ if "result" in result:
+ assert result["result"]
+ assert "name" in result['action_categories'][category_id]
+ assert category_template["name"] == result['action_categories'][category_id]["name"]
+ return category_id
+
+
+def check_action_category(category_id):
+ req = requests.get(URL.format("/action_categories"))
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ assert "action_categories" in result
+ if "result" in result:
+ assert result["result"]
+ assert category_id in result['action_categories']
+ assert "name" in result['action_categories'][category_id]
+ assert category_template["name"] == result['action_categories'][category_id]["name"]
+
+
+def delete_action_category(category_id):
+ req = requests.delete(URL.format("/action_categories/{}".format(category_id)))
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ if "result" in result:
+ assert result["result"]
+
+
+def add_categories_and_meta_rule(name="test_meta_rule"):
+ scat_id = add_subject_category()
+ ocat_id = add_object_category()
+ acat_id = add_action_category()
+ _meta_rule_template = copy.deepcopy(meta_rule_template)
+ _meta_rule_template["name"] = name
+ _meta_rule_template["subject_categories"].append(scat_id)
+ _meta_rule_template["object_categories"].append(ocat_id)
+ _meta_rule_template["action_categories"].append(acat_id)
+ req = requests.post(URL.format("/meta_rules"), json=_meta_rule_template, headers=HEADERS)
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ assert "meta_rules" in result
+ meta_rule_id = list(result['meta_rules'].keys())[0]
+ if "result" in result:
+ assert result["result"]
+ assert "name" in result['meta_rules'][meta_rule_id]
+ assert _meta_rule_template["name"] == result['meta_rules'][meta_rule_id]["name"]
+ return meta_rule_id, scat_id, ocat_id, acat_id
+
+
+def add_meta_rule(name="test_meta_rule", scat=[], ocat=[], acat=[]):
+ _meta_rule_template = copy.deepcopy(meta_rule_template)
+ _meta_rule_template["name"] = name
+ _meta_rule_template["subject_categories"] = []
+ _meta_rule_template["subject_categories"].extend(scat)
+ _meta_rule_template["object_categories"] = []
+ _meta_rule_template["object_categories"].extend(ocat)
+ _meta_rule_template["action_categories"] = []
+ _meta_rule_template["action_categories"].extend(acat)
+ req = requests.post(URL.format("/meta_rules"), json=_meta_rule_template, headers=HEADERS)
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ assert "meta_rules" in result
+ meta_rule_id = list(result['meta_rules'].keys())[0]
+ if "result" in result:
+ assert result["result"]
+ assert "name" in result['meta_rules'][meta_rule_id]
+ assert _meta_rule_template["name"] == result['meta_rules'][meta_rule_id]["name"]
+ return meta_rule_id
+
+
+def check_meta_rule(meta_rule_id, scat_id=None, ocat_id=None, acat_id=None):
+ req = requests.get(URL.format("/meta_rules"))
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ assert "meta_rules" in result
+ if "result" in result:
+ assert result["result"]
+ assert meta_rule_id in result['meta_rules']
+ assert "name" in result['meta_rules'][meta_rule_id]
+ if scat_id:
+ assert scat_id in result['meta_rules'][meta_rule_id]["subject_categories"]
+ if ocat_id:
+ assert ocat_id in result['meta_rules'][meta_rule_id]["object_categories"]
+ if acat_id:
+ assert acat_id in result['meta_rules'][meta_rule_id]["action_categories"]
+
+
+def delete_meta_rule(meta_rule_id):
+ req = requests.delete(URL.format("/meta_rules/{}".format(meta_rule_id)))
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ if "result" in result:
+ assert result["result"]
+
+
+def add_meta_rule_to_model(model_id, meta_rule_id):
+ model = check_model(model_id)
+ meta_rule_list = model[model_id]["meta_rules"]
+ meta_rule_list.append(meta_rule_id)
+ req = requests.patch(URL.format("/models/{}".format(model_id)),
+ json={"meta_rules": meta_rule_list},
+ headers=HEADERS)
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ model_id = list(result['models'].keys())[0]
+ if "result" in result:
+ assert result["result"]
+ assert "meta_rules" in result['models'][model_id]
+ assert meta_rule_list == result['models'][model_id]["meta_rules"]
diff --git a/moonv4/moon_interface/tests/apitests/utils/pdp.py b/moonv4/moon_interface/tests/apitests/utils/pdp.py
new file mode 100644
index 00000000..81b50e38
--- /dev/null
+++ b/moonv4/moon_interface/tests/apitests/utils/pdp.py
@@ -0,0 +1,145 @@
+import requests
+
+URL = "http://172.18.0.11:38001{}"
+HEADERS = {"content-type": "application/json"}
+KEYSTONE_USER = "admin"
+KEYSTONE_PASSWORD = "p4ssw0rd"
+KEYSTONE_PROJECT = "admin"
+
+pdp_template = {
+ "name": "test_pdp",
+ "security_pipeline": [],
+ "keystone_project_id": "",
+ "description": "test",
+}
+
+
+def get_keystone_projects():
+ server = "keystone"
+
+ HEADERS = {
+ "Content-Type": "application/json"
+ }
+
+ data_auth = {
+ "auth": {
+ "identity": {
+ "methods": [
+ "password"
+ ],
+ "password": {
+ "user": {
+ "domain": {
+ "id": "Default"
+ },
+ "name": KEYSTONE_USER,
+ "password": KEYSTONE_PASSWORD
+ }
+ }
+ },
+ "scope": {
+ "project": {
+ "domain": {
+ "id": "Default"
+ },
+ "name": KEYSTONE_PROJECT
+ }
+ }
+ }
+ }
+
+ req = requests.post("http://{}:5000/v3/auth/tokens".format(server), json=data_auth, headers=HEADERS)
+
+ assert req.status_code in (200, 201)
+ TOKEN = req.headers['X-Subject-Token']
+ HEADERS['X-Auth-Token'] = TOKEN
+ req = requests.get("http://{}:5000/v3/projects".format(server), headers=HEADERS)
+ assert req.status_code in (200, 201)
+ return req.json()
+
+
+def check_pdp(pdp_id=None, keystone_project_id=None):
+ req = requests.get(URL.format("/pdp"))
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ assert "pdps" in result
+ if pdp_id:
+ assert result["pdps"]
+ assert pdp_id in result['pdps']
+ assert "name" in result['pdps'][pdp_id]
+ assert pdp_template["name"] == result['pdps'][pdp_id]["name"]
+ if keystone_project_id:
+ assert result["pdps"]
+ assert pdp_id in result['pdps']
+ assert "keystone_project_id" in result['pdps'][pdp_id]
+ assert keystone_project_id == result['pdps'][pdp_id]["keystone_project_id"]
+ return result
+
+
+def add_pdp(name="test_pdp", policy_id=None):
+ pdp_template['name'] = name
+ if policy_id:
+ pdp_template['security_pipeline'].append(policy_id)
+ req = requests.post(URL.format("/pdp"), json=pdp_template, headers=HEADERS)
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ pdp_id = list(result['pdps'].keys())[0]
+ if "result" in result:
+ assert result["result"]
+ assert "name" in result['pdps'][pdp_id]
+ assert pdp_template["name"] == result['pdps'][pdp_id]["name"]
+ return pdp_id
+
+
+def update_pdp(pdp_id, policy_id=None):
+ req = requests.get(URL.format("/pdp/{}".format(pdp_id)))
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ assert "pdps" in result
+ assert pdp_id in result['pdps']
+ pipeline = result['pdps'][pdp_id]["security_pipeline"]
+ if policy_id not in pipeline:
+ pipeline.append(policy_id)
+ req = requests.patch(URL.format("/pdp/{}".format(pdp_id)),
+ json={"security_pipeline": pipeline})
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ assert "pdps" in result
+ assert pdp_id in result['pdps']
+
+ req = requests.get(URL.format("/pdp/{}".format(pdp_id)))
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ assert "pdps" in result
+ assert pdp_id in result['pdps']
+ assert policy_id in pipeline
+
+
+def map_to_keystone(pdp_id, keystone_project_id):
+ req = requests.patch(URL.format("/pdp/{}".format(pdp_id)), json={"keystone_project_id": keystone_project_id},
+ headers=HEADERS)
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ if "result" in result:
+ assert result["result"]
+ assert pdp_id in result['pdps']
+ assert "name" in result['pdps'][pdp_id]
+ assert pdp_template["name"] == result['pdps'][pdp_id]["name"]
+ return pdp_id
+
+
+def delete_pdp(pdp_id):
+ req = requests.delete(URL.format("/pdp/{}".format(pdp_id)))
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ assert "result" in result
+ assert result["result"]
+
+
diff --git a/moonv4/moon_interface/tests/apitests/utils/policies.py b/moonv4/moon_interface/tests/apitests/utils/policies.py
new file mode 100644
index 00000000..67be91e5
--- /dev/null
+++ b/moonv4/moon_interface/tests/apitests/utils/policies.py
@@ -0,0 +1,581 @@
+import requests
+
+URL = "http://172.18.0.11:38001{}"
+HEADERS = {"content-type": "application/json"}
+FILE = open("/tmp/test.log", "w")
+
+policy_template = {
+ "name": "test_policy",
+ "model_id": "",
+ "genre": "authz",
+ "description": "test",
+}
+
+subject_template = {
+ "name": "test_subject",
+ "description": "test",
+ "email": "mail",
+ "password": "my_pass",
+}
+
+object_template = {
+ "name": "test_subject",
+ "description": "test"
+}
+
+action_template = {
+ "name": "test_subject",
+ "description": "test"
+}
+
+subject_data_template = {
+ "name": "subject_data1",
+ "description": "description of the data subject"
+}
+
+object_data_template = {
+ "name": "object_data1",
+ "description": "description of the data subject"
+}
+
+action_data_template = {
+ "name": "action_data1",
+ "description": "description of the data subject"
+}
+
+subject_assignment_template = {
+ "id": "",
+ "category_id": "",
+ "scope_id": ""
+}
+
+
+def check_policy(policy_id=None):
+ req = requests.get(URL.format("/policies"))
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ assert "policies" in result
+ if policy_id:
+ assert result["policies"]
+ assert policy_id in result['policies']
+ assert "name" in result['policies'][policy_id]
+ assert policy_template["name"] == result['policies'][policy_id]["name"]
+
+
+def add_policy(name="test_policy"):
+ policy_template["name"] = name
+ req = requests.post(URL.format("/policies"), json=policy_template, headers=HEADERS)
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ policy_id = list(result['policies'].keys())[0]
+ if "result" in result:
+ assert result["result"]
+ assert "name" in result['policies'][policy_id]
+ assert policy_template["name"] == result['policies'][policy_id]["name"]
+ return policy_id
+
+
+def update_policy(policy_id, model_id):
+ req = requests.patch(URL.format("/policies/{}".format(policy_id)),
+ json={"model_id": model_id}, headers=HEADERS)
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ policy_id = list(result['policies'].keys())[0]
+ if "result" in result:
+ assert result["result"]
+ assert "model_id" in result['policies'][policy_id]
+ assert model_id == result['policies'][policy_id]["model_id"]
+
+
+def delete_policy(policy_id):
+ req = requests.delete(URL.format("/policies/{}".format(policy_id)))
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ assert "result" in result
+ assert result["result"]
+
+
+def add_subject(policy_id=None, name="test_subject"):
+ subject_template['name'] = name
+ if policy_id:
+ req = requests.post(URL.format("/policies/{}/subjects".format(policy_id)),
+ json=subject_template, headers=HEADERS)
+ else:
+ req = requests.post(URL.format("/subjects"), json=subject_template, headers=HEADERS)
+ assert req.status_code == 200
+ result = req.json()
+ assert "subjects" in result
+ subject_id = list(result['subjects'].keys())[0]
+ return subject_id
+
+
+def update_subject(subject_id, policy_id=None, description=None):
+ if policy_id and not description:
+ req = requests.patch(URL.format("/policies/{}/subjects/{}".format(policy_id, subject_id)),
+ json={})
+ elif policy_id and description:
+ req = requests.patch(URL.format("/policies/{}/subjects/{}".format(policy_id, subject_id)),
+ json={"description": description})
+ else:
+ req = requests.patch(URL.format("/subjects/{}".format(subject_id)),
+ json={"description": description})
+ assert req.status_code == 200
+ result = req.json()
+ assert "subjects" in result
+ assert "name" in result["subjects"][subject_id]
+ assert subject_template["name"] == result["subjects"][subject_id]["name"]
+ assert "policy_list" in result["subjects"][subject_id]
+ if policy_id:
+ assert policy_id in result["subjects"][subject_id]["policy_list"]
+ if description:
+ assert description in result["subjects"][subject_id]["description"]
+
+
+def check_subject(subject_id=None, policy_id=None):
+ if policy_id:
+ req = requests.get(URL.format("/policies/{}/subjects".format(policy_id)))
+ else:
+ req = requests.get(URL.format("/subjects"))
+ assert req.status_code == 200
+ result = req.json()
+ assert "subjects" in result
+ assert "name" in result["subjects"][subject_id]
+ assert subject_template["name"] == result["subjects"][subject_id]["name"]
+ if policy_id:
+ assert "policy_list" in result["subjects"][subject_id]
+ assert policy_id in result["subjects"][subject_id]["policy_list"]
+
+
+def delete_subject(subject_id, policy_id=None):
+ if policy_id:
+ req = requests.delete(URL.format("/policies/{}/subjects/{}".format(policy_id, subject_id)))
+ else:
+ req = requests.delete(URL.format("/subjects/{}".format(subject_id)))
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ assert "result" in result
+ assert result["result"]
+
+ if policy_id:
+ req = requests.get(URL.format("/policies/{}/subjects".format(policy_id)))
+ else:
+ req = requests.get(URL.format("/subjects"))
+ assert req.status_code == 200
+ result = req.json()
+ assert "subjects" in result
+ if subject_id in result["subjects"]:
+ assert "name" in result["subjects"][subject_id]
+ assert subject_template["name"] == result["subjects"][subject_id]["name"]
+ if policy_id:
+ assert "policy_list" in result["subjects"][subject_id]
+ assert policy_id not in result["subjects"][subject_id]["policy_list"]
+
+
+def add_object(policy_id=None, name="test_object"):
+ object_template['name'] = name
+ if policy_id:
+ req = requests.post(URL.format("/policies/{}/objects".format(policy_id)),
+ json=object_template, headers=HEADERS)
+ else:
+ req = requests.post(URL.format("/objects"), json=object_template, headers=HEADERS)
+ assert req.status_code == 200
+ result = req.json()
+ assert "objects" in result
+ object_id = list(result['objects'].keys())[0]
+ return object_id
+
+
+def update_object(object_id, policy_id):
+ req = requests.patch(URL.format("/policies/{}/objects/{}".format(policy_id, object_id)), json={})
+ assert req.status_code == 200
+ result = req.json()
+ assert "objects" in result
+ assert "name" in result["objects"][object_id]
+ assert object_template["name"] == result["objects"][object_id]["name"]
+ assert "policy_list" in result["objects"][object_id]
+ assert policy_id in result["objects"][object_id]["policy_list"]
+
+
+def check_object(object_id=None, policy_id=None):
+ if policy_id:
+ req = requests.get(URL.format("/policies/{}/objects".format(policy_id)))
+ else:
+ req = requests.get(URL.format("/objects"))
+ assert req.status_code == 200
+ result = req.json()
+ assert "objects" in result
+ assert "name" in result["objects"][object_id]
+ assert object_template["name"] == result["objects"][object_id]["name"]
+ if policy_id:
+ assert "policy_list" in result["objects"][object_id]
+ assert policy_id in result["objects"][object_id]["policy_list"]
+
+
+def delete_object(object_id, policy_id=None):
+ if policy_id:
+ req = requests.delete(URL.format("/policies/{}/objects/{}".format(policy_id, object_id)))
+ else:
+ req = requests.delete(URL.format("/objects/{}".format(object_id)))
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ assert "result" in result
+ assert result["result"]
+
+ if policy_id:
+ req = requests.get(URL.format("/policies/{}/objects".format(policy_id)))
+ else:
+ req = requests.get(URL.format("/objects"))
+ assert req.status_code == 200
+ result = req.json()
+ assert "objects" in result
+ if object_id in result["objects"]:
+ assert "name" in result["objects"][object_id]
+ assert object_template["name"] == result["objects"][object_id]["name"]
+ if policy_id:
+ assert "policy_list" in result["objects"][object_id]
+ assert policy_id not in result["objects"][object_id]["policy_list"]
+
+
+def add_action(policy_id=None, name="test_action"):
+ action_template['name'] = name
+ if policy_id:
+ req = requests.post(URL.format("/policies/{}/actions".format(policy_id)),
+ json=action_template, headers=HEADERS)
+ else:
+ req = requests.post(URL.format("/actions"), json=action_template, headers=HEADERS)
+ assert req.status_code == 200
+ result = req.json()
+ assert "actions" in result
+ action_id = list(result['actions'].keys())[0]
+ return action_id
+
+
+def update_action(action_id, policy_id):
+ req = requests.patch(URL.format("/policies/{}/actions/{}".format(policy_id, action_id)), json={})
+ assert req.status_code == 200
+ result = req.json()
+ assert "actions" in result
+ assert "name" in result["actions"][action_id]
+ assert action_template["name"] == result["actions"][action_id]["name"]
+ assert "policy_list" in result["actions"][action_id]
+ assert policy_id in result["actions"][action_id]["policy_list"]
+
+
+def check_action(action_id=None, policy_id=None):
+ if policy_id:
+ req = requests.get(URL.format("/policies/{}/actions".format(policy_id)))
+ else:
+ req = requests.get(URL.format("/actions"))
+ assert req.status_code == 200
+ result = req.json()
+ assert "actions" in result
+ assert "name" in result["actions"][action_id]
+ assert action_template["name"] == result["actions"][action_id]["name"]
+ if policy_id:
+ assert "policy_list" in result["actions"][action_id]
+ assert policy_id in result["actions"][action_id]["policy_list"]
+
+
+def delete_action(action_id, policy_id=None):
+ if policy_id:
+ req = requests.delete(URL.format("/policies/{}/actions/{}".format(policy_id, action_id)))
+ else:
+ req = requests.delete(URL.format("/actions/{}".format(action_id)))
+ assert req.status_code == 200
+ result = req.json()
+ assert type(result) is dict
+ assert "result" in result
+ assert result["result"]
+
+ if policy_id:
+ req = requests.get(URL.format("/policies/{}/actions".format(policy_id)))
+ else:
+ req = requests.get(URL.format("/actions"))
+ assert req.status_code == 200
+ result = req.json()
+ assert "actions" in result
+ if action_id in result["actions"]:
+ assert "name" in result["actions"][action_id]
+ assert action_template["name"] == result["actions"][action_id]["name"]
+ if policy_id:
+ assert "policy_list" in result["actions"][action_id]
+ assert policy_id not in result["actions"][action_id]["policy_list"]
+
+
+def add_subject_data(policy_id, category_id, name="subject_data1"):
+ subject_data_template['name'] = name
+ req = requests.post(URL.format("/policies/{}/subject_data/{}".format(policy_id, category_id)),
+ json=subject_data_template, headers=HEADERS)
+ assert req.status_code == 200
+ result = req.json()
+ assert "subject_data" in result
+ subject_id = list(result['subject_data']['data'].keys())[0]
+ return subject_id
+
+
+def check_subject_data(policy_id, data_id, category_id):
+ req = requests.get(URL.format("/policies/{}/subject_data/{}".format(policy_id, category_id)))
+ assert req.status_code == 200
+ result = req.json()
+ assert "subject_data" in result
+ for _data in result['subject_data']:
+ assert data_id in list(_data['data'].keys())
+ assert category_id == _data["category_id"]
+
+
+def add_object_data(policy_id, category_id, name="object_data1"):
+ object_data_template['name'] = name
+ req = requests.post(URL.format("/policies/{}/object_data/{}".format(policy_id, category_id)),
+ json=object_data_template, headers=HEADERS)
+ assert req.status_code == 200
+ result = req.json()
+ assert "object_data" in result
+ object_id = list(result['object_data']['data'].keys())[0]
+ return object_id
+
+
+def check_object_data(policy_id, data_id, category_id):
+ req = requests.get(URL.format("/policies/{}/object_data/{}".format(policy_id, category_id)))
+ assert req.status_code == 200
+ result = req.json()
+ assert "object_data" in result
+ for _data in result['object_data']:
+ assert data_id in list(_data['data'].keys())
+ assert category_id == _data["category_id"]
+
+
+def add_action_data(policy_id, category_id, name="action_data1"):
+ action_data_template['name'] = name
+ req = requests.post(URL.format("/policies/{}/action_data/{}".format(policy_id, category_id)),
+ json=action_data_template, headers=HEADERS)
+ assert req.status_code == 200
+ result = req.json()
+ assert "action_data" in result
+ action_id = list(result['action_data']['data'].keys())[0]
+ return action_id
+
+
+def check_action_data(policy_id, data_id, category_id):
+ req = requests.get(URL.format("/policies/{}/action_data/{}".format(policy_id, category_id)))
+ assert req.status_code == 200
+ result = req.json()
+ assert "action_data" in result
+ for _data in result['action_data']:
+ assert data_id in list(_data['data'].keys())
+ assert category_id == _data["category_id"]
+
+
+def add_subject_assignments(policy_id, subject_id, subject_cat_id, subject_data_id):
+ req = requests.post(URL.format("/policies/{}/subject_assignments".format(policy_id)),
+ json={
+ "id": subject_id,
+ "category_id": subject_cat_id,
+ "data_id": subject_data_id
+ }, headers=HEADERS)
+ assert req.status_code == 200
+ result = req.json()
+ assert "subject_assignments" in result
+ assert result["subject_assignments"]
+
+
+def check_subject_assignments(policy_id, subject_id, subject_cat_id, subject_data_id):
+ req = requests.get(URL.format("/policies/{}/subject_assignments/{}/{}/{}".format(
+ policy_id, subject_id, subject_cat_id, subject_data_id)))
+ assert req.status_code == 200
+ result = req.json()
+ assert "subject_assignments" in result
+ assert result["subject_assignments"]
+ for key in result["subject_assignments"]:
+ assert "subject_id" in result["subject_assignments"][key]
+ assert "category_id" in result["subject_assignments"][key]
+ assert "assignments" in result["subject_assignments"][key]
+ if result["subject_assignments"][key]['subject_id'] == subject_id and \
+ result["subject_assignments"][key]["category_id"] == subject_cat_id:
+ assert subject_data_id in result["subject_assignments"][key]["assignments"]
+
+
+def check_object_assignments(policy_id, object_id, object_cat_id, object_data_id):
+ req = requests.get(URL.format("/policies/{}/object_assignments/{}/{}/{}".format(
+ policy_id, object_id, object_cat_id, object_data_id)))
+ assert req.status_code == 200
+ result = req.json()
+ assert "object_assignments" in result
+ assert result["object_assignments"]
+ for key in result["object_assignments"]:
+ assert "object_id" in result["object_assignments"][key]
+ assert "category_id" in result["object_assignments"][key]
+ assert "assignments" in result["object_assignments"][key]
+ if result["object_assignments"][key]['object_id'] == object_id and \
+ result["object_assignments"][key]["category_id"] == object_cat_id:
+ assert object_data_id in result["object_assignments"][key]["assignments"]
+
+
+def check_action_assignments(policy_id, action_id, action_cat_id, action_data_id):
+ req = requests.get(URL.format("/policies/{}/action_assignments/{}/{}/{}".format(
+ policy_id, action_id, action_cat_id, action_data_id)))
+ assert req.status_code == 200
+ result = req.json()
+ assert "action_assignments" in result
+ assert result["action_assignments"]
+ for key in result["action_assignments"]:
+ assert "action_id" in result["action_assignments"][key]
+ assert "category_id" in result["action_assignments"][key]
+ assert "assignments" in result["action_assignments"][key]
+ if result["action_assignments"][key]['action_id'] == action_id and \
+ result["action_assignments"][key]["category_id"] == action_cat_id:
+ assert action_data_id in result["action_assignments"][key]["assignments"]
+
+
+def add_object_assignments(policy_id, object_id, object_cat_id, object_data_id):
+ req = requests.post(URL.format("/policies/{}/object_assignments".format(policy_id)),
+ json={
+ "id": object_id,
+ "category_id": object_cat_id,
+ "data_id": object_data_id
+ }, headers=HEADERS)
+ assert req.status_code == 200
+ result = req.json()
+ assert "object_assignments" in result
+ assert result["object_assignments"]
+
+
+def add_action_assignments(policy_id, action_id, action_cat_id, action_data_id):
+ req = requests.post(URL.format("/policies/{}/action_assignments".format(policy_id)),
+ json={
+ "id": action_id,
+ "category_id": action_cat_id,
+ "data_id": action_data_id
+ }, headers=HEADERS)
+ assert req.status_code == 200
+ result = req.json()
+ assert "action_assignments" in result
+ assert result["action_assignments"]
+
+
+def delete_subject_assignment(policy_id, subject_id, subject_cat_id, subject_data_id):
+ req = requests.delete(URL.format("/policies/{}/subject_assignments/{}/{}/{}".format(
+ policy_id, subject_id, subject_cat_id, subject_data_id)))
+ assert req.status_code == 200
+ result = req.json()
+ assert "result" in result
+ assert result["result"]
+
+ req = requests.get(URL.format("/policies/{}/subject_assignments/{}/{}/{}".format(
+ policy_id, subject_id, subject_cat_id, subject_data_id)))
+ assert req.status_code == 200
+ result = req.json()
+ assert "subject_assignments" in result
+ assert result["subject_assignments"]
+ for key in result["subject_assignments"]:
+ assert "subject_id" in result["subject_assignments"][key]
+ assert "category_id" in result["subject_assignments"][key]
+ assert "assignments" in result["subject_assignments"][key]
+ if result["subject_assignments"][key]['subject_id'] == subject_id and \
+ result["subject_assignments"][key]["category_id"] == subject_cat_id:
+ assert subject_data_id not in result["subject_assignments"][key]["assignments"]
+
+
+def delete_object_assignment(policy_id, object_id, object_cat_id, object_data_id):
+ req = requests.delete(URL.format("/policies/{}/object_assignments/{}/{}/{}".format(
+ policy_id, object_id, object_cat_id, object_data_id)))
+ assert req.status_code == 200
+ result = req.json()
+ assert "result" in result
+ assert result["result"]
+
+ req = requests.get(URL.format("/policies/{}/object_assignments/{}/{}/{}".format(
+ policy_id, object_id, object_cat_id, object_data_id)))
+ assert req.status_code == 200
+ result = req.json()
+ assert "object_assignments" in result
+ assert result["object_assignments"]
+ for key in result["object_assignments"]:
+ assert "object_id" in result["object_assignments"][key]
+ assert "category_id" in result["object_assignments"][key]
+ assert "assignments" in result["object_assignments"][key]
+ if result["object_assignments"][key]['object_id'] == object_id and \
+ result["object_assignments"][key]["category_id"] == object_cat_id:
+ assert object_data_id not in result["object_assignments"][key]["assignments"]
+
+
+def delete_action_assignment(policy_id, action_id, action_cat_id, action_data_id):
+ req = requests.delete(URL.format("/policies/{}/action_assignments/{}/{}/{}".format(
+ policy_id, action_id, action_cat_id, action_data_id)))
+ assert req.status_code == 200
+ result = req.json()
+ assert "result" in result
+ assert result["result"]
+
+ req = requests.get(URL.format("/policies/{}/action_assignments/{}/{}/{}".format(
+ policy_id, action_id, action_cat_id, action_data_id)))
+ assert req.status_code == 200
+ result = req.json()
+ assert "action_assignments" in result
+ assert result["action_assignments"]
+ for key in result["action_assignments"]:
+ assert "action_id" in result["action_assignments"][key]
+ assert "category_id" in result["action_assignments"][key]
+ assert "assignments" in result["action_assignments"][key]
+ if result["action_assignments"][key]['action_id'] == action_id and \
+ result["action_assignments"][key]["category_id"] == action_cat_id:
+ assert action_data_id not in result["action_assignments"][key]["assignments"]
+
+
+def add_rule(policy_id, meta_rule_id, rule):
+ req = requests.post(URL.format("/policies/{}/rules".format(policy_id)),
+ json={
+ "meta_rule_id": meta_rule_id,
+ "rule": rule,
+ "enabled": True
+ },
+ headers=HEADERS)
+ assert req.status_code == 200
+ result = req.json()
+ assert "rules" in result
+ rule_id = list(result["rules"].keys())[0]
+ assert "policy_id" in result["rules"][rule_id]
+ assert policy_id == result["rules"][rule_id]["policy_id"]
+ assert "meta_rule_id" in result["rules"][rule_id]
+ assert meta_rule_id == result["rules"][rule_id]["meta_rule_id"]
+ assert rule == result["rules"][rule_id]["rule"]
+ return rule_id
+
+
+def check_rule(policy_id, meta_rule_id, rule_id, rule):
+ req = requests.get(URL.format("/policies/{}/rules".format(policy_id)))
+ assert req.status_code == 200
+ result = req.json()
+ assert "rules" in result
+ assert "policy_id" in result["rules"]
+ assert policy_id == result["rules"]["policy_id"]
+ for item in result["rules"]["rules"]:
+ assert "meta_rule_id" in item
+ if meta_rule_id == item["meta_rule_id"]:
+ if rule_id == item["id"]:
+ assert rule == item["rule"]
+
+
+def delete_rule(policy_id, rule_id):
+ req = requests.delete(URL.format("/policies/{}/rules/{}".format(policy_id, rule_id)))
+ assert req.status_code == 200
+ result = req.json()
+ assert "result" in result
+ assert result["result"]
+
+ req = requests.get(URL.format("/policies/{}/rules".format(policy_id)))
+ assert req.status_code == 200
+ result = req.json()
+ assert "rules" in result
+ assert "policy_id" in result["rules"]
+ assert policy_id == result["rules"]["policy_id"]
+ found_rule = False
+ for item in result["rules"]["rules"]:
+ if rule_id == item["id"]:
+ found_rule = True
+ assert not found_rule