Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use cache_delete method in tempest

Best Python code snippet using tempest_python

test_cache.py

Source:test_cache.py

1#2# Copyright 2020 Red Hat, Inc.3#4# This program is free software: you can redistribute it and/or modify5# it under the terms of the GNU Affero General Public License as6# published by the Free Software Foundation, either version 3 of the7# License, or (at your option) any later version.8#9# This program is distributed in the hope that it will be useful,10# but WITHOUT ANY WARRANTY; without even the implied warranty of11# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the12# GNU Affero General Public License for more details.13#14# You should have received a copy of the GNU Affero General Public License15# along with this program.  If not, see <https://www.gnu.org/licenses/>.16#17"""Test the caching system."""18import pickle19from unittest import skipIf20from unittest.mock import call, MagicMock, patch21from rbac.settings import ACCESS_CACHE_ENABLED22from django.conf import settings23from django.db import connection24from django.test import TestCase25from management.cache import TenantCache26from management.models import Access, Group, Permission, Policy, Principal, ResourceDefinition, Role27from api.models import Tenant28@skipIf(not ACCESS_CACHE_ENABLED, "Caching is disabled.")29class AccessCacheTest(TestCase):30    @classmethod31    def setUpClass(self):32        """Set up the tenant."""33        super().setUpClass()34        self.tenant = Tenant.objects.create(tenant_name="acct12345")35        self.tenant.ready = True36        self.tenant.save()37    def setUp(self):38        """Set up AccessCache tests."""39        super().setUp()40        self.principal_a = Principal.objects.create(username="principal_a", tenant=self.tenant)41        self.principal_b = Principal.objects.create(username="principal_b", tenant=self.tenant)42        self.group_a = Group.objects.create(name="group_a", platform_default=True, tenant=self.tenant)43        self.group_b = Group.objects.create(name="group_b", tenant=self.tenant)44        self.policy_a = Policy.objects.create(name="policy_a", tenant=self.tenant)45        self.policy_b = Policy.objects.create(name="policy_b", tenant=self.tenant)46        self.role_a = Role.objects.create(name="role_a", tenant=self.tenant)47        self.role_b = Role.objects.create(name="role_b", tenant=self.tenant)48    @classmethod49    def tearDownClass(self):50        self.tenant.delete()51        super().tearDownClass()52    @patch("management.group.model.AccessCache.delete_policy")53    def test_group_cache_add_remove_signals(self, cache):54        """Test signals attached to Groups"""55        cache.reset_mock()56        # If a Principal is added to a group57        self.group_a.principals.add(self.principal_a)58        cache.assert_called_once()59        cache.assert_called_once_with(self.principal_a.uuid)60        cache.reset_mock()61        # If a Group is added to a Principal62        self.principal_b.group.add(self.group_a)63        cache.asset_called_once()64        cache.asset_called_once_with(self.principal_b.uuid)65        cache.reset_mock()66        # If a Principal is removed from a group67        self.group_a.principals.remove(self.principal_a)68        cache.assert_called_once()69        cache.assert_called_once_with(self.principal_a.uuid)70        cache.reset_mock()71        # If a Group is removed from a Principal72        self.principal_b.group.remove(self.group_a)73        cache.asset_called_once()74        cache.asset_called_once_with(self.principal_b.uuid)75    @patch("management.group.model.AccessCache.delete_policy")76    def test_group_cache_clear_signals(self, cache):77        # If all groups are removed from a Principal78        self.group_a.principals.add(self.principal_a, self.principal_b)79        cache.reset_mock()80        self.principal_a.group.clear()81        cache.assert_called_once()82        cache.assert_called_once_with(self.principal_a.uuid)83        cache.reset_mock()84        # If all Principals are removed from a Group85        self.group_a.principals.clear()86        cache.asset_called_once()87        cache.asset_called_once_with(self.principal_b.uuid)88    @patch("management.group.model.AccessCache.delete_policy")89    def test_group_cache_delete_group_signal(self, cache):90        self.group_a.principals.add(self.principal_a)91        cache.reset_mock()92        self.group_a.delete()93        cache.assert_called_once()94        cache.assert_called_once_with(self.principal_a.uuid)95    @patch("management.policy.model.AccessCache.delete_all_policies_for_tenant")96    @patch("management.policy.model.AccessCache.delete_policy")97    def test_policy_cache_group_signals(self, cache_delete, cache_delete_all):98        """Test signals attached to Groups"""99        self.group_a.principals.add(self.principal_a)100        self.group_b.principals.add(self.principal_b)101        cache_delete.reset_mock()102        # If a policy has its group set103        self.policy_a.group = self.group_a104        self.policy_a.save()105        cache_delete_all.asset_called_once()106        cache_delete.reset_mock()107        # If a policy has its group changed108        self.policy_a.group = self.group_b109        self.policy_a.save()110        cache_delete.asset_called_once()111        cache_delete.asset_called_once_with(self.principal_b.uuid)112        cache_delete.reset_mock()113        # If a policy is deleted114        self.policy_a.delete()115        cache_delete.assert_called_once()116        cache_delete.assert_called_once_with(self.principal_b.uuid)117    @patch("management.policy.model.AccessCache.delete_all_policies_for_tenant")118    @patch("management.policy.model.AccessCache.delete_policy")119    def test_policy_cache_add_remove_roles_signals(self, cache_delete, cache_delete_all):120        """Test signals attached to Policy/Roles"""121        self.group_b.principals.add(self.principal_b)122        self.policy_a.group = self.group_a123        self.policy_a.save()124        self.policy_b.group = self.group_b125        self.policy_b.save()126        cache_delete.reset_mock()127        # If a Role is added to a platform default group's Policy128        self.policy_a.roles.add(self.role_a)129        self.policy_a.save()130        cache_delete_all.asset_called_once()131        cache_delete.reset_mock()132        # If a Policy is added to a Role133        self.role_b.policies.add(self.policy_a)134        cache_delete.asset_called_once()135        cache_delete.asset_called_once_with(self.principal_b.uuid)136        cache_delete.reset_mock()137        # If a Role is removed from a platform default group's Policy138        self.policy_a.roles.remove(self.role_a)139        self.policy_a.save()140        cache_delete_all.asset_called_once()141        cache_delete.reset_mock()142        # If a Role is removed from a Policy143        self.policy_b.roles.remove(self.role_b)144        cache_delete.assert_called_once()145        cache_delete.assert_called_once_with(self.principal_b.uuid)146        cache_delete.reset_mock()147        # If a Policy is removed from a Role148        self.role_b.policies.remove(self.policy_b)149        cache_delete.asset_called_once()150        cache_delete.asset_called_once_with(self.principal_b.uuid)151    @patch("management.policy.model.AccessCache.delete_policy")152    def test_policy_cache_clear_signals(self, cache):153        self.group_a.principals.add(self.principal_a)154        self.group_b.principals.add(self.principal_b)155        self.policy_a.group = self.group_a156        self.policy_a.save()157        self.policy_b.group = self.group_b158        self.policy_b.save()159        self.policy_a.roles.add(self.role_a)160        self.policy_b.roles.add(self.role_b)161        cache.reset_mock()162        # If all policies are removed from a role163        self.role_a.policies.clear()164        cache.assert_called_once()165        cache.assert_called_once_with(self.principal_a.uuid)166        cache.reset_mock()167        # If all Roles are removed from a Policy168        self.policy_b.roles.clear()169        cache.asset_called_once()170        cache.asset_called_once_with(self.principal_b.uuid)171    @patch("management.role.model.AccessCache.delete_policy")172    def test_policy_cache_change_delete_roles_signals(self, cache):173        self.group_a.principals.add(self.principal_a)174        self.group_b.principals.add(self.principal_b)175        self.policy_a.group = self.group_a176        self.policy_a.save()177        self.policy_b.group = self.group_b178        self.policy_b.save()179        self.policy_a.roles.add(self.role_a)180        self.policy_b.roles.add(self.role_b)181        cache.reset_mock()182        # If a role is changed183        self.role_a.version += 1184        self.role_a.save()185        cache.assert_called_once()186        cache.assert_called_once_with(self.principal_a.uuid)187        cache.reset_mock()188        # If Access is added189        self.permission = Permission.objects.create(permission="foo:*:*", tenant=self.tenant)190        self.access_a = Access.objects.create(permission=self.permission, role=self.role_a, tenant=self.tenant)191        cache.assert_called_once()192        cache.assert_called_once_with(self.principal_a.uuid)193        cache.reset_mock()194        # If ResourceDefinition is added195        self.rd_a = ResourceDefinition.objects.create(access=self.access_a, tenant=self.tenant)196        cache.assert_called_once()197        cache.assert_called_once_with(self.principal_a.uuid)198        cache.reset_mock()199        # If ResourceDefinition is destroyed200        self.rd_a.delete()201        cache.assert_called_once()202        cache.assert_called_once_with(self.principal_a.uuid)203        cache.reset_mock()204        # If Access is destroyed205        self.access_a.delete()206        cache.assert_called_once()207        cache.assert_called_once_with(self.principal_a.uuid)208        cache.reset_mock()209        # If Role is destroyed210        self.role_a.delete()211        cache.assert_called_once()212        cache.assert_called_once_with(self.principal_a.uuid)213class TenantCacheTest(TestCase):214    @classmethod215    def setUpClass(self):216        """Set up the tenant."""217        super().setUpClass()218        self.tenant = Tenant.objects.create(tenant_name="acct67890")219    @classmethod220    def tearDownClass(self):221        self.tenant.delete()222        super().tearDownClass()223    @patch("management.cache.TenantCache.connection")224    def test_tenant_cache_functions_success(self, redis_connection):225        tenant_name = self.tenant.tenant_name226        tenant_org_id = self.tenant.org_id227        if settings.AUTHENTICATE_WITH_ORG_ID:228            key = f"rbac::tenant::tenant={tenant_org_id}"229        else:230            key = f"rbac::tenant::tenant={tenant_name}"231        dump_content = pickle.dumps(self.tenant)232        # Save tenant to cache233        tenant_cache = TenantCache()234        tenant_cache.save_tenant(self.tenant)235        self.assertTrue(call().__enter__().set(key, dump_content) in redis_connection.pipeline.mock_calls)236        redis_connection.get.return_value = dump_content237        # Get tenant from cache238        if settings.AUTHENTICATE_WITH_ORG_ID:239            tenant = tenant_cache.get_tenant(tenant_org_id)240        else:241            tenant = tenant_cache.get_tenant(tenant_name)242        redis_connection.get.assert_called_once_with(key)243        self.assertEqual(tenant, self.tenant)244        # Delete tenant from cache245        if settings.AUTHENTICATE_WITH_ORG_ID:246            tenant_cache.delete_tenant(tenant_org_id)247        else:248            tenant_cache.delete_tenant(tenant_name)...

test_invalidating_caches.py

Source:test_invalidating_caches.py

1import pytest2from ..models import CacheVersion3from ..versions import CACHE_NAME, invalidate_all_caches, invalidate_cache4@pytest.fixture5def cache_delete(mocker):6    return mocker.patch("django.core.cache.cache.delete")7def test_invalidating_cache_updates_cache_version_in_database(8    cache_delete, cache_version9):10    invalidate_cache(cache_version.cache)11    updated_cache_version = CacheVersion.objects.get(cache=cache_version.cache)12    assert cache_version.version != updated_cache_version.version13def test_invalidating_cache_deletes_versions_cache(cache_delete, cache_version):14    invalidate_cache(cache_version.cache)15    cache_delete.assert_called_once_with(CACHE_NAME)16def test_invalidating_all_caches_updates_cache_version_in_database(17    cache_delete, cache_version18):19    invalidate_all_caches()...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.