adding unit tests for cache module

Change-Id: Ie6837e63ca652374ce7b5a304d7352b3d6ddba58
Signed-off-by: sgdt6900 <rhanafy.ext@orange.com>

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