diff --git a/awx/main/access.py b/awx/main/access.py index 7e98d4afcc..59995f3b53 100644 --- a/awx/main/access.py +++ b/awx/main/access.py @@ -95,7 +95,7 @@ def check_user_access(user, model_class, action, *args, **kwargs): continue result = access_method(*args, **kwargs) logger.debug('%s.%s %r returned %r', access_instance.__class__.__name__, - access_method.__name__, args, result) + getattr(access_method, '__name__', 'unknown'), args, result) if result: return result return False diff --git a/awx/main/tests/conftest.py b/awx/main/tests/conftest.py index de1f2bdd97..35f77c1f1d 100644 --- a/awx/main/tests/conftest.py +++ b/awx/main/tests/conftest.py @@ -2,6 +2,8 @@ # Python import time import pytest +import mock +from contextlib import contextmanager from awx.main.tests.factories import ( create_organization, @@ -14,6 +16,22 @@ from awx.main.tests.factories import ( ) +@pytest.fixture +def mock_access(): + @contextmanager + def access_given_class(TowerClass): + try: + mock_instance = mock.MagicMock(__name__='foobar') + MockAccess = mock.MagicMock(return_value=mock_instance) + the_patch = mock.patch.dict('awx.main.access.access_registry', + {TowerClass: [MockAccess]}, clear=False) + the_patch.__enter__() + yield mock_instance + finally: + the_patch.__exit__() + return access_given_class + + @pytest.fixture def job_template_factory(): return create_job_template diff --git a/awx/main/tests/functional/api/test_inventory.py b/awx/main/tests/functional/api/test_inventory.py index 3266c3f27c..66b38910b5 100644 --- a/awx/main/tests/functional/api/test_inventory.py +++ b/awx/main/tests/functional/api/test_inventory.py @@ -210,18 +210,18 @@ def test_delete_inventory_host(delete, host, alice, role_field, expected_status_ delete(reverse('api:host_detail', kwargs={'pk': host.id}), alice, expect=expected_status_code) -@pytest.mark.parametrize("role_field,expected_status_code", [ - (None, 403), - ('admin_role', 202), - ('update_role', 202), - ('adhoc_role', 403), - ('use_role', 403) +# See companion test in tests/functional/test_rbac_inventory.py::test_inventory_source_update +@pytest.mark.parametrize("start_access,expected_status_code", [ + (True, 202), + (False, 403) ]) @pytest.mark.django_db -def test_inventory_source_update(post, inventory_source, alice, role_field, expected_status_code): - if role_field: - getattr(inventory_source.inventory, role_field).members.add(alice) - post(reverse('api:inventory_source_update_view', kwargs={'pk': inventory_source.id}), {}, alice, expect=expected_status_code) +def test_inventory_update_access_called(post, inventory_source, alice, mock_access, start_access, expected_status_code): + with mock_access(InventorySource) as mock_instance: + mock_instance.can_start = mock.MagicMock(return_value=start_access) + post(reverse('api:inventory_source_update_view', kwargs={'pk': inventory_source.id}), + {}, alice, expect=expected_status_code) + mock_instance.can_start.assert_called_once_with(inventory_source) @pytest.mark.django_db diff --git a/awx/main/tests/functional/test_rbac_inventory.py b/awx/main/tests/functional/test_rbac_inventory.py index 7b85e1d44a..99937899a8 100644 --- a/awx/main/tests/functional/test_rbac_inventory.py +++ b/awx/main/tests/functional/test_rbac_inventory.py @@ -93,6 +93,21 @@ def test_inventory_update_org_admin(inventory_update, org_admin): assert access.can_delete(inventory_update) +# See companion test in tests/functional/api/test_inventory.py::test_inventory_update_access_called +@pytest.mark.parametrize("role_field,allowed", [ + (None, False), + ('admin_role', True), + ('update_role', True), + ('adhoc_role', False), + ('use_role', False) +]) +@pytest.mark.django_db +def test_inventory_source_update(inventory_source, alice, role_field, allowed): + if role_field: + getattr(inventory_source.inventory, role_field).members.add(alice) + assert allowed == InventorySourceAccess(alice).can_start(inventory_source), '{} test failed'.format(role_field) + + @pytest.mark.django_db def test_host_access(organization, inventory, group, user, group_factory): other_inventory = organization.inventories.create(name='other-inventory')