# Copyright (C) 2022 CVAT.ai Corporation # # SPDX-License-Identifier: MIT from copy import deepcopy from http import HTTPStatus from itertools import product import pytest from deepdiff import DeepDiff from shared.utils.config import delete_method, get_method, patch_method, post_method @pytest.mark.usefixtures('changedb') class TestPostWebhooks: proj_webhook = { 'description': 'webhook description', 'content_type': 'application/json', 'enable_ssl': False, 'events': [ 'create:task', 'delete:task' ], 'is_active': True, 'project_id': 1, 'secret': 'secret', 'target_url': 'http://example.com', 'type': 'project', } org_webhook = { 'description': 'webhook description', 'content_type': 'application/json', 'enable_ssl': False, 'events': [ 'create:task', 'delete:task' ], 'is_active': True, 'secret': 'secret', 'target_url': 'http://example.com', 'type': 'organization', } def test_sandbox_admin_can_create_webhook_for_project(self, projects, users): admin = next((u for u in users if 'admin' in u['groups'])) project = [p for p in projects if p['owner']['id'] != admin['id'] and p['organization'] is None][0] webhook = deepcopy(self.proj_webhook) webhook['project_id'] = project['id'] response = post_method(admin['username'], 'webhooks', webhook) assert response.status_code == HTTPStatus.CREATED assert 'secret' not in response.json() def test_admin_can_create_webhook_for_org(self, users, organizations, is_org_member): admins = [u for u in users if 'admin' in u['groups']] username, org_id = next(( (user['username'], org['id']) for user in admins for org in organizations if not is_org_member(user['id'], org['id']) )) webhook = deepcopy(self.org_webhook) response = post_method(username, 'webhooks', webhook, org_id=org_id) assert response.status_code == HTTPStatus.CREATED assert 'secret' not in response.json() def test_admin_can_create_webhook_for_project_in_org(self, users, projects_by_org, organizations, is_org_member): admins = [u for u in users if 'admin' in u['groups']] not_org_members = [(u, o) for u, o in product(admins, organizations) if not is_org_member(u['id'], o['id'])] username, org_id = next(( (u['username'], o['id']) for u, o in not_org_members for p in projects_by_org.get(o['id'], []) if p['owner']['id'] != u['id'] )) webhook = deepcopy(self.org_webhook) response = post_method(username, 'webhooks', webhook, org_id=org_id) assert response.status_code == HTTPStatus.CREATED assert 'secret' not in response.json() @pytest.mark.parametrize('privilege', ['user', 'business']) def test_sandbox_project_owner_can_create_webhook_for_project(self, privilege, projects, users): users = [user for user in users if privilege in user['groups']] username, project_id = next(( (user['username'], project['id']) for user in users for project in projects if project['owner']['id'] == user['id'] and project['organization'] is None )) webhook = deepcopy(self.proj_webhook) webhook['project_id'] = project_id response = post_method(username, 'webhooks', webhook) assert response.status_code == HTTPStatus.CREATED assert 'secret' not in response.json() @pytest.mark.parametrize('privilege', ['worker', 'user', 'business']) def test_sandbox_project_assignee_cannot_create_webhook_for_project(self, privilege, projects, users): users = [u for u in users if privilege in u['groups']] projects = [p for p in projects if p['assignee'] is not None] username, project_id = next(( (user['username'], project['id']) for user in users for project in projects if project['assignee']['id'] == user['id'] and project['organization'] is None )) webhook = deepcopy(self.proj_webhook) webhook['project_id'] = project_id response = post_method(username, 'webhooks', webhook) assert response.status_code == HTTPStatus.FORBIDDEN @pytest.mark.parametrize('role', ['maintainer', 'owner']) def test_member_can_create_webhook_for_org(self, role, find_users, organizations): username, org_id = next(( (u['username'], o['id']) for o in organizations for u in find_users(org=o['id'], role=role, exclude_privilege='admin') )) webhook = deepcopy(self.org_webhook) response = post_method(username, 'webhooks', webhook, org_id=org_id) assert response.status_code == HTTPStatus.CREATED assert 'secret' not in response.json() @pytest.mark.parametrize('role', ['maintainer', 'owner']) def test_member_can_create_webhook_for_project(self, role, find_users, organizations, projects_by_org, is_project_staff): username, oid, pid = next(( (u['username'], o['id'], p['id']) for o in organizations for u in find_users(org=o['id'], role=role, exclude_privilege='admin') for p in projects_by_org.get(o['id'], []) if not is_project_staff(u['id'], p['id']) )) webhook = deepcopy(self.proj_webhook) webhook['project_id'] = pid response = post_method(username, 'webhooks', webhook, org_id=oid) assert response.status_code == HTTPStatus.CREATED assert 'secret' not in response.json() @pytest.mark.parametrize('role', ['supervisor', 'worker']) def test_member_cannot_create_webhook_for_org(self, role, find_users, organizations): username, org_id = next(( (u['username'], o['id']) for o in organizations for u in find_users(org=o['id'], role=role, exclude_privilege='admin') )) webhook = deepcopy(self.org_webhook) response = post_method(username, 'webhooks', webhook, org_id=org_id) assert response.status_code == HTTPStatus.FORBIDDEN @pytest.mark.parametrize('role', ['supervisor', 'worker']) def test_member_cannot_create_webhook_for_project(self, role, find_users, organizations, projects_by_org, is_project_staff): username, oid, pid = next(( (u['username'], o['id'], p['id']) for o in organizations for u in find_users(org=o['id'], role=role, exclude_privilege='admin') for p in projects_by_org.get(o['id'], []) if not is_project_staff(u['id'], p['id']) )) webhook = deepcopy(self.proj_webhook) webhook['project_id'] = pid response = post_method(username, 'webhooks', webhook, org_id=oid) assert response.status_code == HTTPStatus.FORBIDDEN @pytest.mark.parametrize('role', ['supervisor']) def test_member_project_owner_can_create_webhook_for_project(self, role, find_users, organizations, projects_by_org, is_project_staff): username, oid, pid = next(( (u['username'], o['id'], p['id']) for o in organizations for u in find_users(org=o['id'], role=role, exclude_privilege='admin') for p in projects_by_org.get(o['id'], []) if p['owner']['id'] == u['id'] )) webhook = deepcopy(self.proj_webhook) webhook['project_id'] = pid response = post_method(username, 'webhooks', webhook, org_id=oid) assert response.status_code == HTTPStatus.CREATED assert 'secret' not in response.json() def test_non_member_cannot_create_webhook_for_org(self, find_users, organizations, is_org_member): username, org_id = next(( (u['username'], o['id']) for o in organizations for u in find_users(exclude_privilege='admin') if not is_org_member(u['id'], o['id']) )) webhook = deepcopy(self.org_webhook) response = post_method(username, 'webhooks', webhook, org_id=org_id) assert response.status_code == HTTPStatus.FORBIDDEN def test_can_create_without_unnecessary_fields(self): post_data = deepcopy(self.proj_webhook) post_data.pop('enable_ssl') post_data.pop('content_type') post_data.pop('description') post_data.pop('is_active') post_data.pop('secret') response = post_method('admin2', 'webhooks', post_data) assert response.status_code == HTTPStatus.CREATED def test_cannot_create_without_target_url(self): post_data = deepcopy(self.proj_webhook) post_data.pop('target_url') response = post_method('admin2', 'webhooks', post_data) assert response.status_code == HTTPStatus.BAD_REQUEST def test_cannot_create_without_events_list(self): post_data = deepcopy(self.proj_webhook) post_data.pop('events') response = post_method('admin2', 'webhooks', post_data) assert response.status_code == HTTPStatus.BAD_REQUEST def test_cannot_create_without_type(self): post_data = deepcopy(self.proj_webhook) post_data.pop('type') response = post_method('admin2', 'webhooks', post_data) assert response.status_code == HTTPStatus.BAD_REQUEST def test_cannot_create_without_project_id(self): post_data = deepcopy(self.proj_webhook) post_data.pop('project_id') response = post_method('admin2', 'webhooks', post_data) assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR def test_cannot_create_organization_webhook_when_project_id_is_not_null(self, organizations): post_data = deepcopy(self.proj_webhook) post_data['type'] = 'organization' org_id = organizations.raw[0]['id'] response = post_method('admin2', 'webhooks', post_data, org_id=org_id) assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR def test_cannot_create_non_unique_webhook(self): pytest.skip('Not implemeted yet') response = post_method('admin2', 'webhooks', self.proj_webhook) response = post_method('admin2', 'webhooks', self.proj_webhook) assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR def test_cannot_create_for_non_existent_organization(self, organizations): post_data = deepcopy(self.proj_webhook) post_data['type'] = 'organization' org_id = max(a['id'] for a in organizations.raw) + 1 response = post_method('admin2', 'webhooks', post_data, org_id=org_id) assert response.status_code == HTTPStatus.BAD_REQUEST def test_cannot_create_for_non_existent_project(self, projects): post_data = deepcopy(self.proj_webhook) post_data['project_id'] = max(a['id'] for a in projects.raw) + 1 response = post_method('admin2', 'webhooks', post_data) assert response.status_code == HTTPStatus.BAD_REQUEST def test_cannot_create_with_non_supported_type(self): post_data = deepcopy(self.proj_webhook) post_data['type'] = 'some_type' response = post_method('admin2', 'webhooks', post_data) assert response.status_code == HTTPStatus.BAD_REQUEST def test_cannot_create_with_non_supported_content_type(self): post_data = deepcopy(self.proj_webhook) post_data['content_type'] = ['application/x-www-form-urlencoded'] response = post_method('admin2', 'webhooks', post_data) assert response.status_code == HTTPStatus.BAD_REQUEST @pytest.mark.parametrize('event', ['some:event', 'create:project', 'update:organization', 'create:invitation']) def test_cannot_create_project_webhook_with_non_supported_event_type(self, event): post_data = deepcopy(self.proj_webhook) post_data['events'] = [event] response = post_method('admin2', 'webhooks', post_data) assert response.status_code == HTTPStatus.BAD_REQUEST @pytest.mark.parametrize('event', ['some:event', 'create:organization']) def test_cannot_create_organization_webhook_with_non_supported_event_type(self, event, organizations): post_data = deepcopy(self.proj_webhook) post_data['type'] = 'organization' post_data['events'] = [event] org_id = next(iter(organizations))['id'] response = post_method('admin2', 'webhooks', post_data, org_id=org_id) assert response.status_code == HTTPStatus.BAD_REQUEST @pytest.mark.usefixtures('dontchangedb') class TestGetWebhooks: def test_admin_can_get_webhook(self, webhooks, users, projects): proj_webhooks = [w for w in webhooks if w['type'] == 'project'] username, wid = next(( (user['username'], webhook['id']) for user in users for webhook in proj_webhooks if 'admin' in user['groups'] and webhook['owner']['id'] != user['id'] and projects[webhook['project']]['owner']['id'] != user['id'] )) response = get_method(username, f"webhooks/{wid}") assert response.status_code == HTTPStatus.OK assert 'secret' not in response.json() assert DeepDiff(webhooks[wid], response.json(), ignore_order=True) == {} @pytest.mark.parametrize('privilege', ['user', 'business']) def test_project_owner_can_get_webhook(self, privilege, webhooks, projects, users): proj_webhooks = [w for w in webhooks if w['type'] == 'project'] username, wid = next(( (user['username'], webhook['id']) for user in users for webhook in proj_webhooks if privilege not in user['groups'] and projects[webhook['project']]['owner']['id'] == user['id'] )) response = get_method(username, f"webhooks/{wid}") assert response.status_code == HTTPStatus.OK assert 'secret' not in response.json() assert DeepDiff(webhooks[wid], response.json(), ignore_order=True) == {} @pytest.mark.parametrize('privilege', ['user', 'business']) def test_webhook_owner_can_get_webhook(self, privilege, webhooks, projects, users): proj_webhooks = [w for w in webhooks if w['type'] == 'project'] username, wid = next(( (user['username'], webhook['id']) for user in users for webhook in proj_webhooks if privilege in user['groups'] and webhook['owner']['id'] == user['id'] )) response = get_method(username, f"webhooks/{wid}") assert response.status_code == HTTPStatus.OK assert 'secret' not in response.json() assert DeepDiff(webhooks[wid], response.json(), ignore_order=True) == {} @pytest.mark.parametrize('privilege', ['user', 'business']) def test_not_project_staff_cannot_get_webhook(self, privilege, webhooks, projects, users): proj_webhooks = [w for w in webhooks if w['type'] == 'project'] username, wid = next(( (user['username'], webhook['id']) for user in users for webhook in proj_webhooks if privilege in user['groups'] and projects[webhook['project']]['owner']['id'] != user['id'] and webhook['owner']['id'] != user['id'] )) response = get_method(username, f"webhooks/{wid}") assert response.status_code == HTTPStatus.FORBIDDEN @pytest.mark.parametrize('role', ['owner', 'maintainer']) def test_org_staff_can_see_org_webhook(self, role, webhooks, find_users): webhook = next((w for w in webhooks if w['type'] == 'organization')) username = next(( u['username'] for u in find_users(role=role, org=webhook['organization']) )) response = get_method(username, f"webhooks/{webhook['id']}", org_id=webhook['organization']) assert response.status_code == HTTPStatus.OK assert 'secret' not in response.json() assert DeepDiff(webhook, response.json(), ignore_order=True) == {} @pytest.mark.parametrize('role', ['owner', 'maintainer']) def test_org_staff_can_see_project_webhook_in_org(self, role, webhooks, find_users, projects): proj_webhooks = [w for w in webhooks if w['organization'] is not None and w['type'] == 'project'] username, webhook = next(( (user['username'], webhook) for webhook in proj_webhooks for user in find_users(role=role, org=webhook['organization']) if projects[webhook['project']]['owner']['id'] != user['id'] and webhook['owner']['id'] != user['id'] )) response = get_method(username, f"webhooks/{webhook['id']}", org_id=webhook['organization']) assert response.status_code == HTTPStatus.OK assert 'secret' not in response.json() assert DeepDiff(webhook, response.json(), ignore_order=True) == {} @pytest.mark.parametrize('role', ['worker', 'supervisor']) def test_member_cannot_get_org_webhook(self, role, webhooks, find_users): webhook = next((w for w in webhooks if w['type'] == 'organization')) username = next(( u['username'] for u in find_users(role=role, org=webhook['organization']) )) response = get_method(username, f"webhooks/{webhook['id']}", org_id=webhook['organization']) assert response.status_code == HTTPStatus.FORBIDDEN @pytest.mark.parametrize('role', ['worker', 'supervisor']) def test_member_cannot_get_project_webhook_in_org(self, role, webhooks, find_users, projects): proj_webhooks = [w for w in webhooks if w['organization'] is not None and w['type'] == 'project'] username, webhook = next(( (user['username'], webhook) for webhook in proj_webhooks for user in find_users(role=role, org=webhook['organization']) if projects[webhook['project']]['owner']['id'] != user['id'] and webhook['owner']['id'] != user['id'] )) response = get_method(username, f"webhooks/{webhook['id']}", org_id=webhook['organization']) assert response.status_code == HTTPStatus.FORBIDDEN @pytest.mark.parametrize('role', ['supervisor']) def test_member_can_get_project_webhook_in_org(self, role, webhooks, find_users, projects): proj_webhooks = [w for w in webhooks if w['organization'] is not None and w['type'] == 'project'] username, webhook = next(( (user['username'], webhook) for webhook in proj_webhooks for user in find_users(role=role, org=webhook['organization']) if projects[webhook['project']]['owner']['id'] == user['id'] or webhook['owner']['id'] == user['id'] )) response = get_method(username, f"webhooks/{webhook['id']}", org_id=webhook['organization']) assert response.status_code == HTTPStatus.OK assert 'secret' not in response.json() assert DeepDiff(webhook, response.json(), ignore_order=True) == {} @pytest.mark.usefixtures('dontchangedb') class TestGetListWebhooks: def test_can_get_webhooks_list(self, webhooks): response = get_method('admin2', 'webhooks') assert response.status_code == HTTPStatus.OK assert all(['secret' not in webhook for webhook in response.json()['results']]) assert DeepDiff(webhooks.raw, response.json()['results'], ignore_order=True) == {} def test_admin_can_get_webhooks_for_project(self, webhooks): pid = next((webhook['project'] for webhook in webhooks if webhook['type'] == 'project' and webhook['organization'] is None)) expected_response = [webhook for webhook in webhooks if webhook['type'] == 'project' and webhook['project'] == pid] filter_val = '{"and":[{"==":[{"var":"project_id"},%s]}]}' % pid response = get_method('admin2', "webhooks", filter=filter_val) assert response.status_code == HTTPStatus.OK assert DeepDiff(expected_response, response.json()['results'], ignore_order=True) == {} def test_admin_can_get_webhooks_for_organization(self, webhooks): org_id = next((webhook['organization'] for webhook in webhooks if webhook['organization'] is not None)) expected_response = [webhook for webhook in webhooks if webhook['organization'] == org_id] response = get_method('admin2', "webhooks", org_id=org_id) assert response.status_code == HTTPStatus.OK assert DeepDiff(expected_response, response.json()['results'], ignore_order=True) == {} def test_admin_can_get_webhooks_for_project_in_org(self, webhooks): pid, oid = next(( (webhook['project'], webhook['organization']) for webhook in webhooks if webhook['type'] == 'project' and webhook['organization'] is not None )) expected_response = [webhook for webhook in webhooks if webhook['project'] == pid and webhook['organization'] == oid] filter_val = '{"and":[{"==":[{"var":"project_id"},%s]}]}' % pid response = get_method('admin2', "webhooks", org_id=oid, filter=filter_val) assert response.status_code == HTTPStatus.OK assert DeepDiff(expected_response, response.json()['results'], ignore_order=True) == {} @pytest.mark.parametrize('privilege', ['user', 'business']) def test_user_cannot_get_webhook_list_for_project(self, privilege, find_users, webhooks, projects): username, pid = next(( (user['username'], webhook['project']) for user in find_users(privilege=privilege) for webhook in webhooks if webhook['type'] == 'project' and webhook['organization'] is None and webhook['owner']['id'] != user['id'] and projects[webhook['project']]['owner']['id'] != user['id'] )) filter_val = '{"and":[{"==":[{"var":"project_id"},%s]}]}' % pid response = get_method(username, "webhooks", filter=filter_val) assert response.status_code == HTTPStatus.OK assert DeepDiff([], response.json()['results'], ignore_order=True) == {} @pytest.mark.parametrize('privilege', ['user', 'business']) def test_user_can_get_webhook_list_for_project(self, privilege, find_users, webhooks, projects): username, pid = next(( (user['username'], webhook['project']) for user in find_users(privilege=privilege) for webhook in webhooks if webhook['type'] == 'project' and webhook['organization'] is None and projects[webhook['project']]['owner']['id'] == user['id'] )) expected_response = [w for w in webhooks if w['type'] == 'project' and w['project'] == pid] filter_val = '{"and":[{"==":[{"var":"project_id"},%s]}]}' % pid response = get_method(username, "webhooks", filter=filter_val) assert response.status_code == HTTPStatus.OK assert DeepDiff(expected_response, response.json()['results'], ignore_order=True) == {} def test_non_member_cannot_see_webhook_list_for_org(self, webhooks, users, is_org_member): username, org_id = next(( (user['username'], webhook['organization']) for webhook in webhooks for user in users if webhook['organization'] is not None and not is_org_member(user['id'], webhook['organization']) and 'admin' not in user['groups'] )) response = get_method(username, 'webhooks', org_id=org_id) assert response.status_code == HTTPStatus.FORBIDDEN @pytest.mark.parametrize('role', ['maintainer', 'owner']) def test_org_staff_can_see_all_org_webhooks(self, role, webhooks, organizations, find_users): username, org_id = next(( (user['username'], org['id']) for webhook in webhooks for org in organizations for user in find_users(role=role, org=org['id']) if webhook['organization'] == org['id'] )) expected_response = [webhook for webhook in webhooks if webhook['organization'] == org_id] response = get_method(username, 'webhooks', org_id=org_id) assert response.status_code == HTTPStatus.OK assert DeepDiff(expected_response, response.json()['results'], ignore_order=True) == {} @pytest.mark.parametrize('role', ['worker', 'supervisor']) def test_member_cannot_see_all_org_webhook(self, role, webhooks, organizations, find_users, projects): username, org_id = next(( (user['username'], org['id']) for webhook in webhooks for org in organizations for user in find_users(role=role, org=org['id']) if webhook['organization'] == org['id'] )) expected_response = [ webhook for webhook in webhooks if webhook['organization'] == org_id and (webhook['owner']['username'] == username or (webhook['project'] and projects[webhook['project']]['owner']['username'] == username)) ] response = get_method(username, 'webhooks', org_id=org_id) assert response.status_code == HTTPStatus.OK assert DeepDiff(expected_response, response.json()['results'], ignore_order=True) == {} @pytest.mark.parametrize('role', ['supervisor']) def test_member_can_see_list_of_project_webhooks_in_org(self, role, webhooks, organizations, find_users, projects): username, org_id = next(( (user['username'], org['id']) for webhook in webhooks for org in organizations for user in find_users(role=role, org=org['id']) if webhook['organization'] == org['id'] and webhook['type'] == 'project' and projects[webhook['project']]['owner']['id'] == user['id'] )) expected_response = [ webhook for webhook in webhooks if webhook['organization'] == org_id and webhook['type'] == 'project' and projects[webhook['project']]['owner']['username'] == username ] response = get_method(username, 'webhooks', org_id=org_id) assert response.status_code == HTTPStatus.OK assert DeepDiff(expected_response, response.json()['results'], ignore_order=True) == {} @pytest.mark.usefixtures('changedb') class TestPatchWebhooks: WID = 2 def test_sandbox_admin_can_update_any_webhook(self, webhooks, find_users): username, webhook = next(( (user['username'], deepcopy(webhook)) for user in find_users(privilege='admin') for webhook in webhooks if webhook['owner']['id'] != user['id'] and webhook['organization'] is None )) patch_data = { 'target_url': 'http://newexample.com', 'secret': 'newsecret', 'events': ['create:task'], 'is_active': not webhook['is_active'], 'enable_ssl': not webhook['enable_ssl'], } webhook.update(patch_data) response = patch_method(username, f"webhooks/{webhook['id']}", patch_data) assert response.status_code == HTTPStatus.OK assert 'secret' not in response.json() assert DeepDiff(webhook, response.json(), ignore_order=True, exclude_paths=["root['updated_date']", "root['secret']"]) == {} def test_cannot_update_with_nonexistent_contenttype(self): patch_data = { 'content_type': 'application/x-www-form-urlencoded', } response = patch_method('admin2', f'webhooks/{self.WID}', patch_data) assert response.status_code == HTTPStatus.BAD_REQUEST @pytest.mark.parametrize('privilege', ['user', 'business']) def test_sandbox_user_can_update_webhook(self, privilege, find_users, webhooks): username, webhook = next(( (user['username'], deepcopy(webhook)) for user in find_users(privilege=privilege) for webhook in webhooks if webhook['owner']['id'] == user['id'] )) patch_data = {'target_url': 'http://newexample.com'} webhook.update(patch_data) response = patch_method(username, f"webhooks/{webhook['id']}", patch_data) assert response.status_code == HTTPStatus.OK assert 'secret' not in response.json() assert DeepDiff(webhook, response.json(), ignore_order=True, exclude_paths=["root['updated_date']", "root['secret']"]) == {} @pytest.mark.parametrize('privilege', ['worker', 'user', 'business']) def test_sandbox_user_cannot_update_webhook(self, privilege, find_users, webhooks): username, webhook = next(( (user['username'], deepcopy(webhook)) for user in find_users(privilege=privilege) for webhook in webhooks if webhook['owner']['id'] != user['id'] )) patch_data = {'target_url': 'http://newexample.com'} webhook.update(patch_data) response = patch_method(username, f"webhooks/{webhook['id']}", patch_data) assert response.status_code == HTTPStatus.FORBIDDEN def test_admin_can_update_org_webhook(self, find_users, organizations, webhooks, is_org_member): org_webhooks = [w for w in webhooks if w['type'] == 'organization'] admin, oid, webhook = next(( (u['username'], o['id'], deepcopy(w)) for u in find_users(privilege='admin') for o in organizations for w in org_webhooks if w['organization'] == o['id'] and not is_org_member(u['id'], o['id']) )) patch_data = {'target_url': 'http://newexample.com'} webhook.update(patch_data) response = patch_method(admin, f"webhooks/{webhook['id']}", patch_data, org_id=oid) assert response.status_code == HTTPStatus.OK assert 'secret' not in response.json() assert DeepDiff(webhook, response.json(), ignore_order=True, exclude_paths=["root['updated_date']", "root['secret']"]) == {} @pytest.mark.parametrize('role', ['maintainer', 'owner']) def test_member_can_update_org_webhook(self, role, find_users, organizations, webhooks): org_webhooks = [w for w in webhooks if w['type'] == 'organization'] username, oid, webhook = next(( (u['username'], o['id'], deepcopy(w)) for o in organizations for u in find_users(role=role, org=o['id']) for w in org_webhooks if w['organization'] == o['id'] )) patch_data = {'target_url': 'http://newexample.com'} webhook.update(patch_data) response = patch_method(username, f"webhooks/{webhook['id']}", patch_data, org_id=oid) assert response.status_code == HTTPStatus.OK assert 'secret' not in response.json() assert DeepDiff(webhook, response.json(), ignore_order=True, exclude_paths=["root['updated_date']", "root['secret']"]) == {} @pytest.mark.parametrize('role', ['worker', 'supervisor']) def test_member_cannot_update_org_webhook(self, role, find_users, organizations, webhooks): org_webhooks = [w for w in webhooks if w['type'] == 'organization'] username, oid, webhook = next(( (u['username'], o['id'], deepcopy(w)) for o in organizations for u in find_users(role=role, org=o['id']) for w in org_webhooks if w['organization'] == o['id'] )) patch_data = {'target_url': 'http://newexample.com'} webhook.update(patch_data) response = patch_method(username, f"webhooks/{webhook['id']}", patch_data, org_id=oid) assert response.status_code == HTTPStatus.FORBIDDEN @pytest.mark.parametrize('role, allow', [ ('maintainer', True), ('owner', True), ('supervisor', False), ('worker', False) ]) def test_member_can_update_any_project_webhook_in_org(self, role, allow, find_users, organizations, projects_by_org, webhooks, is_project_staff): proj_webhooks = [w for w in webhooks if w['type'] == 'project'] username, org_id, webhook = next(( (u['username'], o['id'], deepcopy(w)) for o in organizations for u in find_users(role=role, org=o['id']) for w in proj_webhooks for p in projects_by_org.get(o['id'], []) if w['project'] == p['id'] and w['organization'] == o['id'] and not is_project_staff(u['id'], p['id']) and w['owner']['id'] != u['id'] )) patch_data = {'target_url': 'http://newexample.com'} webhook.update(patch_data) response = patch_method(username, f"webhooks/{webhook['id']}", patch_data, org_id=org_id) if not allow: assert response.status_code == HTTPStatus.FORBIDDEN else: assert response.status_code == HTTPStatus.OK assert 'secret' not in response.json() assert DeepDiff(webhook, response.json(), ignore_order=True, exclude_paths=["root['updated_date']", "root['secret']"]) == {} @pytest.mark.parametrize('role', ['supervisor']) def test_member_can_update_project_webhook_in_org(self, role, find_users, organizations, projects_by_org, webhooks): proj_webhooks = [w for w in webhooks if w['type'] == 'project'] username, org_id, webhook = next(( (u['username'], o['id'], deepcopy(w)) for o in organizations for u in find_users(role=role, org=o['id']) for w in proj_webhooks for p in projects_by_org.get(o['id'], []) if w['project'] == p['id'] and w['organization'] == o['id'] and u['id'] == p['owner']['id'] )) patch_data = {'target_url': 'http://newexample.com'} webhook.update(patch_data) response = patch_method(username, f"webhooks/{webhook['id']}", patch_data, org_id=org_id) assert response.status_code == HTTPStatus.OK assert 'secret' not in response.json() assert DeepDiff(webhook, response.json(), ignore_order=True, exclude_paths=["root['updated_date']", "root['secret']"]) == {} @pytest.mark.usefixtures('changedb') class TestDeleteWebhooks: @pytest.mark.parametrize('privilege, allow', [ ('user', False), ('business', False), ('admin', True) ]) def test_user_can_delete_project_webhook(self, privilege, allow, find_users, webhooks, projects): users = find_users(privilege=privilege) username, webhook_id = next(( (user['username'], webhook['id']) for webhook in webhooks for user in users if webhook['type'] == 'project' and webhook['organization'] is None and webhook['owner']['id'] != user['id'] and projects[webhook['project']]['owner']['id'] != user['id'] )) if not allow: response = delete_method(username, f'webhooks/{webhook_id}') assert response.status_code == HTTPStatus.FORBIDDEN else: response = delete_method(username, f'webhooks/{webhook_id}') assert response.status_code == HTTPStatus.NO_CONTENT response = get_method(username, f'webhooks/{webhook_id}') assert response.status_code == HTTPStatus.NOT_FOUND def test_admin_can_delete_project_webhook_in_org(self, find_users, webhooks, projects, is_org_member): admins = find_users(privilege='admin') username, webhook_id = next(( (user['username'], webhook['id']) for user in admins for webhook in webhooks if webhook['type'] == 'project' and webhook['organization'] is not None and webhook['owner']['id'] != user['id'] and projects[webhook['project']]['owner']['id'] != user['id'] and not is_org_member(user['id'], webhook['organization']) )) response = delete_method(username, f'webhooks/{webhook_id}') assert response.status_code == HTTPStatus.NO_CONTENT response = get_method(username, f'webhooks/{webhook_id}') assert response.status_code == HTTPStatus.NOT_FOUND def test_admin_can_delete_org_webhook(self, find_users, webhooks, is_org_member): admins = find_users(privilege='admin') username, webhook_id = next(( (user['username'], webhook['id']) for user in admins for webhook in webhooks if webhook['type'] == 'organization' and webhook['organization'] is not None and webhook['owner']['id'] != user['id'] and not is_org_member(user['id'], webhook['organization']) )) response = delete_method(username, f'webhooks/{webhook_id}') assert response.status_code == HTTPStatus.NO_CONTENT response = get_method(username, f'webhooks/{webhook_id}') assert response.status_code == HTTPStatus.NOT_FOUND @pytest.mark.parametrize('privilege', ['user', 'business']) def test_project_owner_can_delete_project_webhook(self, privilege, find_users, webhooks, projects): users = find_users(privilege=privilege) username, webhook_id = next(( (user['username'], webhook['id']) for user in users for webhook in webhooks if webhook['type'] == 'project' and webhook['organization'] is None and projects[webhook['project']]['owner']['id'] == user['id'] )) response = delete_method(username, f'webhooks/{webhook_id}') assert response.status_code == HTTPStatus.NO_CONTENT response = get_method(username, f'webhooks/{webhook_id}') assert response.status_code == HTTPStatus.NOT_FOUND @pytest.mark.parametrize('privilege', ['user', 'business']) def test_webhook_owner_can_delete_project_webhook(self, privilege, find_users, webhooks, projects): users = find_users(privilege=privilege) username, webhook_id = next(( (user['username'], webhook['id']) for user in users for webhook in webhooks if webhook['type'] == 'project' and webhook['organization'] is None and webhook['owner']['id'] == user['id'] )) response = delete_method(username, f'webhooks/{webhook_id}') assert response.status_code == HTTPStatus.NO_CONTENT response = get_method(username, f'webhooks/{webhook_id}') assert response.status_code == HTTPStatus.NOT_FOUND @pytest.mark.parametrize('role, allow', [ ('owner', True), ('maintainer', True), ('worker', False), ('supervisor', False) ]) def test_member_can_delete_org_webhook(self, role, allow, find_users, organizations, webhooks): org_webhooks = [w for w in webhooks if w['type'] == 'organization'] print(org_webhooks) username, org_id, webhook_id = next(( (user['username'], org['id'], webhook['id']) for org in organizations for webhook in org_webhooks for user in find_users(role=role, org=org['id']) if webhook['organization'] == org['id'] )) if not allow: response = delete_method(username, f'webhooks/{webhook_id}', org_id=org_id) assert response.status_code == HTTPStatus.FORBIDDEN else: response = delete_method(username, f'webhooks/{webhook_id}', org_id=org_id) assert response.status_code == HTTPStatus.NO_CONTENT response = get_method(username, f'webhooks/{webhook_id}', org_id=org_id) assert response.status_code == HTTPStatus.NOT_FOUND @pytest.mark.parametrize('role, allow', [ ('owner', True), ('maintainer', True), ('worker', False), ('supervisor', False) ]) def test_member_can_delete_project_webhook_in_org(self, role, allow, find_users, organizations, projects, webhooks): proj_webhooks = [w for w in webhooks if w['type'] == 'project'] username, org_id, webhook_id = next(( (user['username'], webhook['organization'], webhook['id']) for org in organizations for user in find_users(role=role, org=org['id']) for webhook in proj_webhooks if webhook['organization'] and webhook['organization'] == org['id'] and projects[webhook['project']]['owner']['id'] != user['id'] and webhook['owner']['id'] != user['id'] )) if not allow: response = delete_method(username, f'webhooks/{webhook_id}', org_id=org_id) assert response.status_code == HTTPStatus.FORBIDDEN else: response = delete_method(username, f'webhooks/{webhook_id}', org_id=org_id) assert response.status_code == HTTPStatus.NO_CONTENT response = get_method(username, f'webhooks/{webhook_id}', org_id=org_id) assert response.status_code == HTTPStatus.NOT_FOUND @pytest.mark.parametrize('role', ['supervisor']) def test_member_webhook_staff_can_delete_project_webhook_in_org(self, role, find_users, organizations, projects, webhooks): proj_webhooks = [w for w in webhooks if w['type'] == 'project'] username, org_id, webhook_id = next(( (user['username'], webhook['organization'], webhook['id']) for org in organizations for user in find_users(role=role, org=org['id']) for webhook in proj_webhooks if webhook['organization'] and webhook['organization'] == org['id'] and (projects[webhook['project']]['owner']['id'] == user['id'] or webhook['owner']['id'] == user['id']) )) response = delete_method(username, f'webhooks/{webhook_id}', org_id=org_id) assert response.status_code == HTTPStatus.NO_CONTENT response = get_method(username, f'webhooks/{webhook_id}', org_id=org_id) assert response.status_code == HTTPStatus.NOT_FOUND