diff options
Diffstat (limited to 'moon_utilities/tests/unit_python/helpers')
4 files changed, 457 insertions, 0 deletions
diff --git a/moon_utilities/tests/unit_python/helpers/__init__.py b/moon_utilities/tests/unit_python/helpers/__init__.py new file mode 100644 index 00000000..2a5459c2 --- /dev/null +++ b/moon_utilities/tests/unit_python/helpers/__init__.py @@ -0,0 +1,13 @@ +# Software Name: MOON + +# Version: 5.4 + +# SPDX-FileCopyrightText: Copyright (c) 2018-2020 Orange and its contributors +# SPDX-License-Identifier: Apache-2.0 + +# This software is distributed under the 'Apache License 2.0', +# the text of which is available at 'http://www.apache.org/licenses/LICENSE-2.0.txt' +# or see the "LICENSE" file for more details. + + + diff --git a/moon_utilities/tests/unit_python/helpers/import_export_cache_helper.py b/moon_utilities/tests/unit_python/helpers/import_export_cache_helper.py new file mode 100644 index 00000000..ff22abe2 --- /dev/null +++ b/moon_utilities/tests/unit_python/helpers/import_export_cache_helper.py @@ -0,0 +1,234 @@ +# Software Name: MOON + +# Version: 5.4 + +# SPDX-FileCopyrightText: Copyright (c) 2018-2020 Orange and its contributors +# SPDX-License-Identifier: Apache-2.0 + +# This software is distributed under the 'Apache License 2.0', +# the text of which is available at 'http://www.apache.org/licenses/LICENSE-2.0.txt' +# or see the "LICENSE" file for more details. + + +import logging + +logger = logging.getLogger("moon.manager.test.api." + __name__) + + +def clean_models(): + from moon_cache.cache import Cache + _cache = Cache.getInstance() + keys = list(_cache.models.keys()) + for key in keys: + _cache.delete_model(key) + + +def clean_policies(): + from moon_cache.cache import Cache + _cache = Cache.getInstance() + keys = list(_cache.policies.keys()) + for key in keys: + _cache.delete_policy(key) + + +def clean_subjects(): + from moon_cache.cache import Cache + _cache = Cache.getInstance() + + policy_keys = list(_cache.policies.keys()) + subjects = _cache.subjects + for policy_key in policy_keys: + for key in subjects: + try: + _cache.delete_subject(policy_key, key) + except AttributeError: + pass + _cache.delete_subject() + + +def clean_objects(): + from moon_cache.cache import Cache + _cache = Cache.getInstance() + + policy_keys = list(_cache.policies.keys()) + objects = _cache.objects + for policy_key in policy_keys: + for key in objects: + try: + _cache.delete_object(policy_key, key) + except AttributeError: + pass + _cache.delete_object() + + +def clean_actions(): + from moon_cache.cache import Cache + _cache = Cache.getInstance() + + policy_keys = list(_cache.policies.keys()) + actions = _cache.actions + for policy_key in policy_keys: + for key in actions: + try: + _cache.delete_action(policy_key, key) + except AttributeError: + pass + _cache.delete_action() + + +def clean_subject_categories(): + from moon_cache.cache import Cache + _cache = Cache.getInstance() + + categories = list(_cache.subject_categories.keys()) + for key in categories: + _cache.delete_subject_category(key) + + +def clean_object_categories(): + from moon_cache.cache import Cache + _cache = Cache.getInstance() + + categories = list(_cache.object_categories.keys()) + for key in categories: + _cache.delete_object_category(key) + + +def clean_action_categories(): + from moon_cache.cache import Cache + _cache = Cache.getInstance() + + categories = list(_cache.action_categories.keys()) + for key in categories: + _cache.delete_action_category(key) + + +def clean_subject_data(): + from moon_cache.cache import Cache + _cache = Cache.getInstance() + _cache.delete_subject_data() + + +def clean_object_data(): + from moon_cache.cache import Cache + _cache = Cache.getInstance() + _cache.delete_object_data() + + +def clean_action_data(): + from moon_cache.cache import Cache + _cache = Cache.getInstance() + _cache.delete_action_data() + + +def clean_meta_rule(): + from moon_cache.cache import Cache + _cache = Cache.getInstance() + + categories = list(_cache.meta_rules.keys()) + for key in categories: + _cache.delete_meta_rule(key) + + +def clean_subject_assignments(): + from moon_cache.cache import Cache + _cache = Cache.getInstance() + + if not _cache.subject_assignments: + return + for policy_key in _cache.policies.keys(): + if policy_key not in _cache.subject_assignments: + continue + for key in _cache.subject_assignments[policy_key]: + # policy_key = _cache.subject_assignments[policy_key][key]["policy_id"] + subject_key = _cache.subject_assignments[policy_key][key]["subject_id"] + cat_key = _cache.subject_assignments[policy_key][key]["category_id"] + data_keys = _cache.subject_assignments[policy_key][key]["assignments"] + for data_key in data_keys: + _cache.delete_subject_assignment( + policy_id=policy_key, + perimeter_id=subject_key, + category_id=cat_key, + data_id=data_key + ) + _cache.delete_subject_assignment() + + +def clean_object_assignments(): + from moon_cache.cache import Cache + _cache = Cache.getInstance() + + if not _cache.object_assignments: + return + for policy_key in _cache.policies.keys(): + if policy_key not in _cache.object_assignments: + continue + for key in _cache.object_assignments[policy_key]: + # policy_key = _cache.object_assignments[policy_key][key]["policy_id"] + object_key = _cache.object_assignments[policy_key][key]["object_id"] + cat_key = _cache.object_assignments[policy_key][key]["category_id"] + data_keys = _cache.object_assignments[policy_key][key]["assignments"] + for data_key in data_keys: + _cache.delete_object_assignment( + policy_id=policy_key, + perimeter_id=object_key, + category_id=cat_key, + data_id=data_key + ) + _cache.delete_object_assignment() + + +def clean_action_assignments(): + from moon_cache.cache import Cache + _cache = Cache.getInstance() + + if not _cache.action_assignments: + return + for policy_key in _cache.policies.keys(): + if policy_key not in _cache.action_assignments: + continue + for key in _cache.action_assignments[policy_key]: + action_key = _cache.action_assignments[policy_key][key]["action_id"] + cat_key = _cache.action_assignments[policy_key][key]["category_id"] + data_keys = _cache.action_assignments[policy_key][key]["assignments"] + for data_key in data_keys: + _cache.delete_action_assignment( + policy_id=policy_key, + perimeter_id=action_key, + category_id=cat_key, + data_id=data_key + ) + _cache.delete_action_assignment() + + +def clean_rules(): + from moon_cache.cache import Cache + _cache = Cache.getInstance() + + rules = list(_cache.rules.keys()) + for key in rules: + _cache.delete_rule(key) + + +def clean_all(): + clean_rules() + + clean_subject_assignments() + clean_object_assignments() + clean_action_assignments() + + clean_subject_data() + clean_object_data() + clean_action_data() + + clean_actions() + clean_objects() + clean_subjects() + + clean_policies() + clean_models() + clean_meta_rule() + + clean_subject_categories() + clean_object_categories() + clean_action_categories() diff --git a/moon_utilities/tests/unit_python/helpers/import_export_db_helper.py b/moon_utilities/tests/unit_python/helpers/import_export_db_helper.py new file mode 100644 index 00000000..ecb9fa26 --- /dev/null +++ b/moon_utilities/tests/unit_python/helpers/import_export_db_helper.py @@ -0,0 +1,178 @@ +# Software Name: MOON + +# Version: 5.4 + +# SPDX-FileCopyrightText: Copyright (c) 2018-2020 Orange and its contributors +# SPDX-License-Identifier: Apache-2.0 + +# This software is distributed under the 'Apache License 2.0', +# the text of which is available at 'http://www.apache.org/licenses/LICENSE-2.0.txt' +# or see the "LICENSE" file for more details. + + +import logging + +logger = logging.getLogger("moon.manager.test.api." + __name__) + + +def clean_models(): + from moon_manager import db_driver as driver + keys = driver.ModelManager.get_models(moon_user_id="admin") + for key in keys: + driver.ModelManager.delete_model(moon_user_id="admin", model_id=key) + + +def clean_policies(): + from moon_manager import db_driver as driver + keys = driver.PolicyManager.get_policies(moon_user_id="admin") + for key in keys: + driver.PolicyManager.delete_policy(moon_user_id="admin", policy_id=key) + + +def clean_subjects(): + from moon_manager import db_driver as driver + for policy in driver.PolicyManager.get_policies(moon_user_id="admin"): + subjects = driver.PolicyManager.get_subjects(moon_user_id="admin", policy_id=policy) + for key in subjects: + driver.PolicyManager.delete_subject(moon_user_id="admin", policy_id=policy, + perimeter_id=key) + + +def clean_objects(): + from moon_manager import db_driver as driver + for policy in driver.PolicyManager.get_policies(moon_user_id="admin"): + objects = driver.PolicyManager.get_objects(moon_user_id="admin", policy_id=policy) + for key in objects: + driver.PolicyManager.delete_object(moon_user_id="admin", policy_id=policy, + perimeter_id=key) + + +def clean_actions(): + from moon_manager import db_driver as driver + for policy in driver.PolicyManager.get_policies(moon_user_id="admin"): + actions = driver.PolicyManager.get_actions(moon_user_id="admin", policy_id=policy) + for key in actions: + driver.PolicyManager.delete_action(moon_user_id="admin", policy_id=policy, + perimeter_id=key) + + +def clean_subject_categories(): + from moon_manager import db_driver as driver + categories = driver.ModelManager.get_subject_categories(moon_user_id="admin") + + for key in categories: + driver.ModelManager.delete_subject_category(moon_user_id="admin", category_id=key) + + +def clean_object_categories(): + from moon_manager import db_driver as driver + categories = driver.ModelManager.get_object_categories(moon_user_id="admin") + + for key in categories: + driver.ModelManager.delete_object_category(moon_user_id="admin", category_id=key) + + +def clean_action_categories(): + from moon_manager import db_driver as driver + categories = driver.ModelManager.get_action_categories(moon_user_id="admin") + + for key in categories: + driver.ModelManager.delete_action_category(moon_user_id="admin", category_id=key) + + +def clean_subject_data(): + from moon_manager import db_driver as driver + policies = driver.PolicyManager.get_policies(moon_user_id="admin") + categories = driver.ModelManager.get_subject_categories(moon_user_id="admin") + for policy in policies: + for category in categories: + data_object = driver.PolicyManager.get_subject_data( + moon_user_id="admin", policy_id=policy, category_id=category) + for data_item in data_object: + for data in data_item.get("data", {}): + driver.PolicyManager.delete_subject_data(moon_user_id="admin", policy_id=policy, + category_id=category, data_id=data) + + +def clean_object_data(): + from moon_manager import db_driver as driver + policies = driver.PolicyManager.get_policies(moon_user_id="admin") + categories = driver.ModelManager.get_object_categories(moon_user_id="admin") + for policy in policies: + for category in categories: + data_object = driver.PolicyManager.get_object_data( + moon_user_id="admin", policy_id=policy, category_id=category) + for data_item in data_object: + for data in data_item.get("data", {}): + driver.PolicyManager.delete_object_data(moon_user_id="admin", policy_id=policy, + category_id=category, data_id=data) + + +def clean_action_data(): + from moon_manager import db_driver as driver + policies = driver.PolicyManager.get_policies(moon_user_id="admin") + categories = driver.ModelManager.get_action_categories(moon_user_id="admin") + for policy in policies: + for category in categories: + data_object = driver.PolicyManager.get_action_data( + moon_user_id="admin", policy_id=policy, category_id=category) + for data_item in data_object: + for data in data_item.get("data", {}): + driver.PolicyManager.delete_action_data(moon_user_id="admin", policy_id=policy, + category_id=category, data_id=data) + + +def clean_meta_rule(): + from moon_manager import db_driver as driver + keys = driver.ModelManager.get_meta_rules(moon_user_id="admin") + + for key in keys: + driver.ModelManager.delete_meta_rule(moon_user_id="admin", meta_rule_id=key) + + +def clean_subject_assignments(): + pass + + +def clean_object_assignments(): + pass + + +def clean_action_assignments(): + pass + + +def clean_rules(): + from moon_manager import db_driver as driver + + policies = driver.PolicyManager.get_policies(moon_user_id="admin") + + for policy in policies: + rules = driver.PolicyManager.get_rules(moon_user_id="admin", policy_id=policy) + + for rule in rules: + driver.PolicyManager.delete_rule(moon_user_id="admin", policy_id=policy, rule_id=rule) + + +def clean_all(): + clean_rules() + + clean_subject_assignments() + clean_object_assignments() + clean_action_assignments() + + clean_subject_data() + clean_object_data() + clean_action_data() + + clean_actions() + clean_objects() + clean_subjects() + + clean_policies() + clean_models() + clean_meta_rule() + + clean_subject_categories() + clean_object_categories() + clean_action_categories() diff --git a/moon_utilities/tests/unit_python/helpers/slaves_helpers.py b/moon_utilities/tests/unit_python/helpers/slaves_helpers.py new file mode 100644 index 00000000..1b2764d2 --- /dev/null +++ b/moon_utilities/tests/unit_python/helpers/slaves_helpers.py @@ -0,0 +1,32 @@ +# Software Name: MOON + +# Version: 5.4 + +# SPDX-FileCopyrightText: Copyright (c) 2018-2020 Orange and its contributors +# SPDX-License-Identifier: Apache-2.0 + +# This software is distributed under the 'Apache License 2.0', +# the text of which is available at 'http://www.apache.org/licenses/LICENSE-2.0.txt' +# or see the "LICENSE" file for more details. + + + +SLAVES = { + "slaves": { + "d464cc58a0cd46dea3191ba70f4e7df8": { + "name": "slave_test", + "address": "", + "description": "...", + "extra": { + "description": "...", + "starttime": 1543851265.76279, + "port": 10000, + "server_ip": "127.0.0.1", + "status": "down", + "api_key": "e58a882a6b658a22660f00a0c273e7f6b4c4eb5abe54eccba2cae307905d67e3746537" + "bd790c41887e11840c2d186b6d6eeec0e426bcfa7a872cc3417a35124a" + } + } + } +} + |