aboutsummaryrefslogtreecommitdiffstats
path: root/python_moonutilities/tests/unit_python/test_cache.py
diff options
context:
space:
mode:
Diffstat (limited to 'python_moonutilities/tests/unit_python/test_cache.py')
-rw-r--r--python_moonutilities/tests/unit_python/test_cache.py189
1 files changed, 173 insertions, 16 deletions
diff --git a/python_moonutilities/tests/unit_python/test_cache.py b/python_moonutilities/tests/unit_python/test_cache.py
index c479395b..db1e3ae7 100644
--- a/python_moonutilities/tests/unit_python/test_cache.py
+++ b/python_moonutilities/tests/unit_python/test_cache.py
@@ -1,4 +1,5 @@
import pytest
+import mock_repo.data as data_mock
def test_authz_request():
@@ -7,63 +8,219 @@ def test_authz_request():
assert isinstance(c.authz_requests, dict)
+# tests for get (subject, object, action) in cache
+# ================================================
def test_get_subject_success():
from python_moonutilities import cache
cache_obj = cache.Cache()
- policy_id = 'policy_id_1'
name = 'subject_name'
- subject_id = cache_obj.get_subject(policy_id, name)
+ subject_id = cache_obj.get_subject(data_mock.shared_ids["policy"]["policy_id_1"], name)
assert subject_id is not None
-def test_get_subject_failure():
+def test_get_subject_not_found():
from python_moonutilities import cache
- cache_obj = cache.Cache()
- policy_id = 'policy_id_1'
+ cache_obj2 = cache.Cache()
name = 'invalid name'
with pytest.raises(Exception) as exception_info:
- cache_obj.get_subject(policy_id, name)
+ cache_obj2.get_subject(data_mock.shared_ids["policy"]["policy_id_1"], name)
assert str(exception_info.value) == '400: Subject Unknown'
+# [TODO] this test used to test the invalid response
+# it should be un commented and run after refactoring the related part
+def test_get_subject_invalid_response():
+ from python_moonutilities import cache
+ cache_obj2 = cache.Cache()
+ # policy_id = 'policy_id_invalid_response'
+ name = 'invalid name'
+
+
+# with pytest.raises(Exception) as exception_info:
+# cache_obj2.get_subject(data_mock.shared_ids["policy"]["policy_id_invalid_response"], name)
+# assert str(exception_info.value) == '400: Subject Unknown'
+
+
def test_get_object_success():
from python_moonutilities import cache
cache_obj = cache.Cache()
- policy_id = 'policy_id_1'
name = 'object_name'
- object_id = cache_obj.get_object(policy_id, name)
+ object_id = cache_obj.get_object(data_mock.shared_ids["policy"]["policy_id_1"], name)
assert object_id is not None
def test_get_object_failure():
from python_moonutilities import cache
cache_obj = cache.Cache()
- policy_id = 'policy_id_1'
name = 'invalid name'
with pytest.raises(Exception) as exception_info:
- cache_obj.get_object(policy_id, name)
+ cache_obj.get_object(data_mock.shared_ids["policy"]["policy_id_1"], name)
assert str(exception_info.value) == '400: Subject Unknown'
def test_get_action_success():
from python_moonutilities import cache
cache_obj = cache.Cache()
- policy_id = 'policy_id_1'
name = 'action_name'
- action_id = cache_obj.get_action(policy_id, name)
+ action_id = cache_obj.get_action(data_mock.shared_ids["policy"]["policy_id_1"], name)
assert action_id is not None
def test_get_action_failure():
from python_moonutilities import cache
cache_obj = cache.Cache()
- policy_id = 'policy_id_1'
name = 'invalid name'
with pytest.raises(Exception) as exception_info:
- cache_obj.get_action(policy_id, name)
+ cache_obj.get_action(data_mock.shared_ids["policy"]["policy_id_1"], name)
assert str(exception_info.value) == '400: Subject Unknown'
+# ====================================================================================================
+
+# tests for get (subject_assignment, object_assignment, action_assignment) in cache
+# =================================================================================
+
+def test_get_subject_assignment_success():
+ from python_moonutilities import cache
+ cache_obj = cache.Cache()
+ subject_assignments = cache_obj.get_subject_assignments(data_mock.shared_ids["policy"]["policy_id_1"],
+ data_mock.shared_ids["perimeter"]["perimeter_id_1"],
+ data_mock.shared_ids["category"]["category_id_1"])
+ assert subject_assignments is not None
+
+
+def test_get_subject_assignment_failure():
+ from python_moonutilities import cache
+ cache_obj = cache.Cache()
+ subject_assignments = cache_obj.get_subject_assignments(data_mock.shared_ids["policy"]["policy_id_2"],
+ '',
+ data_mock.shared_ids["category"]["category_id_1"])
+ assert len(subject_assignments) == 0
+
+
+def test_get_subject_assignment_invalid_category_failure():
+ from python_moonutilities import cache
+ cache_obj = cache.Cache()
+ subject_assignments = cache_obj.get_subject_assignments(data_mock.shared_ids["policy"]["policy_id_1"],
+ data_mock.shared_ids["perimeter"]["perimeter_id_1"],
+ data_mock.shared_ids["category"]["invalid_category_id_1"])
+ assert len(subject_assignments) == 0
+
+
+def test_get_object_assignment_success():
+ from python_moonutilities import cache
+ cache_obj = cache.Cache()
+ object_assignments = cache_obj.get_object_assignments(data_mock.shared_ids["policy"]["policy_id_1"],
+ data_mock.shared_ids["perimeter"]["perimeter_id_2"],
+ data_mock.shared_ids["category"]["category_id_1"])
+ assert object_assignments is not None
+
+
+def test_get_object_assignment_failure():
+ from python_moonutilities import cache
+ cache_obj = cache.Cache()
+ object_assignments = cache_obj.get_object_assignments(data_mock.shared_ids["policy"]["policy_id_2"],
+ '',
+ data_mock.shared_ids["category"]["category_id_1"])
+ assert len(object_assignments) == 0
+
+
+def test_get_object_assignment_invalid_category_failure():
+ from python_moonutilities import cache
+ cache_obj = cache.Cache()
+ object_assignments = cache_obj.get_object_assignments(data_mock.shared_ids["policy"]["policy_id_1"],
+ data_mock.shared_ids["perimeter"]["perimeter_id_1"],
+ data_mock.shared_ids["category"]["invalid_category_id_1"])
+ assert len(object_assignments) == 0
+
+
+def test_get_action_assignment_success():
+ from python_moonutilities import cache
+ cache_obj = cache.Cache()
+ action_assignments = cache_obj.get_action_assignments(data_mock.shared_ids["policy"]["policy_id_1"],
+ data_mock.shared_ids["perimeter"]["perimeter_id_3"],
+ data_mock.shared_ids["category"]["category_id_1"])
+ assert action_assignments is not None
+
+
+def test_get_action_assignment_failure():
+ from python_moonutilities import cache
+ cache_obj = cache.Cache()
+ action_assignments = cache_obj.get_action_assignments(data_mock.shared_ids["policy"]["policy_id_2"],
+ '',
+ data_mock.shared_ids["category"]["category_id_1"])
+ assert len(action_assignments) == 0
+
+
+def test_get_action_assignment_invalid_category_failure():
+ from python_moonutilities import cache
+ cache_obj = cache.Cache()
+ action_assignments = cache_obj.get_action_assignments(data_mock.shared_ids["policy"]["policy_id_1"],
+ data_mock.shared_ids["perimeter"]["perimeter_id_1"],
+ data_mock.shared_ids["category"]["invalid_category_id_1"])
+ assert len(action_assignments) == 0
+
+
+# ====================================================================================================
+
+# tests for helper function in cache
+# ==================================
+def test_get_policy_from_meta_rules_success():
+ from python_moonutilities import cache
+ cache_obj = cache.Cache()
+ policy_id = cache_obj.get_policy_from_meta_rules(data_mock.shared_ids["meta_rule"]["meta_rule_id_1"])
+ assert policy_id is not None
+
+
+# def test_get_policy_from_meta_rules_failure():
+# from python_moonutilities import cache
+# cache_obj = cache.Cache()
+# meta_rule_id = 'meta_rule_id3'
+# policy_id = cache_obj.get_policy_from_meta_rules(meta_rule_id)
+# assert policy_id is None
+
+
+def test_get_pdp_from_keystone_project_success():
+ from python_moonutilities import cache
+ cache_obj = cache.Cache()
+ keystone_project_id = 'keystone_project_id1'
+ pdp_key = cache_obj.get_pdp_from_keystone_project(keystone_project_id)
+ assert pdp_key is not None
+
+
+def test_get_pdp_from_keystone_project_failure():
+ from python_moonutilities import cache
+ cache_obj = cache.Cache()
+ keystone_project_id = 'keystone_project_id2'
+ pdp_key = cache_obj.get_pdp_from_keystone_project(keystone_project_id)
+ assert pdp_key is None
+
+
+def test_get_keystone_project_id_from_policy_id_success():
+ from python_moonutilities import cache
+ cache_obj = cache.Cache()
+ keystone_project_id = cache_obj.get_keystone_project_id_from_policy_id(
+ data_mock.shared_ids["policy"]["policy_id_1"])
+ assert keystone_project_id is not None
+
+
+def test_get_keystone_project_id_from_policy_id_failure():
+ from python_moonutilities import cache
+ cache_obj = cache.Cache()
+ policy_id = 'policy_id_3'
+ keystone_project_id = cache_obj.get_keystone_project_id_from_policy_id(policy_id)
+ assert keystone_project_id is None
+
+
+# def test_get_containers_from_keystone_project_id_success():
+# from python_moonutilities import cache
+# cache_obj = cache.Cache()
+# keystone_project_id = 1
+# meta_rule_id = 'meta_rule_id1'
+# container_id, container_value = cache_obj.get_containers_from_keystone_project_id(keystone_project_id, meta_rule_id)
+# assert container_id, container_value is not None
+
+
def test_cache_manager():
from python_moonutilities import cache
cache_obj = cache.Cache()
@@ -71,5 +228,5 @@ def test_cache_manager():
assert cache_obj.meta_rules is not None
assert len(cache_obj.meta_rules) == 2
assert cache_obj.policies is not None
- assert len(cache_obj.policies) == 2
- assert cache_obj.models is not None \ No newline at end of file
+ assert len(cache_obj.policies) == 1
+ assert cache_obj.models is not None