| |
| |
| """ |
| (c) 2015-2018 - Copyright Red Hat Inc |
| |
| Authors: |
| Pierre-Yves Chibon <pingou@pingoured.fr> |
| Karsten Hopp <karsten@redhat.com> |
| |
| """ |
| |
| from __future__ import unicode_literals, absolute_import |
| |
| import datetime |
| import json |
| import unittest |
| import shutil |
| import sys |
| import tempfile |
| import os |
| |
| import pygit2 |
| from celery.result import EagerResult |
| from mock import patch, Mock |
| |
| sys.path.insert(0, os.path.join(os.path.dirname( |
| os.path.abspath(__file__)), '..')) |
| |
| import pagure.flask_app |
| import pagure.lib.query |
| import tests |
| from pagure.lib.repo import PagureRepo |
| |
| |
| class PagureFlaskApiProjecttests(tests.Modeltests): |
| """ Tests for the flask API of pagure for issue """ |
| |
| def setUp(self): |
| super(PagureFlaskApiProjecttests, self).setUp() |
| self.gga_patcher = patch( |
| 'pagure.lib.tasks.generate_gitolite_acls.delay') |
| self.mock_gen_acls = self.gga_patcher.start() |
| task_result = EagerResult('abc-1234', True, "SUCCESS") |
| self.mock_gen_acls.return_value = task_result |
| |
| def tearDown(self): |
| self.gga_patcher.stop() |
| super(PagureFlaskApiProjecttests, self).tearDown() |
| |
| def test_api_git_tags(self): |
| """ Test the api_git_tags method of the flask api. """ |
| tests.create_projects(self.session) |
| |
| |
| gitrepo = os.path.join(self.path, 'repos', 'test.git') |
| repo = pygit2.init_repository(gitrepo, bare=True) |
| |
| newpath = tempfile.mkdtemp(prefix='pagure-fork-test') |
| repopath = os.path.join(newpath, 'test') |
| clone_repo = pygit2.clone_repository(gitrepo, repopath) |
| |
| |
| with open(os.path.join(repopath, 'sources'), 'w') as stream: |
| stream.write('foo\n bar') |
| clone_repo.index.add('sources') |
| clone_repo.index.write() |
| |
| |
| tree = clone_repo.index.write_tree() |
| author = pygit2.Signature( |
| 'Alice Author', 'alice@authors.tld') |
| committer = pygit2.Signature( |
| 'Cecil Committer', 'cecil@committers.tld') |
| clone_repo.create_commit( |
| 'refs/heads/master', |
| author, |
| committer, |
| 'Add sources file for testing', |
| |
| tree, |
| |
| [] |
| ) |
| refname = 'refs/heads/master:refs/heads/master' |
| ori_remote = clone_repo.remotes[0] |
| PagureRepo.push(ori_remote, refname) |
| |
| |
| first_commit = repo.revparse_single('HEAD') |
| tagger = pygit2.Signature('Alice Doe', 'adoe@example.com', 12347, 0) |
| repo.create_tag( |
| "0.0.1", first_commit.oid.hex, pygit2.GIT_OBJ_COMMIT, tagger, |
| "Release 0.0.1") |
| |
| |
| output = self.app.get('/api/0/test/git/tags') |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| {'tags': ['0.0.1'], 'total_tags': 1} |
| ) |
| |
| |
| output = self.app.get('/api/0/test/git/tags?with_commits=True') |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data['tags']['0.0.1'] = 'bb8fa2aa199da08d6085e1c9badc3d83d188d38c' |
| self.assertDictEqual( |
| data, |
| { |
| 'tags': {'0.0.1': 'bb8fa2aa199da08d6085e1c9badc3d83d188d38c'}, |
| 'total_tags': 1} |
| ) |
| |
| shutil.rmtree(newpath) |
| |
| def test_api_git_branches(self): |
| """ Test the api_git_branches method of the flask api. """ |
| |
| tests.create_projects(self.session) |
| repo_path = os.path.join(self.path, 'repos', 'test.git') |
| tests.add_content_git_repo(repo_path) |
| new_repo_path = tempfile.mkdtemp(prefix='pagure-api-git-branches-test') |
| clone_repo = pygit2.clone_repository(repo_path, new_repo_path) |
| |
| |
| for branch in ['pats-win-49', 'pats-win-51']: |
| clone_repo.create_branch(branch, clone_repo.head.peel()) |
| refname = 'refs/heads/{0}:refs/heads/{0}'.format(branch) |
| PagureRepo.push(clone_repo.remotes[0], refname) |
| |
| |
| output = self.app.get('/api/0/test/git/branches') |
| |
| shutil.rmtree(new_repo_path) |
| |
| |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| 'branches': ['master', 'pats-win-49', 'pats-win-51'], |
| 'total_branches': 3 |
| } |
| ) |
| |
| def test_api_git_branches_empty_repo(self): |
| """ Test the api_git_branches method of the flask api when the repo is |
| empty. |
| """ |
| |
| tests.create_projects(self.session) |
| repo_base_path = os.path.join(self.path, 'repos') |
| tests.create_projects_git(repo_base_path) |
| |
| |
| output = self.app.get('/api/0/test/git/branches') |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| 'branches': [], |
| 'total_branches': 0 |
| } |
| ) |
| |
| def test_api_git_branches_no_repo(self): |
| """ Test the api_git_branches method of the flask api when there is no |
| repo on a project. |
| """ |
| tests.create_projects(self.session) |
| output = self.app.get('/api/0/test/git/branches') |
| self.assertEqual(output.status_code, 404) |
| |
| def test_api_git_urls(self): |
| """ Test the api_project_git_urls method of the flask api. |
| """ |
| tests.create_projects(self.session) |
| output = self.app.get('/api/0/test/git/urls') |
| self.assertEqual(output.status_code, 200) |
| expected_rv = { |
| 'urls': { |
| 'git': 'git://localhost.localdomain/test.git', |
| 'ssh': 'ssh://git@localhost.localdomain/test.git' |
| }, |
| 'total_urls': 2 |
| } |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(data, expected_rv) |
| |
| def test_api_git_urls_no_project(self): |
| """ Test the api_project_git_urls method of the flask api when there is |
| no project. |
| """ |
| output = self.app.get('/api/0/test1234/git/urls') |
| self.assertEqual(output.status_code, 404) |
| expected_rv = { |
| 'error': 'Project not found', |
| 'error_code': 'ENOPROJECT' |
| } |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(data, expected_rv) |
| |
| @patch.dict('pagure.config.config', {'PRIVATE_PROJECTS': True}) |
| def test_api_git_urls_private_project(self): |
| """ Test the api_project_git_urls method of the flask api when the |
| project is private. |
| """ |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session, 'aaabbbcccddd') |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| |
| test_project = pagure.lib.query._get_project(self.session, 'test') |
| test_project.private = True |
| self.session.add(test_project) |
| self.session.commit() |
| |
| output = self.app.get('/api/0/test/git/urls', headers=headers) |
| self.assertEqual(output.status_code, 200) |
| expected_rv = { |
| 'urls': { |
| 'git': 'git://localhost.localdomain/test.git', |
| 'ssh': 'ssh://git@localhost.localdomain/test.git' |
| }, |
| 'total_urls': 2 |
| } |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(data, expected_rv) |
| |
| @patch.dict('pagure.config.config', {'PRIVATE_PROJECTS': True}) |
| def test_api_git_urls_private_project_no_login(self): |
| """ Test the api_project_git_urls method of the flask api when the |
| project is private and the user is not logged in. |
| """ |
| tests.create_projects(self.session) |
| test_project = pagure.lib.query._get_project(self.session, 'test') |
| test_project.private = True |
| self.session.add(test_project) |
| self.session.commit() |
| |
| output = self.app.get('/api/0/test/git/urls') |
| self.assertEqual(output.status_code, 404) |
| expected_rv = { |
| 'error': 'Project not found', |
| 'error_code': 'ENOPROJECT' |
| } |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(data, expected_rv) |
| |
| def test_api_projects_pattern(self): |
| """ Test the api_projects method of the flask api. """ |
| tests.create_projects(self.session) |
| |
| output = self.app.get('/api/0/projects?pattern=test') |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data['projects'][0]['date_created'] = "1436527638" |
| data['projects'][0]['date_modified'] = "1436527638" |
| del data['pagination'] |
| expected_data = { |
| "args": { |
| "fork": None, |
| "namespace": None, |
| "owner": None, |
| "page": 1, |
| "pattern": "test", |
| "per_page": 20, |
| "short": False, |
| "tags": [], |
| "username": None |
| }, |
| "projects": [ |
| { |
| "access_groups": { |
| "admin": [], |
| "commit": [], |
| "ticket": [] |
| }, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": [ |
| "pingou" |
| ], |
| "ticket": [] |
| }, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate" |
| ], |
| "custom_keys": [], |
| "date_created": "1436527638", |
| "date_modified": "1436527638", |
| "description": "test project #1", |
| "fullname": "test", |
| "url_path": "test", |
| "id": 1, |
| "milestones": {}, |
| "name": "test", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": [], |
| "user": { |
| "fullname": "PY C", |
| "name": "pingou" |
| } |
| } |
| ], |
| "total_projects": 1 |
| } |
| self.assertDictEqual(data, expected_data) |
| |
| def test_api_projects_pattern_short(self): |
| """ Test the api_projects method of the flask api. """ |
| tests.create_projects(self.session) |
| |
| output = self.app.get('/api/0/projects?pattern=te*&short=1') |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| del data['pagination'] |
| expected_data = { |
| "args": { |
| "fork": None, |
| "namespace": None, |
| "owner": None, |
| "page": 1, |
| "pattern": "te*", |
| "per_page": 20, |
| "short": True, |
| "tags": [], |
| "username": None |
| }, |
| "projects": [ |
| { |
| "description": "test project #1", |
| "fullname": "test", |
| "name": "test", |
| "namespace": None |
| }, |
| { |
| "description": "test project #2", |
| "fullname": "test2", |
| "name": "test2", |
| "namespace": None |
| }, |
| { |
| "description": "namespaced test project", |
| "fullname": "somenamespace/test3", |
| "name": "test3", |
| "namespace": "somenamespace" |
| } |
| ], |
| "total_projects": 3 |
| } |
| self.maxDiff = None |
| self.assertDictEqual(data, expected_data) |
| |
| def test_api_projects_owner(self): |
| """ Test the api_projects method of the flask api. """ |
| tests.create_projects(self.session) |
| |
| output = self.app.get('/api/0/projects?owner=foo') |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| del data['pagination'] |
| expected_data = { |
| "args": { |
| "fork": None, |
| "namespace": None, |
| "owner": "foo", |
| "page": 1, |
| "pattern": None, |
| "per_page": 20, |
| "short": False, |
| "tags": [], |
| "username": None |
| }, |
| "projects": [], |
| "total_projects": 0 |
| } |
| self.maxDiff = None |
| self.assertDictEqual(data, expected_data) |
| |
| def test_api_projects_not_owner(self): |
| """ Test the api_projects method of the flask api. """ |
| tests.create_projects(self.session) |
| |
| output = self.app.get('/api/0/projects?owner=!foo&short=1') |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| del data['pagination'] |
| expected_data = { |
| "args": { |
| "fork": None, |
| "namespace": None, |
| "owner": "!foo", |
| "page": 1, |
| "pattern": None, |
| "per_page": 20, |
| "short": True, |
| "tags": [], |
| "username": None |
| }, |
| "projects": [ |
| { |
| "description": "test project #1", |
| "fullname": "test", |
| "name": "test", |
| "namespace": None |
| }, |
| { |
| "description": "test project #2", |
| "fullname": "test2", |
| "name": "test2", |
| "namespace": None |
| }, |
| { |
| "description": "namespaced test project", |
| "fullname": "somenamespace/test3", |
| "name": "test3", |
| "namespace": "somenamespace" |
| } |
| ], |
| "total_projects": 3 |
| } |
| self.maxDiff = None |
| self.assertDictEqual(data, expected_data) |
| |
| def test_api_projects(self): |
| """ Test the api_projects method of the flask api. """ |
| tests.create_projects(self.session) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, 'test') |
| self.assertEqual(repo.tags, []) |
| |
| |
| output = pagure.lib.query.update_tags( |
| self.session, repo, 'infra', 'pingou') |
| self.assertEqual(output, ['Project tagged with: infra']) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, 'test') |
| self.assertEqual(len(repo.tags), 1) |
| self.assertEqual(repo.tags_text, ['infra']) |
| |
| |
| output = self.app.get('/api/0/projects?tags=inf') |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| null = None |
| del data['pagination'] |
| self.assertDictEqual( |
| data, |
| { |
| "total_projects": 0, |
| "projects": [], |
| "args": { |
| "fork": None, |
| "namespace": None, |
| "owner": None, |
| "page": 1, |
| "pattern": None, |
| "per_page": 20, |
| "short": False, |
| "tags": ["inf"], |
| "username": None |
| }, |
| } |
| ) |
| output = self.app.get('/api/0/projects?tags=infra') |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data['projects'][0]['date_created'] = "1436527638" |
| data['projects'][0]['date_modified'] = "1436527638" |
| del data['pagination'] |
| expected_data = { |
| "args": { |
| "fork": None, |
| "namespace": None, |
| "owner": None, |
| "page": 1, |
| "pattern": None, |
| "per_page": 20, |
| "short": False, |
| "tags": ["infra"], |
| "username": None |
| }, |
| "projects": [{ |
| "access_groups": { |
| "admin": [], |
| "commit": [], |
| "ticket": []}, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": ["pingou"], |
| "ticket": []}, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate" |
| ], |
| "custom_keys": [], |
| "date_created": "1436527638", |
| "date_modified": "1436527638", |
| "description": "test project #1", |
| "fullname": "test", |
| "url_path": "test", |
| "id": 1, |
| "milestones": {}, |
| "name": "test", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": ["infra"], |
| "user": { |
| "fullname": "PY C", |
| "name": "pingou" |
| } |
| }], |
| "total_projects": 1 |
| } |
| self.assertDictEqual(data, expected_data) |
| |
| output = self.app.get('/api/0/projects?owner=pingou') |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data['projects'][0]['date_created'] = "1436527638" |
| data['projects'][0]['date_modified'] = "1436527638" |
| data['projects'][1]['date_created'] = "1436527638" |
| data['projects'][1]['date_modified'] = "1436527638" |
| data['projects'][2]['date_created'] = "1436527638" |
| data['projects'][2]['date_modified'] = "1436527638" |
| del data['pagination'] |
| expected_data = { |
| "args": { |
| "fork": None, |
| "namespace": None, |
| "owner": "pingou", |
| "page": 1, |
| "pattern": None, |
| "per_page": 20, |
| "short": False, |
| "tags": [], |
| "username": None |
| }, |
| "projects": [ |
| { |
| "access_groups": { |
| "admin": [], |
| "commit": [], |
| "ticket": [] |
| }, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": ["pingou"], |
| "ticket": [] |
| }, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate" |
| ], |
| "custom_keys": [], |
| "date_created": "1436527638", |
| "date_modified": "1436527638", |
| "description": "test project #1", |
| "fullname": "test", |
| "url_path": "test", |
| "id": 1, |
| "milestones": {}, |
| "name": "test", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": ["infra"], |
| "user": { |
| "fullname": "PY C", |
| "name": "pingou" |
| } |
| }, |
| { |
| "access_groups": { |
| "admin": [], |
| "commit": [], |
| "ticket": [] |
| }, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": ["pingou"], |
| "ticket": [] |
| }, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate" |
| ], |
| "custom_keys": [], |
| "date_created": "1436527638", |
| "date_modified": "1436527638", |
| "description": "test project #2", |
| "fullname": "test2", |
| "url_path": "test2", |
| "id": 2, |
| "milestones": {}, |
| "name": "test2", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": [], |
| "user": { |
| "fullname": "PY C", |
| "name": "pingou" |
| } |
| }, |
| { |
| "access_groups": { |
| "admin": [], |
| "commit": [], |
| "ticket": [] |
| }, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": ["pingou"], |
| "ticket": [] |
| }, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate" |
| ], |
| "custom_keys": [], |
| "date_created": "1436527638", |
| "date_modified": "1436527638", |
| "description": "namespaced test project", |
| "fullname": "somenamespace/test3", |
| "url_path": "somenamespace/test3", |
| "id": 3, |
| "milestones": {}, |
| "name": "test3", |
| "namespace": "somenamespace", |
| "parent": None, |
| "priorities": {}, |
| "tags": [], |
| "user": { |
| "fullname": "PY C", |
| "name": "pingou" |
| } |
| } |
| ], |
| "total_projects": 3 |
| } |
| self.assertDictEqual(data, expected_data) |
| |
| output = self.app.get('/api/0/projects?username=pingou') |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data['projects'][0]['date_created'] = "1436527638" |
| data['projects'][0]['date_modified'] = "1436527638" |
| data['projects'][1]['date_created'] = "1436527638" |
| data['projects'][1]['date_modified'] = "1436527638" |
| data['projects'][2]['date_created'] = "1436527638" |
| data['projects'][2]['date_modified'] = "1436527638" |
| del data['pagination'] |
| expected_data = { |
| "args": { |
| "fork": None, |
| "namespace": None, |
| "owner": None, |
| "page": 1, |
| "pattern": None, |
| "per_page": 20, |
| "short": False, |
| "tags": [], |
| "username": "pingou" |
| }, |
| "projects": [ |
| { |
| "access_groups": { |
| "admin": [], |
| "commit": [], |
| "ticket": []}, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": ["pingou"], |
| "ticket": [] |
| }, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate" |
| ], |
| "custom_keys": [], |
| "date_created": "1436527638", |
| "date_modified": "1436527638", |
| "description": "test project #1", |
| "fullname": "test", |
| "url_path": "test", |
| "id": 1, |
| "milestones": {}, |
| "name": "test", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": ["infra"], |
| "user": { |
| "fullname": "PY C", |
| "name": "pingou" |
| } |
| }, |
| { |
| "access_groups": { |
| "admin": [], |
| "commit": [], |
| "ticket": [] |
| }, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": ["pingou"], |
| "ticket": [] |
| }, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate" |
| ], |
| "custom_keys": [], |
| "date_created": "1436527638", |
| "date_modified": "1436527638", |
| "description": "test project #2", |
| "fullname": "test2", |
| "url_path": "test2", |
| "id": 2, |
| "milestones": {}, |
| "name": "test2", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": [], |
| "user": { |
| "fullname": "PY C", |
| "name": "pingou" |
| } |
| }, |
| { |
| "access_groups": { |
| "admin": [], |
| "commit": [], |
| "ticket": []}, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": ["pingou"], |
| "ticket": []}, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate" |
| ], |
| "custom_keys": [], |
| "date_created": "1436527638", |
| "date_modified": "1436527638", |
| "description": "namespaced test project", |
| "fullname": "somenamespace/test3", |
| "url_path": "somenamespace/test3", |
| "id": 3, |
| "milestones": {}, |
| "name": "test3", |
| "namespace": "somenamespace", |
| "parent": None, |
| "priorities": {}, |
| "tags": [], |
| "user": { |
| "fullname": "PY C", |
| "name": "pingou" |
| } |
| } |
| ], |
| "total_projects": 3 |
| } |
| self.assertDictEqual(data, expected_data) |
| |
| output = self.app.get('/api/0/projects?username=pingou&tags=infra') |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data['projects'][0]['date_created'] = "1436527638" |
| data['projects'][0]['date_modified'] = "1436527638" |
| del data['pagination'] |
| expected_data = { |
| "args": { |
| "fork": None, |
| "namespace": None, |
| "owner": None, |
| "page": 1, |
| "pattern": None, |
| "per_page": 20, |
| "short": False, |
| "tags": ["infra"], |
| "username": "pingou", |
| }, |
| "projects": [{ |
| "access_groups": { |
| "admin": [], |
| "commit": [], |
| "ticket": [] |
| }, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": ["pingou"], |
| "ticket": []}, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate"], |
| "custom_keys": [], |
| "date_created": "1436527638", |
| "date_modified": "1436527638", |
| "description": "test project #1", |
| "fullname": "test", |
| "url_path": "test", |
| "id": 1, |
| "milestones": {}, |
| "name": "test", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": ["infra"], |
| "user": { |
| "fullname": "PY C", |
| "name": "pingou" |
| } |
| }], |
| "total_projects": 1 |
| } |
| self.assertDictEqual(data, expected_data) |
| |
| output = self.app.get('/api/0/projects?namespace=somenamespace') |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data['projects'][0]['date_created'] = "1436527638" |
| data['projects'][0]['date_modified'] = "1436527638" |
| del data['pagination'] |
| expected_data = { |
| "args": { |
| "fork": None, |
| "owner": None, |
| "page": 1, |
| "namespace": "somenamespace", |
| "per_page": 20, |
| "pattern": None, |
| "short": False, |
| "tags": [], |
| "username": None |
| }, |
| "projects": [ |
| { |
| "access_groups": { |
| "admin": [], |
| "commit": [], |
| "ticket": []}, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": ["pingou"], |
| "ticket": []}, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate" |
| ], |
| "custom_keys": [], |
| "date_created": "1436527638", |
| "date_modified": "1436527638", |
| "description": "namespaced test project", |
| "fullname": "somenamespace/test3", |
| "url_path": "somenamespace/test3", |
| "id": 3, |
| "milestones": {}, |
| "name": "test3", |
| "namespace": "somenamespace", |
| "parent": None, |
| "priorities": {}, |
| "tags": [], |
| "user": { |
| "fullname": "PY C", |
| "name": "pingou" |
| } |
| } |
| ], |
| "total_projects": 1 |
| } |
| self.assertDictEqual(data, expected_data) |
| |
| def test_api_project(self): |
| """ Test the api_project method of the flask api. """ |
| tests.create_projects(self.session) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, 'test') |
| self.assertEqual(repo.tags, []) |
| |
| |
| output = pagure.lib.query.update_tags( |
| self.session, repo, 'infra', 'pingou') |
| self.assertEqual(output, ['Project tagged with: infra']) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, 'test') |
| self.assertEqual(len(repo.tags), 1) |
| self.assertEqual(repo.tags_text, ['infra']) |
| |
| |
| |
| |
| output = self.app.get('/api/0/random') |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| {'error_code': 'ENOPROJECT', 'error': 'Project not found'} |
| ) |
| |
| |
| output = self.app.get('/api/0/test') |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data['date_created'] = "1436527638" |
| data['date_modified'] = "1436527638" |
| expected_data ={ |
| "access_groups": { |
| "admin": [], |
| "commit": [], |
| "ticket": [] |
| }, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": ["pingou"], |
| "ticket": []}, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate" |
| ], |
| "custom_keys": [], |
| "date_created": "1436527638", |
| "date_modified": "1436527638", |
| "description": "test project #1", |
| "fullname": "test", |
| "url_path": "test", |
| "id": 1, |
| "milestones": {}, |
| "name": "test", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": ["infra"], |
| "user": { |
| "fullname": "PY C", |
| "name": "pingou" |
| } |
| } |
| self.assertDictEqual(data, expected_data) |
| |
| def test_api_project_group(self): |
| """ Test the api_project method of the flask api. """ |
| tests.create_projects(self.session) |
| repo = pagure.lib.query.get_authorized_project(self.session, 'test') |
| |
| |
| output = pagure.lib.query.update_tags( |
| self.session, repo, 'infra', 'pingou') |
| self.assertEqual(output, ['Project tagged with: infra']) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, 'test') |
| self.assertEqual(len(repo.tags), 1) |
| self.assertEqual(repo.tags_text, ['infra']) |
| |
| |
| msg = pagure.lib.query.add_group( |
| self.session, |
| group_name='some_group', |
| display_name='Some Group', |
| description=None, |
| group_type='bar', |
| user='foo', |
| is_admin=False, |
| blacklist=[], |
| ) |
| self.session.commit() |
| |
| project = pagure.lib.query.get_authorized_project(self.session, 'test') |
| group = pagure.lib.query.search_groups( |
| self.session, group_name='some_group') |
| |
| pagure.lib.query.add_group_to_project( |
| self.session, |
| project, |
| new_group='some_group', |
| user='pingou', |
| access='commit', |
| create=False, |
| is_admin=True |
| ) |
| self.session.commit() |
| |
| |
| |
| |
| output = self.app.get('/api/0/test?expand_group=1') |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data['date_created'] = "1436527638" |
| data['date_modified'] = "1436527638" |
| expected_data ={ |
| "access_groups": { |
| "admin": [], |
| "commit": ["some_group"], |
| "ticket": [] |
| }, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": ["pingou"], |
| "ticket": []}, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate" |
| ], |
| "custom_keys": [], |
| "date_created": "1436527638", |
| "date_modified": "1436527638", |
| "description": "test project #1", |
| "fullname": "test", |
| "url_path": "test", |
| "group_details": { |
| "some_group": [ |
| "foo" |
| ] |
| }, |
| "id": 1, |
| "milestones": {}, |
| "name": "test", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": ["infra"], |
| "user": { |
| "fullname": "PY C", |
| "name": "pingou" |
| } |
| } |
| self.assertDictEqual(data, expected_data) |
| |
| def test_api_project_group_but_no_group(self): |
| """ Test the api_project method of the flask api when asking for |
| group details while there are none associated. |
| """ |
| tests.create_projects(self.session) |
| repo = pagure.lib.query.get_authorized_project(self.session, 'test') |
| |
| |
| output = pagure.lib.query.update_tags( |
| self.session, repo, 'infra', 'pingou') |
| self.assertEqual(output, ['Project tagged with: infra']) |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, 'test') |
| self.assertEqual(len(repo.tags), 1) |
| self.assertEqual(repo.tags_text, ['infra']) |
| |
| |
| |
| |
| output = self.app.get('/api/0/test?expand_group=0') |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data['date_created'] = "1436527638" |
| data['date_modified'] = "1436527638" |
| expected_data ={ |
| "access_groups": { |
| "admin": [], |
| "commit": [], |
| "ticket": [] |
| }, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": ["pingou"], |
| "ticket": []}, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate" |
| ], |
| "custom_keys": [], |
| "date_created": "1436527638", |
| "date_modified": "1436527638", |
| "description": "test project #1", |
| "fullname": "test", |
| "url_path": "test", |
| "id": 1, |
| "milestones": {}, |
| "name": "test", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": ["infra"], |
| "user": { |
| "fullname": "PY C", |
| "name": "pingou" |
| } |
| } |
| self.assertDictEqual(data, expected_data) |
| |
| def test_api_projects_pagination(self): |
| """ Test the api_projects method of the flask api with pagination. """ |
| tests.create_projects(self.session) |
| |
| output = self.app.get('/api/0/projects?page=1') |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| for i in range(3): |
| data['projects'][i]['date_created'] = "1436527638" |
| data['projects'][i]['date_modified'] = "1436527638" |
| expected_data = { |
| "args": { |
| "fork": None, |
| "namespace": None, |
| "owner": None, |
| "page": 1, |
| "per_page": 20, |
| "pattern": None, |
| "short": False, |
| "tags": [], |
| "username": None |
| }, |
| "pagination": { |
| "next": None, |
| "page": 1, |
| "pages": 1, |
| "per_page": 20, |
| "prev": None |
| }, |
| "projects": [ |
| { |
| "access_groups": { |
| "admin": [], |
| "commit": [], |
| "ticket": []}, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": ["pingou"], |
| "ticket": [] |
| }, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate" |
| ], |
| "custom_keys": [], |
| "date_created": "1436527638", |
| "date_modified": "1436527638", |
| "description": "test project #1", |
| "fullname": "test", |
| "url_path": "test", |
| "id": 1, |
| "milestones": {}, |
| "name": "test", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": [], |
| "user": { |
| "fullname": "PY C", |
| "name": "pingou" |
| } |
| }, |
| { |
| "access_groups": { |
| "admin": [], |
| "commit": [], |
| "ticket": [] |
| }, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": ["pingou"], |
| "ticket": [] |
| }, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate" |
| ], |
| "custom_keys": [], |
| "date_created": "1436527638", |
| "date_modified": "1436527638", |
| "description": "test project #2", |
| "fullname": "test2", |
| "url_path": "test2", |
| "id": 2, |
| "milestones": {}, |
| "name": "test2", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": [], |
| "user": { |
| "fullname": "PY C", |
| "name": "pingou" |
| } |
| }, |
| { |
| "access_groups": { |
| "admin": [], |
| "commit": [], |
| "ticket": []}, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": ["pingou"], |
| "ticket": []}, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate" |
| ], |
| "custom_keys": [], |
| "date_created": "1436527638", |
| "date_modified": "1436527638", |
| "description": "namespaced test project", |
| "fullname": "somenamespace/test3", |
| "url_path": "somenamespace/test3", |
| "id": 3, |
| "milestones": {}, |
| "name": "test3", |
| "namespace": "somenamespace", |
| "parent": None, |
| "priorities": {}, |
| "tags": [], |
| "user": { |
| "fullname": "PY C", |
| "name": "pingou" |
| } |
| } |
| ], |
| "total_projects": 3 |
| } |
| |
| self.assertURLEqual( |
| data["pagination"].pop("first"), |
| "http://localhost/api/0/projects?per_page=20&page=1", |
| ) |
| self.assertURLEqual( |
| data["pagination"].pop("last"), |
| "http://localhost/api/0/projects?per_page=20&page=1", |
| ) |
| self.assertDictEqual(data, expected_data) |
| |
| def test_api_projects_pagination_per_page(self): |
| """ Test the api_projects method of the flask api with pagination and |
| the `per_page` argument set. """ |
| tests.create_projects(self.session) |
| |
| output = self.app.get('/api/0/projects?page=2&per_page=2') |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data['projects'][0]['date_created'] = "1436527638" |
| data['projects'][0]['date_modified'] = "1436527638" |
| expected_data = { |
| "args": { |
| "fork": None, |
| "namespace": None, |
| "owner": None, |
| "page": 2, |
| "per_page": 2, |
| "pattern": None, |
| "short": False, |
| "tags": [], |
| "username": None |
| }, |
| "pagination": { |
| "next": None, |
| "page": 2, |
| "pages": 2, |
| "per_page": 2, |
| }, |
| "projects": [ |
| { |
| "access_groups": { |
| "admin": [], |
| "commit": [], |
| "ticket": [] |
| }, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": ["pingou"], |
| "ticket": [] |
| }, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate" |
| ], |
| "custom_keys": [], |
| "date_created": "1436527638", |
| "date_modified": "1436527638", |
| "description": "namespaced test project", |
| "fullname": "somenamespace/test3", |
| "url_path": "somenamespace/test3", |
| "id": 3, |
| "milestones": {}, |
| "name": "test3", |
| "namespace": "somenamespace", |
| "parent": None, |
| "priorities": {}, |
| "tags": [], |
| "user": { |
| "fullname": "PY C", |
| "name": "pingou" |
| } |
| } |
| ], |
| "total_projects": 3 |
| } |
| self.assertURLEqual( |
| data["pagination"].pop("first"), |
| "http://localhost/api/0/projects?per_page=2&page=1", |
| ) |
| self.assertURLEqual( |
| data["pagination"].pop("prev"), |
| "http://localhost/api/0/projects?per_page=2&page=1", |
| ) |
| self.assertURLEqual( |
| data["pagination"].pop("last"), |
| "http://localhost/api/0/projects?per_page=2&page=2", |
| ) |
| self.assertDictEqual(data, expected_data) |
| |
| def test_api_projects_pagination_invalid_page(self): |
| """ Test the api_projects method of the flask api when an invalid page |
| value is entered. """ |
| tests.create_projects(self.session) |
| |
| output = self.app.get('/api/0/projects?page=-3') |
| self.assertEqual(output.status_code, 400) |
| |
| def test_api_projects_pagination_invalid_page_str(self): |
| """ Test the api_projects method of the flask api when an invalid type |
| for the page value is entered. """ |
| tests.create_projects(self.session) |
| |
| output = self.app.get('/api/0/projects?page=abcd') |
| self.assertEqual(output.status_code, 400) |
| |
| def test_api_projects_pagination_invalid_per_page_too_low(self): |
| """ Test the api_projects method of the flask api when a per_page |
| value is below 1. """ |
| tests.create_projects(self.session) |
| |
| output = self.app.get('/api/0/projects?page=1&per_page=0') |
| self.assertEqual(output.status_code, 400) |
| error = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| error['error'], 'The per_page value must be between 1 and 100') |
| |
| def test_api_projects_pagination_invalid_per_page_too_high(self): |
| """ Test the api_projects method of the flask api when a per_page |
| value is above 100. """ |
| tests.create_projects(self.session) |
| |
| output = self.app.get('/api/0/projects?page=1&per_page=101') |
| self.assertEqual(output.status_code, 400) |
| error = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| error['error'], 'The per_page value must be between 1 and 100') |
| |
| def test_api_projects_pagination_invalid_per_page_str(self): |
| """ Test the api_projects method of the flask api when an invalid type |
| for the per_page value is entered. """ |
| tests.create_projects(self.session) |
| |
| output = self.app.get('/api/0/projects?page=1&per_page=abcd') |
| self.assertEqual(output.status_code, 400) |
| |
| def test_api_projects_pagination_beyond_last_page(self): |
| """ Test the api_projects method of the flask api when a page value |
| that is larger than the last page is entered. """ |
| tests.create_projects(self.session) |
| |
| output = self.app.get('/api/0/projects?page=99999') |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertURLEqual( |
| data["pagination"].pop("first"), |
| "http://localhost/api/0/projects?per_page=20&page=1", |
| ) |
| self.assertURLEqual( |
| data["pagination"].pop("last"), |
| "http://localhost/api/0/projects?per_page=20&page=1", |
| ) |
| self.assertURLEqual( |
| data["pagination"].pop("prev"), |
| "http://localhost/api/0/projects?per_page=20&page=99998", |
| ) |
| self.assertEqual( |
| data, |
| { |
| "args": { |
| "fork": None, |
| "namespace": None, |
| "owner": None, |
| "page": 99999, |
| "pattern": None, |
| "per_page": 20, |
| "short": False, |
| "tags": [], |
| "username": None |
| }, |
| "pagination": { |
| "next": None, |
| "page": 99999, |
| "pages": 1, |
| "per_page": 20, |
| }, |
| "projects": [], |
| "total_projects": 3 |
| } |
| ) |
| |
| def test_api_modify_project_main_admin(self): |
| """ Test the api_modify_project method of the flask api when the |
| request is to change the main_admin of the project. """ |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session, project_id=None) |
| tests.create_tokens_acl(self.session, 'aaabbbcccddd', 'modify_project') |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| |
| output = self.app.patch( |
| '/api/0/test', headers=headers, |
| data={'main_admin': 'foo'}) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data['date_created'] = '1496338274' |
| data['date_modified'] = '1496338274' |
| expected_output = { |
| "access_groups": { |
| "admin": [], |
| "commit": [], |
| "ticket": [] |
| }, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": [ |
| "foo" |
| ], |
| "ticket": [] |
| }, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate" |
| ], |
| "custom_keys": [], |
| "date_created": "1496338274", |
| "date_modified": "1496338274", |
| "description": "test project #1", |
| "fullname": "test", |
| "url_path": "test", |
| "id": 1, |
| "milestones": {}, |
| "name": "test", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": [], |
| "user": { |
| "default_email": "foo@bar.com", |
| "emails": [ |
| "foo@bar.com" |
| ], |
| "fullname": "foo bar", |
| "name": "foo" |
| } |
| } |
| self.assertEqual(data, expected_output) |
| |
| def test_api_modify_project_main_admin_retain_access(self): |
| """ Test the api_modify_project method of the flask api when the |
| request is to change the main_admin of the project and retain_access |
| is true. """ |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session, project_id=None) |
| tests.create_tokens_acl(self.session, 'aaabbbcccddd', 'modify_project') |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| |
| output = self.app.patch( |
| '/api/0/test', headers=headers, |
| data={'main_admin': 'foo', 'retain_access': True}) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data['date_created'] = '1496338274' |
| data['date_modified'] = '1496338274' |
| expected_output = { |
| "access_groups": { |
| "admin": [], |
| "commit": [], |
| "ticket": [] |
| }, |
| "access_users": { |
| "admin": [ |
| "pingou" |
| ], |
| "commit": [], |
| "owner": [ |
| "foo" |
| ], |
| "ticket": [] |
| }, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate" |
| ], |
| "custom_keys": [], |
| "date_created": "1496338274", |
| "date_modified": "1496338274", |
| "description": "test project #1", |
| "fullname": "test", |
| "url_path": "test", |
| "id": 1, |
| "milestones": {}, |
| "name": "test", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": [], |
| "user": { |
| "default_email": "foo@bar.com", |
| "emails": [ |
| "foo@bar.com" |
| ], |
| "fullname": "foo bar", |
| "name": "foo" |
| } |
| } |
| self.assertEqual(data, expected_output) |
| |
| def test_api_modify_project_main_admin_retain_access_already_user(self): |
| """ Test the api_modify_project method of the flask api when the |
| request is to change the main_admin of the project and retain_access |
| is true and the user becoming the main_admin already has access. """ |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session, project_id=None) |
| tests.create_tokens_acl(self.session, 'aaabbbcccddd', 'modify_project') |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| |
| project = pagure.lib.query._get_project(self.session, 'test') |
| pagure.lib.query.add_user_to_project( |
| self.session, project, |
| new_user='foo', |
| user='pingou', |
| access='commit' |
| ) |
| self.session.commit() |
| |
| output = self.app.patch( |
| '/api/0/test', headers=headers, |
| data={'main_admin': 'foo', 'retain_access': True}) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data['date_created'] = '1496338274' |
| data['date_modified'] = '1496338274' |
| expected_output = { |
| "access_groups": { |
| "admin": [], |
| "commit": [], |
| "ticket": [] |
| }, |
| "access_users": { |
| "admin": [ |
| "pingou" |
| ], |
| "commit": [], |
| "owner": [ |
| "foo" |
| ], |
| "ticket": [] |
| }, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate" |
| ], |
| "custom_keys": [], |
| "date_created": "1496338274", |
| "date_modified": "1496338274", |
| "description": "test project #1", |
| "fullname": "test", |
| "url_path": "test", |
| "id": 1, |
| "milestones": {}, |
| "name": "test", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": [], |
| "user": { |
| "default_email": "foo@bar.com", |
| "emails": [ |
| "foo@bar.com" |
| ], |
| "fullname": "foo bar", |
| "name": "foo" |
| } |
| } |
| self.assertEqual(data, expected_output) |
| |
| def test_api_modify_project_main_admin_json(self): |
| """ Test the api_modify_project method of the flask api when the |
| request is to change the main_admin of the project using JSON. """ |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session, project_id=None) |
| tests.create_tokens_acl(self.session, 'aaabbbcccddd', 'modify_project') |
| headers = {'Authorization': 'token aaabbbcccddd', |
| 'Content-Type': 'application/json'} |
| |
| output = self.app.patch( |
| '/api/0/test', headers=headers, |
| data=json.dumps({'main_admin': 'foo'})) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data['date_created'] = '1496338274' |
| data['date_modified'] = '1496338274' |
| expected_output = { |
| "access_groups": { |
| "admin": [], |
| "commit": [], |
| "ticket": [] |
| }, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": [ |
| "foo" |
| ], |
| "ticket": [] |
| }, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate" |
| ], |
| "custom_keys": [], |
| "date_created": "1496338274", |
| "date_modified": "1496338274", |
| "description": "test project #1", |
| "fullname": "test", |
| "url_path": "test", |
| "id": 1, |
| "milestones": {}, |
| "name": "test", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": [], |
| "user": { |
| "default_email": "foo@bar.com", |
| "emails": [ |
| "foo@bar.com" |
| ], |
| "fullname": "foo bar", |
| "name": "foo" |
| } |
| } |
| self.assertEqual(data, expected_output) |
| |
| @patch.dict('pagure.config.config', {'PAGURE_ADMIN_USERS': 'foo'}) |
| def test_api_modify_project_main_admin_as_site_admin(self): |
| """ Test the api_modify_project method of the flask api when the |
| request is to change the main_admin of the project and the user is a |
| Pagure site admin. """ |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session, user_id=2, project_id=None) |
| tests.create_tokens_acl(self.session, 'aaabbbcccddd', 'modify_project') |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| |
| |
| project = pagure.lib.query.get_authorized_project( |
| self.session, 'test') |
| date_before = project.date_modified |
| self.assertIsNotNone(date_before) |
| |
| output = self.app.patch( |
| '/api/0/test', headers=headers, |
| data={'main_admin': 'foo'}) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data['date_created'] = '1496338274' |
| data['date_modified'] = '1496338274' |
| expected_output = { |
| "access_groups": { |
| "admin": [], |
| "commit": [], |
| "ticket": [] |
| }, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": [ |
| "foo" |
| ], |
| "ticket": [] |
| }, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate" |
| ], |
| "custom_keys": [], |
| "date_created": "1496338274", |
| "date_modified": "1496338274", |
| "description": "test project #1", |
| "fullname": "test", |
| "url_path": "test", |
| "id": 1, |
| "milestones": {}, |
| "name": "test", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": [], |
| "user": { |
| "default_email": "foo@bar.com", |
| "emails": [ |
| "foo@bar.com" |
| ], |
| "fullname": "foo bar", |
| "name": "foo" |
| } |
| } |
| self.assertEqual(data, expected_output) |
| |
| |
| self.session = pagure.lib.query.create_session(self.dbpath) |
| project = pagure.lib.query.get_authorized_project( |
| self.session, 'test') |
| self.assertNotEqual(date_before, project.date_modified) |
| |
| def test_api_modify_project_main_admin_not_main_admin(self): |
| """ Test the api_modify_project method of the flask api when the |
| requester is not the main_admin of the project and requests to change |
| the main_admin. |
| """ |
| tests.create_projects(self.session) |
| project_user = pagure.lib.query.model.ProjectUser( |
| project_id=1, |
| user_id=2, |
| access='admin', |
| ) |
| self.session.add(project_user) |
| self.session.commit() |
| tests.create_tokens(self.session, project_id=None, user_id=2) |
| tests.create_tokens_acl(self.session, 'aaabbbcccddd', 'modify_project') |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| |
| output = self.app.patch( |
| '/api/0/test', headers=headers, |
| data={'main_admin': 'foo'}) |
| self.assertEqual(output.status_code, 401) |
| expected_error = { |
| 'error': ('Only the main admin can set the main admin of a ' |
| 'project'), |
| 'error_code': 'ENOTMAINADMIN' |
| } |
| self.assertEqual( |
| json.loads(output.get_data(as_text=True)), expected_error) |
| |
| def test_api_modify_project_not_admin(self): |
| """ Test the api_modify_project method of the flask api when the |
| requester is not an admin of the project. |
| """ |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session, project_id=None, user_id=2) |
| tests.create_tokens_acl(self.session, 'aaabbbcccddd', 'modify_project') |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| |
| output = self.app.patch( |
| '/api/0/test', headers=headers, |
| data={'main_admin': 'foo'}) |
| self.assertEqual(output.status_code, 401) |
| expected_error = { |
| 'error': 'You are not allowed to modify this project', |
| 'error_code': 'EMODIFYPROJECTNOTALLOWED' |
| } |
| self.assertEqual( |
| json.loads(output.get_data(as_text=True)), expected_error) |
| |
| def test_api_modify_project_invalid_request(self): |
| """ Test the api_modify_project method of the flask api when the |
| request data is invalid. |
| """ |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session, project_id=None) |
| tests.create_tokens_acl(self.session, 'aaabbbcccddd', 'modify_project') |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| |
| output = self.app.patch( |
| '/api/0/test', headers=headers, |
| data='invalid') |
| self.assertEqual(output.status_code, 400) |
| expected_error = { |
| 'error': 'Invalid or incomplete input submitted', |
| 'error_code': 'EINVALIDREQ' |
| } |
| self.assertEqual( |
| json.loads(output.get_data(as_text=True)), expected_error) |
| |
| def test_api_modify_project_invalid_keys(self): |
| """ Test the api_modify_project method of the flask api when the |
| request data contains an invalid key. |
| """ |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session, project_id=None) |
| tests.create_tokens_acl(self.session, 'aaabbbcccddd', 'modify_project') |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| |
| output = self.app.patch( |
| '/api/0/test', headers=headers, |
| data={'invalid': 'invalid'}) |
| self.assertEqual(output.status_code, 400) |
| expected_error = { |
| 'error': 'Invalid or incomplete input submitted', |
| 'error_code': 'EINVALIDREQ' |
| } |
| self.assertEqual( |
| json.loads(output.get_data(as_text=True)), expected_error) |
| |
| def test_api_modify_project_invalid_new_main_admin(self): |
| """ Test the api_modify_project method of the flask api when the |
| request is to change the main_admin of the project to a main_admin |
| that doesn't exist. |
| """ |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session, project_id=None) |
| tests.create_tokens_acl(self.session, 'aaabbbcccddd', 'modify_project') |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| |
| output = self.app.patch( |
| '/api/0/test', headers=headers, |
| data={'main_admin': 'tbrady'}) |
| self.assertEqual(output.status_code, 400) |
| expected_error = { |
| 'error': 'No such user found', |
| 'error_code': 'ENOUSER' |
| } |
| self.assertEqual( |
| json.loads(output.get_data(as_text=True)), expected_error) |
| |
| def test_api_project_watchers(self): |
| """ Test the api_project_watchers method of the flask api. """ |
| tests.create_projects(self.session) |
| |
| output = self.app.get('/api/0/test/watchers') |
| self.assertEqual(output.status_code, 200) |
| expected_data = { |
| "total_watchers": 1, |
| "watchers": { |
| "pingou": [ |
| "issues" |
| ] |
| } |
| } |
| self.assertDictEqual(json.loads(output.get_data(as_text=True)), expected_data) |
| |
| user = tests.FakeUser(username='pingou') |
| with tests.user_set(self.app.application, user): |
| |
| output = self.app.get('/api/0/random/watchers') |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| {'error_code': 'ENOPROJECT', 'error': 'Project not found'} |
| ) |
| |
| |
| output = self.app.get('/api/0/test/watchers') |
| self.assertEqual(output.status_code, 200) |
| expected_data = { |
| "total_watchers": 1, |
| "watchers": { |
| "pingou": [ |
| "issues" |
| ] |
| } |
| } |
| self.assertDictEqual(json.loads(output.get_data(as_text=True)), expected_data) |
| |
| project = pagure.lib.query.get_authorized_project(self.session, 'test') |
| |
| |
| pagure.lib.query.update_watch_status( |
| self.session, project, 'pingou', '3') |
| self.session.commit() |
| output = self.app.get('/api/0/test/watchers') |
| self.assertEqual(output.status_code, 200) |
| expected_data = { |
| "total_watchers": 1, |
| "watchers": { |
| "pingou": [ |
| "issues", |
| "commits" |
| ] |
| } |
| } |
| self.assertDictEqual(json.loads(output.get_data(as_text=True)), expected_data) |
| |
| |
| pagure.lib.query.update_watch_status( |
| self.session, project, 'pingou', '1') |
| self.session.commit() |
| output = self.app.get('/api/0/test/watchers') |
| self.assertEqual(output.status_code, 200) |
| expected_data = { |
| "total_watchers": 1, |
| "watchers": { |
| "pingou": [ |
| "issues" |
| ] |
| } |
| } |
| self.assertDictEqual(json.loads(output.get_data(as_text=True)), expected_data) |
| |
| |
| pagure.lib.query.update_watch_status( |
| self.session, project, 'pingou', '2') |
| self.session.commit() |
| output = self.app.get('/api/0/test/watchers') |
| self.assertEqual(output.status_code, 200) |
| expected_data = { |
| "total_watchers": 1, |
| "watchers": { |
| "pingou": [ |
| "commits" |
| ] |
| } |
| } |
| self.assertDictEqual(json.loads(output.get_data(as_text=True)), expected_data) |
| |
| |
| |
| project_user = pagure.lib.model.ProjectUser( |
| project_id=project.id, |
| user_id=2, |
| access='commit', |
| ) |
| pagure.lib.query.update_watch_status( |
| self.session, project, 'pingou', '2') |
| self.session.add(project_user) |
| self.session.commit() |
| |
| output = self.app.get('/api/0/test/watchers') |
| self.assertEqual(output.status_code, 200) |
| expected_data = { |
| "total_watchers": 2, |
| "watchers": { |
| "foo": ["issues"], |
| "pingou": ["commits"] |
| } |
| } |
| self.assertDictEqual(json.loads(output.get_data(as_text=True)), expected_data) |
| |
| |
| pagure.lib.query.update_watch_status( |
| self.session, project, 'pingou', '-1') |
| self.session.commit() |
| |
| output = self.app.get('/api/0/test/watchers') |
| self.assertEqual(output.status_code, 200) |
| expected_data = { |
| "total_watchers": 2, |
| "watchers": { |
| "foo": ["issues"], |
| "pingou": ["issues"] |
| } |
| } |
| self.assertDictEqual(json.loads(output.get_data(as_text=True)), expected_data) |
| |
| |
| |
| pagure.lib.query.update_watch_status( |
| self.session, project, 'pingou', '-1') |
| project_membership = self.session.query( |
| pagure.lib.model.ProjectUser).filter_by( |
| user_id=2, project_id=project.id).one() |
| self.session.delete(project_membership) |
| self.session.commit() |
| |
| msg = pagure.lib.query.add_group( |
| self.session, |
| group_name='some_group', |
| display_name='Some Group', |
| description=None, |
| group_type='bar', |
| user='pingou', |
| is_admin=False, |
| blacklist=[], |
| ) |
| self.session.commit() |
| |
| project = pagure.lib.query.get_authorized_project(self.session, 'test') |
| group = pagure.lib.query.search_groups( |
| self.session, group_name='some_group') |
| pagure.lib.query.add_user_to_group( |
| self.session, 'foo', group, 'pingou', False) |
| |
| pagure.lib.query.add_group_to_project( |
| self.session, |
| project, |
| new_group='some_group', |
| user='pingou', |
| access='commit', |
| create=False, |
| is_admin=True |
| ) |
| self.session.commit() |
| |
| output = self.app.get('/api/0/test/watchers') |
| self.assertEqual(output.status_code, 200) |
| expected_data = { |
| "total_watchers": 2, |
| "watchers": { |
| "@some_group": ["issues"], |
| "pingou": ["issues"] |
| } |
| } |
| self.assertDictEqual(json.loads(output.get_data(as_text=True)), expected_data) |
| |
| |
| |
| pagure.lib.query.update_watch_status( |
| self.session, project, 'pingou', '-1') |
| pagure.lib.query.update_watch_status( |
| self.session, project, 'foo', '2') |
| self.session.commit() |
| |
| output = self.app.get('/api/0/test/watchers') |
| self.assertEqual(output.status_code, 200) |
| expected_data = { |
| "total_watchers": 3, |
| "watchers": { |
| "@some_group": ["issues"], |
| "foo": ["commits"], |
| "pingou": ["issues"] |
| } |
| } |
| self.assertDictEqual(json.loads(output.get_data(as_text=True)), expected_data) |
| |
| def test_api_new_project(self): |
| """ Test the api_new_project method of the flask api. """ |
| |
| tests.create_projects(self.session) |
| tests.create_projects_git(os.path.join(self.path, 'tickets')) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| headers = {'Authorization': 'token foo_token'} |
| |
| |
| output = self.app.post('/api/0/new', headers=headers) |
| self.assertEqual(output.status_code, 401) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(sorted(data.keys()), [ |
| 'error', 'error_code', "errors"]) |
| self.assertEqual( |
| pagure.api.APIERROR.EINVALIDTOK.value, data['error']) |
| self.assertEqual( |
| pagure.api.APIERROR.EINVALIDTOK.name, data['error_code']) |
| self.assertEqual( |
| data['errors'], "Missing ACLs: create_project") |
| |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| |
| |
| output = self.app.post('/api/0/new', headers=headers) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Invalid or incomplete input submitted", |
| "error_code": "EINVALIDREQ", |
| "errors": { |
| "name": ["This field is required."], |
| "description": ["This field is required."] |
| } |
| } |
| ) |
| |
| data = { |
| 'name': 'test', |
| } |
| |
| |
| output = self.app.post( |
| '/api/0/new', data=data, headers=headers) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Invalid or incomplete input submitted", |
| "error_code": "EINVALIDREQ", |
| "errors": {"description": ["This field is required."]} |
| } |
| ) |
| |
| data = { |
| 'name': 'test', |
| 'description': 'Just a small test project', |
| } |
| |
| |
| output = self.app.post( |
| '/api/0/new/', data=data, headers=headers) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "It is not possible to create the repo \"test\"", |
| "error_code": "ENOCODE" |
| } |
| ) |
| |
| data = { |
| 'name': 'api1', |
| 'description': 'Mighty mighty description', |
| 'avatar_email': 123 |
| } |
| |
| |
| output = self.app.post( |
| '/api/0/new/', data=data, headers=headers) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(data, |
| {"error": "Invalid or incomplete input submitted", |
| "error_code": "EINVALIDREQ", |
| "errors": |
| {"avatar_email": ['avatar_email must be an email']} |
| } |
| ) |
| |
| data = { |
| 'name': 'api1', |
| 'description': 'Mighty mighty description', |
| 'avatar_email': [1,2,3] |
| } |
| |
| |
| output = self.app.post( |
| '/api/0/new/', data=data, headers=headers) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(data, |
| {"error": "Invalid or incomplete input submitted", |
| "error_code": "EINVALIDREQ", |
| "errors": |
| {"avatar_email": ['avatar_email must be an email']} |
| } |
| ) |
| |
| data = { |
| 'name': 'api1', |
| 'description': 'Mighty mighty description', |
| 'avatar_email': True |
| } |
| |
| |
| output = self.app.post( |
| '/api/0/new/', data=data, headers=headers) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual(data, |
| {"error": "Invalid or incomplete input submitted", |
| "error_code": "EINVALIDREQ", |
| "errors": |
| {"avatar_email": ['avatar_email must be an email']} |
| } |
| ) |
| |
| data = { |
| 'name': 'api1', |
| 'description': 'Mighty mighty description', |
| 'avatar_email': 'mighty@email.com' |
| } |
| |
| |
| output = self.app.post( |
| '/api/0/new/', data=data, headers=headers) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| {'message': 'Project "api1" created'}) |
| |
| data = { |
| 'name': 'test_42', |
| 'description': 'Just another small test project', |
| } |
| |
| |
| output = self.app.post( |
| '/api/0/new/', data=data, headers=headers) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| {'message': 'Project "test_42" created'} |
| ) |
| |
| @patch.dict('pagure.config.config', {'PAGURE_ADMIN_USERS': ['pingou'], |
| 'ALLOW_ADMIN_IGNORE_EXISTING_REPOS': True}) |
| def test_adopt_repos(self): |
| """ Test the new_project endpoint with existing git repo. """ |
| |
| projects = pagure.lib.query.search_projects(self.session) |
| self.assertEqual(len(projects), 0) |
| |
| tests.create_projects_git(os.path.join(self.path, 'repos'), bare=True) |
| tests.add_content_git_repo(os.path.join(self.path, 'repos', 'test.git')) |
| item = pagure.lib.model.Token( |
| id='aaabbbcccddd', |
| user_id=1, |
| project_id=None, |
| expiration=datetime.datetime.utcnow() + datetime.timedelta(days=10) |
| ) |
| self.session.add(item) |
| self.session.commit() |
| tests.create_tokens_acl(self.session) |
| |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| |
| user = tests.FakeUser(username='pingou') |
| with tests.user_set(self.app.application, user): |
| input_data = { |
| 'name': 'test', |
| 'description': 'Project #1', |
| } |
| |
| |
| output = self.app.post( |
| '/api/0/new/', data=input_data, headers=headers) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| 'error': 'The main repo test.git already exists', |
| 'error_code': 'ENOCODE' |
| } |
| ) |
| |
| input_data['ignore_existing_repos'] = 'y' |
| |
| output = self.app.post( |
| '/api/0/new/', data=input_data, headers=headers) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| {'message': 'Project "test" created'} |
| ) |
| |
| @patch.dict('pagure.config.config', {'PRIVATE_PROJECTS': True}) |
| def test_api_new_project_private(self): |
| """ Test the api_new_project method of the flask api to create |
| a private project. """ |
| |
| tests.create_projects(self.session) |
| tests.create_projects_git(os.path.join(self.path, 'tickets')) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| |
| data = { |
| 'name': 'test', |
| 'description': 'Just a small test project', |
| 'private': True, |
| } |
| |
| |
| output = self.app.post( |
| '/api/0/new/', data=data, headers=headers) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| {'message': 'Project "pingou/test" created'} |
| ) |
| |
| def test_api_new_project_user_token(self): |
| """ Test the api_new_project method of the flask api. """ |
| tests.create_projects(self.session) |
| tests.create_projects_git(os.path.join(self.path, 'tickets')) |
| tests.create_tokens(self.session, project_id=None) |
| tests.create_tokens_acl(self.session) |
| |
| headers = {'Authorization': 'token foo_token'} |
| |
| |
| output = self.app.post('/api/0/new', headers=headers) |
| self.assertEqual(output.status_code, 401) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(sorted(data.keys()), [ |
| 'error', 'error_code', "errors"]) |
| self.assertEqual( |
| pagure.api.APIERROR.EINVALIDTOK.value, data['error']) |
| self.assertEqual( |
| pagure.api.APIERROR.EINVALIDTOK.name, data['error_code']) |
| self.assertEqual( |
| data['errors'], "Missing ACLs: create_project") |
| |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| |
| |
| output = self.app.post('/api/0/new', headers=headers) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Invalid or incomplete input submitted", |
| "error_code": "EINVALIDREQ", |
| "errors": { |
| "name": ["This field is required."], |
| "description": ["This field is required."] |
| } |
| } |
| ) |
| |
| data = { |
| 'name': 'test', |
| } |
| |
| |
| output = self.app.post( |
| '/api/0/new', data=data, headers=headers) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Invalid or incomplete input submitted", |
| "error_code": "EINVALIDREQ", |
| "errors": {"description": ["This field is required."]} |
| } |
| ) |
| |
| data = { |
| 'name': 'test', |
| 'description': 'Just a small test project', |
| } |
| |
| |
| output = self.app.post( |
| '/api/0/new/', data=data, headers=headers) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "It is not possible to create the repo \"test\"", |
| "error_code": "ENOCODE" |
| } |
| ) |
| |
| data = { |
| 'name': 'test_42', |
| 'description': 'Just another small test project', |
| } |
| |
| |
| output = self.app.post( |
| '/api/0/new/', data=data, headers=headers) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| {'message': 'Project "test_42" created'} |
| ) |
| |
| |
| pagure.config.config['ALLOWED_PREFIX'] = ['rpms'] |
| data = { |
| 'name': 'test_42', |
| 'namespace': 'pingou', |
| 'description': 'Just another small test project', |
| } |
| |
| |
| output = self.app.post( |
| '/api/0/new/', data=data, headers=headers) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Invalid or incomplete input submitted", |
| "error_code": "EINVALIDREQ", |
| "errors": { |
| "namespace": [ |
| "Not a valid choice" |
| ] |
| } |
| } |
| ) |
| |
| data = { |
| 'name': 'test_42', |
| 'namespace': 'rpms', |
| 'description': 'Just another small test project', |
| } |
| |
| |
| output = self.app.post( |
| '/api/0/new/', data=data, headers=headers) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| {'message': 'Project "rpms/test_42" created'} |
| ) |
| |
| @patch.dict('pagure.config.config', {'USER_NAMESPACE': True}) |
| def test_api_new_project_user_ns(self): |
| """ Test the api_new_project method of the flask api. """ |
| tests.create_projects(self.session) |
| tests.create_projects_git(os.path.join(self.path, 'tickets')) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| |
| |
| data = { |
| 'name': 'testproject', |
| 'description': 'Just another small test project', |
| } |
| |
| |
| output = self.app.post( |
| '/api/0/new/', data=data, headers=headers) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| {'message': 'Project "pingou/testproject" created'} |
| ) |
| |
| |
| data = { |
| 'name': 'testproject2', |
| 'namespace': 'testns', |
| 'description': 'Just another small test project', |
| } |
| |
| |
| with patch.dict('pagure.config.config', {'ALLOWED_PREFIX': ['testns']}): |
| output = self.app.post( |
| '/api/0/new/', data=data, headers=headers) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| {'message': 'Project "testns/testproject2" created'} |
| ) |
| |
| def test_api_fork_project(self): |
| """ Test the api_fork_project method of the flask api. """ |
| tests.create_projects(self.session) |
| for folder in ['docs', 'tickets', 'requests', 'repos']: |
| tests.create_projects_git( |
| os.path.join(self.path, folder), bare=True) |
| tests.create_tokens(self.session) |
| tests.create_tokens_acl(self.session) |
| |
| headers = {'Authorization': 'token foo_token'} |
| |
| |
| output = self.app.post('/api/0/fork', headers=headers) |
| self.assertEqual(output.status_code, 401) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(sorted(data.keys()), [ |
| 'error', 'error_code', "errors"]) |
| self.assertEqual( |
| pagure.api.APIERROR.EINVALIDTOK.value, data['error']) |
| self.assertEqual( |
| pagure.api.APIERROR.EINVALIDTOK.name, data['error_code']) |
| self.assertEqual( |
| data['errors'], "Missing ACLs: fork_project") |
| |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| |
| |
| output = self.app.post('/api/0/fork', headers=headers) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Invalid or incomplete input submitted", |
| "error_code": "EINVALIDREQ", |
| "errors": {"repo": ["This field is required."]} |
| } |
| ) |
| |
| data = { |
| 'name': 'test', |
| } |
| |
| |
| output = self.app.post( |
| '/api/0/fork', data=data, headers=headers) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Invalid or incomplete input submitted", |
| "error_code": "EINVALIDREQ", |
| "errors": {"repo": ["This field is required."]} |
| } |
| ) |
| |
| data = { |
| 'repo': 'test', |
| } |
| |
| |
| output = self.app.post( |
| '/api/0/fork/', data=data, headers=headers) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "message": "Repo \"test\" cloned to \"pingou/test\"" |
| } |
| ) |
| |
| data = { |
| 'repo': 'test', |
| } |
| |
| |
| output = self.app.post( |
| '/api/0/fork/', data=data, headers=headers) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Repo \"forks/pingou/test\" already exists", |
| "error_code": "ENOCODE" |
| } |
| ) |
| |
| data = { |
| 'repo': 'test', |
| 'username': 'pingou', |
| } |
| |
| |
| output = self.app.post( |
| '/api/0/fork/', data=data, headers=headers) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Repo \"forks/pingou/test\" already exists", |
| "error_code": "ENOCODE" |
| } |
| ) |
| |
| data = { |
| 'repo': 'test', |
| 'namespace': 'pingou', |
| } |
| |
| |
| output = self.app.post( |
| '/api/0/fork/', data=data, headers=headers) |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Project not found", |
| "error_code": "ENOPROJECT" |
| } |
| ) |
| |
| def test_api_fork_project_user_token(self): |
| """ Test the api_fork_project method of the flask api. """ |
| tests.create_projects(self.session) |
| for folder in ['docs', 'tickets', 'requests', 'repos']: |
| tests.create_projects_git( |
| os.path.join(self.path, folder), bare=True) |
| tests.create_tokens(self.session, project_id=None) |
| tests.create_tokens_acl(self.session) |
| |
| headers = {'Authorization': 'token foo_token'} |
| |
| |
| output = self.app.post('/api/0/fork', headers=headers) |
| self.assertEqual(output.status_code, 401) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(sorted(data.keys()), [ |
| 'error', 'error_code', "errors"]) |
| self.assertEqual( |
| pagure.api.APIERROR.EINVALIDTOK.value, data['error']) |
| self.assertEqual( |
| pagure.api.APIERROR.EINVALIDTOK.name, data['error_code']) |
| self.assertEqual( |
| data['errors'], "Missing ACLs: fork_project") |
| |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| |
| |
| output = self.app.post('/api/0/fork', headers=headers) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Invalid or incomplete input submitted", |
| "error_code": "EINVALIDREQ", |
| "errors": {"repo": ["This field is required."]} |
| } |
| ) |
| |
| data = { |
| 'name': 'test', |
| } |
| |
| |
| output = self.app.post( |
| '/api/0/fork', data=data, headers=headers) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Invalid or incomplete input submitted", |
| "error_code": "EINVALIDREQ", |
| "errors": {"repo": ["This field is required."]} |
| } |
| ) |
| |
| data = { |
| 'repo': 'test', |
| } |
| |
| |
| output = self.app.post( |
| '/api/0/fork/', data=data, headers=headers) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "message": "Repo \"test\" cloned to \"pingou/test\"" |
| } |
| ) |
| |
| data = { |
| 'repo': 'test', |
| } |
| |
| |
| output = self.app.post( |
| '/api/0/fork/', data=data, headers=headers) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Repo \"forks/pingou/test\" already exists", |
| "error_code": "ENOCODE" |
| } |
| ) |
| |
| data = { |
| 'repo': 'test', |
| 'username': 'pingou', |
| } |
| |
| |
| output = self.app.post( |
| '/api/0/fork/', data=data, headers=headers) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Repo \"forks/pingou/test\" already exists", |
| "error_code": "ENOCODE" |
| } |
| ) |
| |
| data = { |
| 'repo': 'test', |
| 'namespace': 'pingou', |
| } |
| |
| |
| output = self.app.post( |
| '/api/0/fork/', data=data, headers=headers) |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertDictEqual( |
| data, |
| { |
| "error": "Project not found", |
| "error_code": "ENOPROJECT" |
| } |
| ) |
| |
| def test_api_generate_acls(self): |
| """ Test the api_generate_acls method of the flask api """ |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session, project_id=None) |
| tests.create_tokens_acl( |
| self.session, 'aaabbbcccddd', 'generate_acls_project') |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| |
| user = pagure.lib.query.get_user(self.session, 'pingou') |
| output = self.app.post( |
| '/api/0/test/git/generateacls', headers=headers, |
| data={'wait': False}) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| expected_output = { |
| 'message': 'Project ACL generation queued', |
| 'taskid': 'abc-1234' |
| } |
| self.assertEqual(data, expected_output) |
| self.mock_gen_acls.assert_called_once_with( |
| name='test', namespace=None, user=None, group=None) |
| |
| def test_api_generate_acls_json(self): |
| """ Test the api_generate_acls method of the flask api using JSON """ |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session, project_id=None) |
| tests.create_tokens_acl( |
| self.session, 'aaabbbcccddd', 'generate_acls_project') |
| headers = {'Authorization': 'token aaabbbcccddd', |
| 'Content-Type': 'application/json'} |
| |
| user = pagure.lib.query.get_user(self.session, 'pingou') |
| |
| output = self.app.post( |
| '/api/0/test/git/generateacls', headers=headers, |
| data=json.dumps({'wait': False})) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| expected_output = { |
| 'message': 'Project ACL generation queued', |
| 'taskid': 'abc-1234' |
| } |
| self.assertEqual(data, expected_output) |
| self.mock_gen_acls.assert_called_once_with( |
| name='test', namespace=None, user=None, group=None) |
| |
| def test_api_generate_acls_wait_true(self): |
| """ Test the api_generate_acls method of the flask api when wait is |
| set to True """ |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session, project_id=None) |
| tests.create_tokens_acl( |
| self.session, 'aaabbbcccddd', 'generate_acls_project') |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| |
| task_result = Mock() |
| task_result.id = 'abc-1234' |
| self.mock_gen_acls.return_value = task_result |
| |
| user = pagure.lib.query.get_user(self.session, 'pingou') |
| output = self.app.post( |
| '/api/0/test/git/generateacls', headers=headers, |
| data={'wait': True}) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| expected_output = { |
| 'message': 'Project ACLs generated', |
| } |
| self.assertEqual(data, expected_output) |
| self.mock_gen_acls.assert_called_once_with( |
| name='test', namespace=None, user=None, group=None) |
| self.assertTrue(task_result.get.called) |
| |
| def test_api_generate_acls_no_project(self): |
| """ Test the api_generate_acls method of the flask api when the project |
| doesn't exist """ |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session, project_id=None) |
| tests.create_tokens_acl( |
| self.session, 'aaabbbcccddd', 'generate_acls_project') |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| |
| user = pagure.lib.query.get_user(self.session, 'pingou') |
| output = self.app.post( |
| '/api/0/test12345123/git/generateacls', headers=headers, |
| data={'wait': False}) |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| expected_output = { |
| 'error_code': 'ENOPROJECT', |
| 'error': 'Project not found' |
| } |
| self.assertEqual(data, expected_output) |
| |
| def test_api_new_git_branch(self): |
| """ Test the api_new_branch method of the flask api """ |
| tests.create_projects(self.session) |
| repo_path = os.path.join(self.path, 'repos') |
| tests.create_projects_git(repo_path, bare=True) |
| tests.add_content_git_repo(os.path.join(repo_path, 'test.git')) |
| tests.create_tokens(self.session, project_id=None) |
| tests.create_tokens_acl( |
| self.session, 'aaabbbcccddd', 'create_branch') |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| args = {'branch': 'test123'} |
| output = self.app.post('/api/0/test/git/branch', headers=headers, |
| data=args) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| expected_output = { |
| 'message': 'Project branch was created', |
| } |
| self.assertEqual(data, expected_output) |
| git_path = os.path.join(self.path, 'repos', 'test.git') |
| repo_obj = pygit2.Repository(git_path) |
| self.assertIn('test123', repo_obj.listall_branches()) |
| |
| def test_api_new_git_branch_json(self): |
| """ Test the api_new_branch method of the flask api """ |
| tests.create_projects(self.session) |
| repo_path = os.path.join(self.path, 'repos') |
| tests.create_projects_git(repo_path, bare=True) |
| tests.add_content_git_repo(os.path.join(repo_path, 'test.git')) |
| tests.create_tokens(self.session, project_id=None) |
| tests.create_tokens_acl( |
| self.session, 'aaabbbcccddd', 'create_branch') |
| headers = {'Authorization': 'token aaabbbcccddd', |
| 'Content-Type': 'application/json'} |
| args = {'branch': 'test123'} |
| output = self.app.post('/api/0/test/git/branch', headers=headers, |
| data=json.dumps(args)) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| expected_output = { |
| 'message': 'Project branch was created', |
| } |
| self.assertEqual(data, expected_output) |
| git_path = os.path.join(self.path, 'repos', 'test.git') |
| repo_obj = pygit2.Repository(git_path) |
| self.assertIn('test123', repo_obj.listall_branches()) |
| |
| def test_api_new_git_branch_from_branch(self): |
| """ Test the api_new_branch method of the flask api """ |
| tests.create_projects(self.session) |
| repo_path = os.path.join(self.path, 'repos') |
| tests.create_projects_git(repo_path, bare=True) |
| tests.add_content_git_repo(os.path.join(repo_path, 'test.git')) |
| tests.create_tokens(self.session, project_id=None) |
| tests.create_tokens_acl( |
| self.session, 'aaabbbcccddd', 'create_branch') |
| git_path = os.path.join(self.path, 'repos', 'test.git') |
| repo_obj = pygit2.Repository(git_path) |
| parent = pagure.lib.git.get_branch_ref(repo_obj, 'master').peel() |
| repo_obj.create_branch('dev123', parent) |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| args = {'branch': 'test123', 'from_branch': 'dev123'} |
| output = self.app.post('/api/0/test/git/branch', headers=headers, |
| data=args) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| expected_output = { |
| 'message': 'Project branch was created', |
| } |
| self.assertEqual(data, expected_output) |
| self.assertIn('test123', repo_obj.listall_branches()) |
| |
| def test_api_new_git_branch_already_exists(self): |
| """ Test the api_new_branch method of the flask api when branch already |
| exists """ |
| tests.create_projects(self.session) |
| repo_path = os.path.join(self.path, 'repos') |
| tests.create_projects_git(repo_path, bare=True) |
| tests.add_content_git_repo(os.path.join(repo_path, 'test.git')) |
| tests.create_tokens(self.session, project_id=None) |
| tests.create_tokens_acl( |
| self.session, 'aaabbbcccddd', 'create_branch') |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| args = {'branch': 'master'} |
| output = self.app.post('/api/0/test/git/branch', headers=headers, |
| data=args) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| expected_output = { |
| 'error': 'The branch "master" already exists', |
| 'error_code': 'ENOCODE' |
| } |
| self.assertEqual(data, expected_output) |
| |
| def test_api_new_git_branch_from_commit(self): |
| """ Test the api_new_branch method of the flask api """ |
| tests.create_projects(self.session) |
| repos_path = os.path.join(self.path, 'repos') |
| tests.create_projects_git(repos_path, bare=True) |
| git_path = os.path.join(repos_path, 'test.git') |
| tests.add_content_git_repo(git_path) |
| tests.create_tokens(self.session, project_id=None) |
| tests.create_tokens_acl( |
| self.session, 'aaabbbcccddd', 'create_branch') |
| repo_obj = pygit2.Repository(git_path) |
| from_commit = repo_obj.revparse_single('HEAD').oid.hex |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| args = {'branch': 'test123', 'from_commit': from_commit} |
| output = self.app.post('/api/0/test/git/branch', headers=headers, |
| data=args) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| expected_output = { |
| 'message': 'Project branch was created', |
| } |
| self.assertEqual(data, expected_output) |
| self.assertIn('test123', repo_obj.listall_branches()) |
| |
| |
| class PagureFlaskApiProjectFlagtests(tests.Modeltests): |
| """ Tests for the flask API of pagure for flagging commit in project |
| """ |
| |
| def setUp(self): |
| """ Set up the environnment, ran before every tests. """ |
| super(PagureFlaskApiProjectFlagtests, self).setUp() |
| |
| tests.create_projects(self.session) |
| repo_path = os.path.join(self.path, 'repos') |
| self.git_path = os.path.join(repo_path, 'test.git') |
| tests.create_projects_git(repo_path, bare=True) |
| tests.add_content_git_repo(self.git_path) |
| tests.create_tokens(self.session, project_id=None) |
| tests.create_tokens_acl( |
| self.session, 'aaabbbcccddd', 'commit_flag') |
| |
| def test_flag_commit_missing_status(self): |
| """ Test flagging a commit with missing precentage. """ |
| repo_obj = pygit2.Repository(self.git_path) |
| commit = repo_obj.revparse_single('HEAD') |
| |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| data = { |
| 'username': 'Jenkins', |
| 'comment': 'Tests passed', |
| 'url': 'http://jenkins.cloud.fedoraproject.org/', |
| 'uid': 'jenkins_build_pagure_100+seed', |
| } |
| output = self.app.post( |
| '/api/0/test/c/%s/flag' % commit.oid.hex, |
| headers=headers, data=data) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| expected_output = { |
| "error": "Invalid or incomplete input submitted", |
| "error_code": "EINVALIDREQ", |
| "errors": { |
| "status": [ |
| "Not a valid choice" |
| ] |
| } |
| } |
| self.assertEqual(data, expected_output) |
| |
| def test_flag_commit_missing_username(self): |
| """ Test flagging a commit with missing username. """ |
| repo_obj = pygit2.Repository(self.git_path) |
| commit = repo_obj.revparse_single('HEAD') |
| |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| data = { |
| 'percent': 100, |
| 'comment': 'Tests passed', |
| 'url': 'http://jenkins.cloud.fedoraproject.org/', |
| 'uid': 'jenkins_build_pagure_100+seed', |
| 'status': 'success', |
| } |
| output = self.app.post( |
| '/api/0/test/c/%s/flag' % commit.oid.hex, |
| headers=headers, data=data) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| expected_output = { |
| "error": "Invalid or incomplete input submitted", |
| "error_code": "EINVALIDREQ", |
| "errors": { |
| "username": [ |
| "This field is required." |
| ] |
| } |
| } |
| self.assertEqual(data, expected_output) |
| |
| def test_flag_commit_missing_comment(self): |
| """ Test flagging a commit with missing comment. """ |
| repo_obj = pygit2.Repository(self.git_path) |
| commit = repo_obj.revparse_single('HEAD') |
| |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| data = { |
| 'username': 'Jenkins', |
| 'percent': 100, |
| 'url': 'http://jenkins.cloud.fedoraproject.org/', |
| 'uid': 'jenkins_build_pagure_100+seed', |
| 'status': 'success', |
| } |
| output = self.app.post( |
| '/api/0/test/c/%s/flag' % commit.oid.hex, |
| headers=headers, data=data) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| expected_output = { |
| "error": "Invalid or incomplete input submitted", |
| "error_code": "EINVALIDREQ", |
| "errors": { |
| "comment": [ |
| "This field is required." |
| ] |
| } |
| } |
| self.assertEqual(data, expected_output) |
| |
| def test_flag_commit_missing_url(self): |
| """ Test flagging a commit with missing url. """ |
| repo_obj = pygit2.Repository(self.git_path) |
| commit = repo_obj.revparse_single('HEAD') |
| |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| data = { |
| 'username': 'Jenkins', |
| 'percent': 100, |
| 'comment': 'Tests passed', |
| 'uid': 'jenkins_build_pagure_100+seed', |
| 'status': 'success', |
| } |
| output = self.app.post( |
| '/api/0/test/c/%s/flag' % commit.oid.hex, |
| headers=headers, data=data) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| expected_output = { |
| "error": "Invalid or incomplete input submitted", |
| "error_code": "EINVALIDREQ", |
| "errors": { |
| "url": [ |
| "This field is required." |
| ] |
| } |
| } |
| self.assertEqual(data, expected_output) |
| |
| def test_flag_commit_invalid_token(self): |
| """ Test flagging a commit with missing info. """ |
| repo_obj = pygit2.Repository(self.git_path) |
| commit = repo_obj.revparse_single('HEAD') |
| |
| headers = {'Authorization': 'token 123'} |
| data = { |
| 'username': 'Jenkins', |
| 'percent': 100, |
| 'comment': 'Tests passed', |
| 'url': 'http://jenkins.cloud.fedoraproject.org/', |
| 'uid': 'jenkins_build_pagure_100+seed', |
| } |
| output = self.app.post( |
| '/api/0/test/c/%s/flag' % commit.oid.hex, |
| headers=headers, data=data) |
| self.assertEqual(output.status_code, 401) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(sorted(data.keys()), [ |
| 'error', 'error_code', "errors"]) |
| self.assertEqual( |
| pagure.api.APIERROR.EINVALIDTOK.value, data['error']) |
| self.assertEqual( |
| pagure.api.APIERROR.EINVALIDTOK.name, data['error_code']) |
| self.assertEqual( |
| data['errors'], "Invalid token") |
| |
| def test_flag_commit_invalid_status(self): |
| """ Test flagging a commit with an invalid status. """ |
| repo_obj = pygit2.Repository(self.git_path) |
| commit = repo_obj.revparse_single('HEAD') |
| |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| data = { |
| 'username': 'Jenkins', |
| 'percent': 100, |
| 'comment': 'Tests passed', |
| 'url': 'http://jenkins.cloud.fedoraproject.org/', |
| 'status': 'foobar', |
| } |
| output = self.app.post( |
| '/api/0/test/c/%s/flag' % commit.oid.hex, |
| headers=headers, data=data) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| data, |
| { |
| 'errors': {'status': ['Not a valid choice']}, |
| 'error_code': 'EINVALIDREQ', |
| 'error': 'Invalid or incomplete input submitted' |
| } |
| ) |
| |
| def test_flag_commit_with_uid(self): |
| """ Test flagging a commit with provided uid. """ |
| repo_obj = pygit2.Repository(self.git_path) |
| commit = repo_obj.revparse_single('HEAD') |
| |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| data = { |
| 'username': 'Jenkins', |
| 'percent': 100, |
| 'comment': 'Tests passed', |
| 'url': 'http://jenkins.cloud.fedoraproject.org/', |
| 'uid': 'jenkins_build_pagure_100+seed', |
| 'status': 'success', |
| } |
| output = self.app.post( |
| '/api/0/test/c/%s/flag' % commit.oid.hex, |
| headers=headers, data=data) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data['flag']['date_created'] = '1510742565' |
| data['flag']['date_updated'] = '1510742565' |
| data['flag']['commit_hash'] = '62b49f00d489452994de5010565fab81' |
| expected_output = { |
| 'flag': { |
| 'comment': 'Tests passed', |
| 'commit_hash': '62b49f00d489452994de5010565fab81', |
| 'date_created': '1510742565', |
| 'date_updated': '1510742565', |
| 'percent': 100, |
| 'status': 'success', |
| 'url': 'http://jenkins.cloud.fedoraproject.org/', |
| 'user': { |
| 'default_email': 'bar@pingou.com', |
| 'emails': ['bar@pingou.com', 'foo@pingou.com'], |
| 'fullname': 'PY C', |
| 'name': 'pingou'}, |
| 'username': 'Jenkins' |
| }, |
| 'message': 'Flag added', |
| 'uid': 'jenkins_build_pagure_100+seed' |
| } |
| |
| self.assertEqual(data, expected_output) |
| |
| @patch('pagure.lib.notify.send_email') |
| def test_flag_commit_without_uid(self, mock_email): |
| """ Test flagging a commit with missing info. |
| |
| Also ensure notifications aren't sent when they are not asked for. |
| """ |
| repo_obj = pygit2.Repository(self.git_path) |
| commit = repo_obj.revparse_single('HEAD') |
| |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| data = { |
| 'username': 'Jenkins', |
| 'percent': 100, |
| 'comment': 'Tests passed', |
| 'url': 'http://jenkins.cloud.fedoraproject.org/', |
| 'status': 'success', |
| } |
| output = self.app.post( |
| '/api/0/test/c/%s/flag' % commit.oid.hex, |
| headers=headers, data=data) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertNotEqual( |
| data['uid'], |
| 'jenkins_build_pagure_100+seed' |
| ) |
| data['flag']['date_created'] = '1510742565' |
| data['flag']['date_updated'] = '1510742565' |
| data['uid'] = 'b1de8f80defd4a81afe2e09f39678087' |
| expected_output = { |
| 'flag': { |
| 'comment': 'Tests passed', |
| 'commit_hash': commit.oid.hex, |
| 'date_created': '1510742565', |
| 'date_updated': '1510742565', |
| 'percent': 100, |
| 'status': 'success', |
| 'url': 'http://jenkins.cloud.fedoraproject.org/', |
| 'user': { |
| 'default_email': 'bar@pingou.com', |
| 'emails': ['bar@pingou.com', 'foo@pingou.com'], |
| 'fullname': 'PY C', |
| 'name': 'pingou'}, |
| 'username': 'Jenkins' |
| }, |
| 'message': 'Flag added', |
| 'uid': 'b1de8f80defd4a81afe2e09f39678087' |
| } |
| self.assertEqual(data, expected_output) |
| |
| mock_email.assert_not_called() |
| |
| @patch('pagure.lib.notify.send_email') |
| def test_flag_commit_with_notification(self, mock_email): |
| """ Test flagging a commit with notification enabled. """ |
| |
| |
| repo = pagure.lib.query.get_authorized_project(self.session, 'test') |
| settings = repo.settings |
| settings['notify_on_commit_flag'] = True |
| repo.settings = settings |
| self.session.add(repo) |
| self.session.commit() |
| |
| repo_obj = pygit2.Repository(self.git_path) |
| commit = repo_obj.revparse_single('HEAD') |
| |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| data = { |
| 'username': 'Jenkins', |
| 'percent': 100, |
| 'comment': 'Tests passed', |
| 'url': 'http://jenkins.cloud.fedoraproject.org/', |
| 'status': 'success', |
| } |
| output = self.app.post( |
| '/api/0/test/c/%s/flag' % commit.oid.hex, |
| headers=headers, data=data) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertNotEqual( |
| data['uid'], |
| 'jenkins_build_pagure_100+seed' |
| ) |
| data['flag']['date_created'] = '1510742565' |
| data['flag']['date_updated'] = '1510742565' |
| data['uid'] = 'b1de8f80defd4a81afe2e09f39678087' |
| expected_output = { |
| 'flag': { |
| 'comment': 'Tests passed', |
| 'commit_hash': commit.oid.hex, |
| 'date_created': '1510742565', |
| 'date_updated': '1510742565', |
| 'percent': 100, |
| 'status': 'success', |
| 'url': 'http://jenkins.cloud.fedoraproject.org/', |
| 'user': { |
| 'default_email': 'bar@pingou.com', |
| 'emails': ['bar@pingou.com', 'foo@pingou.com'], |
| 'fullname': 'PY C', |
| 'name': 'pingou'}, |
| 'username': 'Jenkins' |
| }, |
| 'message': 'Flag added', |
| 'uid': 'b1de8f80defd4a81afe2e09f39678087' |
| } |
| self.assertEqual(data, expected_output) |
| |
| mock_email.assert_called_once_with( |
| '\nJenkins flagged the commit ' |
| '`' + commit.oid.hex + '` as success: ' |
| 'Tests passed\n\n' |
| 'http://localhost.localdomain/test/c/' + commit.oid.hex + '\n', |
| 'Coommit #' + commit.oid.hex + ' - Jenkins: success', |
| 'bar@pingou.com', |
| in_reply_to='test-project-1', |
| mail_id='test-commit-1-1', |
| project_name='test', |
| user_from='Jenkins' |
| ) |
| |
| @patch.dict('pagure.config.config', |
| { |
| 'FLAG_STATUSES_LABELS': |
| { |
| 'pend!': 'label-info', |
| 'succeed!': 'label-success', |
| 'fail!': 'label-danger', |
| 'what?': 'label-warning', |
| }, |
| 'FLAG_PENDING': 'pend!', |
| 'FLAG_SUCCESS': 'succeed!', |
| 'FLAG_FAILURE': 'fail!', |
| }) |
| def test_flag_commit_with_custom_flags(self): |
| """ Test flagging when custom flags are set up |
| """ |
| repo_obj = pygit2.Repository(self.git_path) |
| commit = repo_obj.revparse_single('HEAD') |
| |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| send_data = { |
| 'username': 'Jenkins', |
| 'percent': 100, |
| 'comment': 'Tests passed', |
| 'url': 'http://jenkins.cloud.fedoraproject.org/', |
| 'status': 'succeed!', |
| } |
| output = self.app.post( |
| '/api/0/test/c/%s/flag' % commit.oid.hex, |
| headers=headers, data=send_data) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(data['flag']['status'], 'succeed!') |
| |
| |
| send_data['status'] = 'nooooo....' |
| output = self.app.post( |
| '/api/0/test/c/%s/flag' % commit.oid.hex, |
| headers=headers, data=send_data) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| data, |
| { |
| 'errors': {'status': ['Not a valid choice']}, |
| 'error_code': 'EINVALIDREQ', |
| 'error': 'Invalid or incomplete input submitted' |
| } |
| ) |
| |
| def test_commit_flags(self): |
| """ Test retrieving commit flags. """ |
| repo = pagure.lib.query.get_authorized_project(self.session, 'test') |
| repo_obj = pygit2.Repository(self.git_path) |
| commit = repo_obj.revparse_single('HEAD') |
| |
| |
| output = self.app.get('/api/0/test/c/%s/flag' % commit.oid.hex) |
| self.assertEqual(json.loads(output.get_data(as_text=True)), {'total_flags': 0, 'flags': []}) |
| self.assertEqual(output.status_code, 200) |
| |
| |
| pagure.lib.query.add_commit_flag( |
| session=self.session, |
| repo=repo, |
| commit_hash=commit.oid.hex, |
| username='simple-koji-ci', |
| status='pending', |
| percent=None, |
| comment='Build is running', |
| url='https://koji.fp.o/koji...', |
| uid='uid', |
| user='foo', |
| token='aaabbbcccddd' |
| ) |
| |
| pagure.lib.query.add_commit_flag( |
| session=self.session, |
| repo=repo, |
| commit_hash=commit.oid.hex, |
| username='complex-koji-ci', |
| status='success', |
| percent=None, |
| comment='Build succeeded', |
| url='https://koji.fp.o/koji...', |
| uid='uid2', |
| user='foo', |
| token='aaabbbcccddd' |
| ) |
| self.session.commit() |
| |
| output = self.app.get('/api/0/test/c/%s/flag' % commit.oid.hex) |
| data = json.loads(output.get_data(as_text=True)) |
| |
| for f in data['flags']: |
| f['date_created'] = '1510742565' |
| f['date_updated'] = '1510742565' |
| f['commit_hash'] = '62b49f00d489452994de5010565fab81' |
| expected_output = { |
| "flags": [ |
| { |
| "comment": "Build is running", |
| "commit_hash": "62b49f00d489452994de5010565fab81", |
| "date_created": "1510742565", |
| 'date_updated': '1510742565', |
| "percent": None, |
| "status": "pending", |
| "url": "https://koji.fp.o/koji...", |
| "user": { |
| "fullname": "foo bar", |
| "name": "foo" |
| }, |
| "username": "simple-koji-ci" |
| }, |
| { |
| "comment": "Build succeeded", |
| "commit_hash": "62b49f00d489452994de5010565fab81", |
| "date_created": "1510742565", |
| 'date_updated': '1510742565', |
| "percent": None, |
| "status": "success", |
| "url": "https://koji.fp.o/koji...", |
| "user": { |
| "fullname": "foo bar", |
| "name": "foo" |
| }, |
| "username": "complex-koji-ci" |
| } |
| ], |
| "total_flags": 2 |
| } |
| |
| self.assertEqual(data, expected_output) |
| |
| |
| class PagureFlaskApiProjectModifyAclTests(tests.Modeltests): |
| """ Tests for the flask API of pagure for modifying ACLs in a project |
| """ |
| |
| maxDiff = None |
| |
| def setUp(self): |
| """ Set up the environnment, ran before every tests. """ |
| super(PagureFlaskApiProjectModifyAclTests, self).setUp() |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session, project_id=None) |
| tests.create_tokens_acl( |
| self.session, 'aaabbbcccddd', 'modify_project') |
| |
| project = pagure.lib.query._get_project(self.session, 'test') |
| self.assertEquals( |
| project.access_users, |
| {u'admin': [], u'commit': [], u'ticket': []} |
| ) |
| |
| def test_api_modify_acls_no_project(self): |
| """ Test the api_modify_acls method of the flask api when the project |
| doesn't exist """ |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| |
| data = { |
| 'user_type': 'user', |
| 'name': 'bar', |
| 'acl': 'commit' |
| } |
| output = self.app.post( |
| '/api/0/test12345123/git/modifyacls', |
| headers=headers, data=data) |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| expected_output = { |
| 'error_code': 'ENOPROJECT', |
| 'error': 'Project not found' |
| } |
| self.assertEqual(data, expected_output) |
| |
| def test_api_modify_acls_no_user(self): |
| """ Test the api_modify_acls method of the flask api when the user |
| doesn't exist """ |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| |
| data = { |
| 'user_type': 'user', |
| 'name': 'nosuchuser', |
| 'acl': 'commit' |
| } |
| output = self.app.post( |
| '/api/0/test/git/modifyacls', |
| headers=headers, data=data) |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| expected_output = { |
| 'error': 'No such user found', |
| 'error_code': u'ENOUSER' |
| } |
| self.assertEqual(data, expected_output) |
| |
| def test_api_modify_acls_no_group(self): |
| """ Test the api_modify_acls method of the flask api when the group |
| doesn't exist """ |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| |
| data = { |
| 'user_type': 'group', |
| 'name': 'nosuchgroup', |
| 'acl': 'commit' |
| } |
| output = self.app.post( |
| '/api/0/test/git/modifyacls', |
| headers=headers, data=data) |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| expected_output = { |
| 'error': 'Group not found', |
| 'error_code': 'ENOGROUP' |
| } |
| self.assertEqual(data, expected_output) |
| |
| def test_api_modify_acls_no_permission(self): |
| """ Test the api_modify_acls method of the flask api when the user |
| doesn't have permissions """ |
| item = pagure.lib.model.Token( |
| id='foo_token2', |
| user_id=2, |
| project_id=None, |
| expiration=datetime.datetime.utcnow() |
| + datetime.timedelta(days=30) |
| ) |
| self.session.add(item) |
| self.session.commit() |
| tests.create_tokens_acl( |
| self.session, 'foo_token2', 'modify_project') |
| |
| headers = {'Authorization': 'token foo_token2'} |
| |
| data = { |
| 'user_type': 'user', |
| 'name': 'foo', |
| 'acl': 'commit' |
| } |
| output = self.app.post( |
| '/api/0/test/git/modifyacls', |
| headers=headers, data=data) |
| self.assertEqual(output.status_code, 401) |
| data = json.loads(output.get_data(as_text=True)) |
| expected_output = { |
| 'error': 'You are not allowed to modify this project', |
| 'error_code': 'EMODIFYPROJECTNOTALLOWED' |
| } |
| self.assertEqual(data, expected_output) |
| |
| def test_api_modify_acls_neither_user_nor_group(self): |
| """ Test the api_modify_acls method of the flask api when neither |
| user nor group was set """ |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| |
| data = { |
| 'acl': 'commit' |
| } |
| output = self.app.post( |
| '/api/0/test/git/modifyacls', |
| headers=headers, data=data) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| expected_output = { |
| 'error': 'Invalid or incomplete input submitted', |
| 'error_code': 'EINVALIDREQ', |
| 'errors': {'name': ['This field is required.'], |
| 'user_type': ['Not a valid choice']} |
| } |
| self.assertEqual(data, expected_output) |
| |
| def test_api_modify_acls_invalid_acl(self): |
| """ Test the api_modify_acls method of the flask api when the ACL |
| doesn't exist. Must be one of ticket, commit or admin. """ |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| |
| data = { |
| 'user_type': 'user', |
| 'name': 'bar', |
| 'acl': 'invalidacl' |
| } |
| output = self.app.post( |
| '/api/0/test/git/modifyacls', |
| headers=headers, data=data) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| expected_output = { |
| 'error': 'Invalid or incomplete input submitted', |
| 'error_code': 'EINVALIDREQ', |
| 'errors': { |
| 'acl': ['Not a valid choice'] |
| } |
| } |
| self.assertEqual(data, expected_output) |
| |
| def test_api_modify_acls_user(self): |
| """ Test the api_modify_acls method of the flask api for |
| setting an ACL for a user. """ |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| |
| data = { |
| 'user_type': 'user', |
| 'name': 'foo', |
| 'acl': 'commit' |
| } |
| output = self.app.post( |
| '/api/0/test/git/modifyacls', |
| headers=headers, data=data) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data['date_created'] = '1510742565' |
| data['date_modified'] = '1510742566' |
| |
| expected_output = { |
| 'access_groups': {'admin': [], 'commit': [], 'ticket': []}, |
| 'access_users': {'admin': [], |
| 'commit': ['foo'], |
| 'owner': ['pingou'], |
| 'ticket': []}, |
| 'close_status': |
| ['Invalid', 'Insufficient data', 'Fixed', 'Duplicate'], |
| 'custom_keys': [], |
| 'date_created': '1510742565', |
| 'date_modified': '1510742566', |
| 'description': 'test project #1', |
| 'fullname': 'test', |
| 'id': 1, |
| 'milestones': {}, |
| 'name': 'test', |
| 'namespace': None, |
| 'parent': None, |
| 'priorities': {}, |
| 'tags': [], |
| 'url_path': 'test', |
| 'user': {'fullname': 'PY C', 'name': 'pingou'} |
| } |
| self.assertEqual(data, expected_output) |
| |
| def test_api_modify_acls_group(self): |
| """ Test the api_modify_acls method of the flask api for |
| setting an ACL for a group. """ |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| |
| |
| msg = pagure.lib.query.add_group( |
| self.session, |
| group_name='baz', |
| display_name='baz group', |
| description=None, |
| group_type='bar', |
| user='foo', |
| is_admin=False, |
| blacklist=[], |
| ) |
| self.session.commit() |
| self.assertEqual(msg, 'User `foo` added to the group `baz`.') |
| |
| data = { |
| 'user_type': 'group', |
| 'name': 'baz', |
| 'acl': 'ticket' |
| } |
| output = self.app.post( |
| '/api/0/test/git/modifyacls', |
| headers=headers, data=data) |
| |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data['date_created'] = '1510742565' |
| data['date_modified'] = '1510742566' |
| |
| expected_output = { |
| 'access_groups': { |
| 'admin': [], |
| 'commit': [], |
| 'ticket': ['baz'] |
| }, |
| 'access_users': { |
| 'admin': [], |
| 'commit': [], |
| 'owner': ['pingou'], |
| 'ticket': [] |
| }, |
| 'close_status': [ |
| 'Invalid', |
| 'Insufficient data', |
| 'Fixed', |
| 'Duplicate' |
| ], |
| 'custom_keys': [], |
| 'date_created': '1510742565', |
| 'date_modified': '1510742566', |
| 'description': 'test project #1', |
| 'fullname': 'test', |
| 'id': 1, |
| 'milestones': {}, |
| 'name': 'test', |
| 'namespace': None, |
| 'parent': None, |
| 'priorities': {}, |
| 'tags': [], |
| 'url_path': 'test', |
| 'user': {'fullname': 'PY C', 'name': 'pingou'} |
| } |
| self.assertEqual(data, expected_output) |
| |
| def test_api_modify_acls_no_acl(self): |
| """ Test the api_modify_acls method of the flask api when no ACL |
| are specified. """ |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| |
| project = pagure.lib.query._get_project(self.session, 'test') |
| self.assertEquals( |
| project.access_users, |
| {u'admin': [], u'commit': [], u'ticket': []} |
| ) |
| |
| data = { |
| 'user_type': 'user', |
| 'name': 'foo', |
| } |
| output = self.app.post( |
| '/api/0/test/git/modifyacls', headers=headers, data=data) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| expected_output = { |
| "error": "Invalid or incomplete input submitted", |
| "error_code": "EINVALIDREQ", |
| "errors": "User does not have any access on the repo" |
| } |
| |
| self.assertEqual(data, expected_output) |
| |
| def test_api_modify_acls_remove_own_acl_no_access(self): |
| """ Test the api_modify_acls method of the flask api when no ACL |
| are specified, so the user tries to remove their own access but the |
| user is the project owner. """ |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| |
| data = { |
| 'user_type': 'user', |
| 'name': 'pingou', |
| } |
| output = self.app.post( |
| '/api/0/test/git/modifyacls', headers=headers, data=data) |
| self.assertEqual(output.status_code, 400) |
| data = json.loads(output.get_data(as_text=True)) |
| expected_output = { |
| "error": "Invalid or incomplete input submitted", |
| "error_code": "EINVALIDREQ", |
| "errors": "User does not have any access on the repo" |
| } |
| |
| self.assertEqual(data, expected_output) |
| |
| def test_api_modify_acls_remove_own_acl_(self): |
| """ Test the api_modify_acls method of the flask api when no ACL |
| are specified, so the user tries to remove their own access but the |
| user is the project owner. """ |
| |
| self.test_api_modify_acls_user() |
| |
| |
| project = pagure.lib.query._get_project(self.session, 'test') |
| user_foo = pagure.lib.query.search_user(self.session, username='foo') |
| self.assertEquals( |
| project.access_users, |
| {u'admin': [], u'commit': [user_foo], u'ticket': []} |
| ) |
| |
| |
| item = pagure.lib.model.Token( |
| id='foo_test_token', |
| user_id=2, |
| project_id=1, |
| expiration=datetime.datetime.utcnow() + datetime.timedelta(days=10) |
| ) |
| self.session.add(item) |
| self.session.commit() |
| tests.create_tokens_acl( |
| self.session, 'foo_test_token', 'modify_project') |
| |
| headers = {'Authorization': 'token foo_test_token'} |
| |
| data = { |
| 'user_type': 'user', |
| 'name': 'foo', |
| } |
| output = self.app.post( |
| '/api/0/test/git/modifyacls', headers=headers, data=data) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data['date_created'] = '1510742565' |
| data['date_modified'] = '1510742566' |
| |
| expected_output = { |
| "access_groups": { |
| "admin": [], |
| "commit": [], |
| "ticket": [] |
| }, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": [ |
| "pingou" |
| ], |
| "ticket": [] |
| }, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate" |
| ], |
| "custom_keys": [], |
| "date_created": "1510742565", |
| "date_modified": "1510742566", |
| "description": "test project #1", |
| "fullname": "test", |
| "id": 1, |
| "milestones": {}, |
| "name": "test", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": [], |
| "url_path": "test", |
| "user": { |
| "fullname": "PY C", |
| "name": "pingou" |
| } |
| } |
| |
| self.assertEqual(data, expected_output) |
| |
| |
| self.session = pagure.lib.query.create_session(self.dbpath) |
| project = pagure.lib.query._get_project(self.session, 'test') |
| self.assertEquals( |
| project.access_users, |
| {u'admin': [], u'commit': [], u'ticket': []} |
| ) |
| |
| def test_api_modify_acls_remove_someone_else_acl(self): |
| """ Test the api_modify_acls method of the flask api an admin tries |
| to remove access from someone else. """ |
| |
| self.test_api_modify_acls_user() |
| |
| |
| project = pagure.lib.query._get_project(self.session, 'test') |
| user_foo = pagure.lib.query.search_user(self.session, username='foo') |
| self.assertEquals( |
| project.access_users, |
| {u'admin': [], u'commit': [user_foo], u'ticket': []} |
| ) |
| |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| data = { |
| 'user_type': 'user', |
| 'name': 'foo', |
| } |
| output = self.app.post( |
| '/api/0/test/git/modifyacls', headers=headers, data=data) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| data['date_created'] = '1510742565' |
| data['date_modified'] = '1510742566' |
| |
| expected_output = { |
| "access_groups": { |
| "admin": [], |
| "commit": [], |
| "ticket": [] |
| }, |
| "access_users": { |
| "admin": [], |
| "commit": [], |
| "owner": [ |
| "pingou" |
| ], |
| "ticket": [] |
| }, |
| "close_status": [ |
| "Invalid", |
| "Insufficient data", |
| "Fixed", |
| "Duplicate" |
| ], |
| "custom_keys": [], |
| "date_created": "1510742565", |
| "date_modified": "1510742566", |
| "description": "test project #1", |
| "fullname": "test", |
| "id": 1, |
| "milestones": {}, |
| "name": "test", |
| "namespace": None, |
| "parent": None, |
| "priorities": {}, |
| "tags": [], |
| "url_path": "test", |
| "user": { |
| "fullname": "PY C", |
| "name": "pingou" |
| } |
| } |
| |
| self.assertEqual(data, expected_output) |
| |
| |
| self.session = pagure.lib.query.create_session(self.dbpath) |
| project = pagure.lib.query._get_project(self.session, 'test') |
| self.assertEquals( |
| project.access_users, |
| {u'admin': [], u'commit': [], u'ticket': []} |
| ) |
| |
| |
| class PagureFlaskApiProjectOptionsTests(tests.Modeltests): |
| """ Tests for the flask API of pagure for modifying options ofs a project |
| """ |
| |
| maxDiff = None |
| |
| def setUp(self): |
| """ Set up the environnment, ran before every tests. """ |
| super(PagureFlaskApiProjectOptionsTests, self).setUp() |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session, project_id=None) |
| tests.create_tokens_acl( |
| self.session, 'aaabbbcccddd', 'modify_project') |
| |
| project = pagure.lib.query._get_project(self.session, 'test') |
| self.assertEquals( |
| project.access_users, |
| {u'admin': [], u'commit': [], u'ticket': []} |
| ) |
| |
| def test_api_get_project_options_wrong_project(self): |
| """ Test accessing api_get_project_options w/o auth header. """ |
| |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| output = self.app.get('/api/0/unknown/options', headers=headers) |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| data, |
| {u'error': u'Project not found', u'error_code': u'ENOPROJECT'} |
| ) |
| |
| def test_api_get_project_options_wo_header(self): |
| """ Test accessing api_get_project_options w/o auth header. """ |
| |
| output = self.app.get('/api/0/test/options') |
| self.assertEqual(output.status_code, 401) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| data, |
| { |
| u'error': u'Invalid or expired token. Please visit ' |
| 'http://localhost.localdomain/settings#api-keys to get ' |
| 'or renew your API token.', |
| u'error_code': u'EINVALIDTOK', |
| u'errors': u'Invalid token', |
| } |
| ) |
| |
| def test_api_get_project_options_w_header(self): |
| """ Test accessing api_get_project_options w/ auth header. """ |
| |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| output = self.app.get('/api/0/test/options', headers=headers) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| data, |
| { |
| "settings": { |
| "Enforce_signed-off_commits_in_pull-request": False, |
| "Minimum_score_to_merge_pull-request": -1, |
| "Only_assignee_can_merge_pull-request": False, |
| "Web-hooks": None, |
| "always_merge": False, |
| "disable_non_fast-forward_merges": False, |
| "fedmsg_notifications": True, |
| "issue_tracker": True, |
| "issue_tracker_read_only": False, |
| "issues_default_to_private": False, |
| "mqtt_notifications": True, |
| "notify_on_commit_flag": False, |
| "notify_on_pull-request_flag": False, |
| "open_metadata_access_to_all": False, |
| "project_documentation": False, |
| "pull_request_access_only": False, |
| "pull_requests": True, |
| "stomp_notifications": True |
| }, |
| "status": "ok" |
| } |
| ) |
| |
| def test_api_modify_project_options_wrong_project(self): |
| """ Test accessing api_modify_project_options w/ an invalid project. |
| """ |
| |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| output = self.app.post('/api/0/unknown/options/update', headers=headers) |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| data, |
| {u'error': u'Project not found', u'error_code': u'ENOPROJECT'} |
| ) |
| |
| def test_api_modify_project_options_wo_header(self): |
| """ Test accessing api_modify_project_options w/o auth header. """ |
| |
| output = self.app.post('/api/0/test/options/update') |
| self.assertEqual(output.status_code, 401) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| data, |
| { |
| u'error': u'Invalid or expired token. Please visit ' |
| 'http://localhost.localdomain/settings#api-keys to get ' |
| 'or renew your API token.', |
| u'error_code': u'EINVALIDTOK', |
| u'errors': u'Invalid token', |
| } |
| ) |
| |
| def test_api_modify_project_options_no_data(self): |
| """ Test accessing api_modify_project_options w/ auth header. """ |
| |
| |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| output = self.app.get('/api/0/test/options', headers=headers) |
| self.assertEqual(output.status_code, 200) |
| before = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| before, |
| { |
| "settings": { |
| "Enforce_signed-off_commits_in_pull-request": False, |
| "Minimum_score_to_merge_pull-request": -1, |
| "Only_assignee_can_merge_pull-request": False, |
| "Web-hooks": None, |
| "always_merge": False, |
| "disable_non_fast-forward_merges": False, |
| "fedmsg_notifications": True, |
| "issue_tracker": True, |
| "issue_tracker_read_only": False, |
| "issues_default_to_private": False, |
| "mqtt_notifications": True, |
| "notify_on_commit_flag": False, |
| "notify_on_pull-request_flag": False, |
| "open_metadata_access_to_all": False, |
| "project_documentation": False, |
| "pull_request_access_only": False, |
| "pull_requests": True, |
| "stomp_notifications": True |
| }, |
| "status": "ok" |
| } |
| ) |
| |
| |
| data = {} |
| output = self.app.post( |
| '/api/0/test/options/update', headers=headers, data=data) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| data, |
| { |
| u'message': u'No settings to change', |
| u'status': u'ok' |
| } |
| ) |
| |
| |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| output = self.app.get('/api/0/test/options', headers=headers) |
| self.assertEqual(output.status_code, 200) |
| after = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(after, before) |
| |
| def test_api_modify_project_options(self): |
| """ Test accessing api_modify_project_options w/ auth header. """ |
| |
| |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| output = self.app.get('/api/0/test/options', headers=headers) |
| self.assertEqual(output.status_code, 200) |
| before = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| before, |
| { |
| "settings": { |
| "Enforce_signed-off_commits_in_pull-request": False, |
| "Minimum_score_to_merge_pull-request": -1, |
| "Only_assignee_can_merge_pull-request": False, |
| "Web-hooks": None, |
| "always_merge": False, |
| "disable_non_fast-forward_merges": False, |
| "fedmsg_notifications": True, |
| "issue_tracker": True, |
| "issue_tracker_read_only": False, |
| "issues_default_to_private": False, |
| "mqtt_notifications": True, |
| "notify_on_commit_flag": False, |
| "notify_on_pull-request_flag": False, |
| "open_metadata_access_to_all": False, |
| "project_documentation": False, |
| "pull_request_access_only": False, |
| "pull_requests": True, |
| "stomp_notifications": True |
| }, |
| "status": "ok" |
| } |
| ) |
| |
| |
| data = {"issues_default_to_private": True} |
| output = self.app.post( |
| '/api/0/test/options/update', headers=headers, data=data) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| data, |
| { |
| u'message': u'Edited successfully settings of repo: test', |
| u'status': u'ok' |
| } |
| ) |
| |
| |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| output = self.app.get('/api/0/test/options', headers=headers) |
| self.assertEqual(output.status_code, 200) |
| after = json.loads(output.get_data(as_text=True)) |
| self.assertNotEqual(before, after) |
| before["settings"]["issues_default_to_private"] = True |
| self.assertEqual(after, before) |
| |
| class PagureFlaskApiProjectCreateAPITokenTests(tests.Modeltests): |
| """ Tests for the flask API of pagure for creating user project API token |
| """ |
| |
| maxDiff = None |
| |
| def setUp(self): |
| """ Set up the environnment, ran before every tests. """ |
| super(PagureFlaskApiProjectCreateAPITokenTests, self).setUp() |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session, project_id=None) |
| tests.create_tokens_acl( |
| self.session, 'aaabbbcccddd', 'modify_project') |
| |
| def test_api_createapitoken_as_owner(self): |
| """ Test accessing api_project_create_token as owner. """ |
| |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| project = pagure.lib.query._get_project(self.session, 'test') |
| tdescription = 'my new token' |
| |
| |
| data = { |
| 'description': tdescription, |
| 'acls': ['pull_request_merge', 'pull_request_comment'] |
| } |
| output = self.app.post('/api/0/test/token/new', |
| headers=headers, data=data) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| tid = pagure.lib.query.search_token( |
| self.session, None, description=tdescription)[0].id |
| self.assertEqual( |
| data, |
| {"token": { |
| "description": tdescription, |
| "id": tid |
| } |
| } |
| ) |
| |
| |
| data = { |
| 'description': tdescription, |
| 'acl': ['foo', 'bar'] |
| } |
| output = self.app.post('/api/0/test/token/new', |
| headers=headers, data=data) |
| self.assertEqual(output.status_code, 400) |
| |
| def test_api_createapitoken_as_admin(self): |
| """ Test accessing api_project_create_token as admin. """ |
| |
| project = pagure.lib.query._get_project(self.session, 'test') |
| |
| |
| pagure.lib.query.add_user_to_project( |
| self.session, project, |
| new_user='foo', |
| user='pingou', |
| access='admin' |
| ) |
| self.session.commit() |
| |
| |
| token = pagure.lib.query.add_token_to_user( |
| self.session, |
| project=None, |
| acls=['modify_project'], |
| username='foo') |
| |
| |
| headers = {'Authorization': 'token %s' % token.id} |
| tdescription = 'my new token' |
| |
| |
| data = { |
| 'description': tdescription, |
| 'acls': ['pull_request_merge', 'pull_request_comment'] |
| } |
| output = self.app.post('/api/0/test/token/new', |
| headers=headers, data=data) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| tid = pagure.lib.query.search_token( |
| self.session, None, user='foo', description=tdescription)[0].id |
| self.assertEqual( |
| data, |
| {"token": { |
| "description": tdescription, |
| "id": tid |
| } |
| } |
| ) |
| |
| def test_api_createapitoken_as_unauthorized(self): |
| """ Test accessing api_project_create_token as project admin |
| but with unauthorized token ACL. |
| """ |
| |
| project = pagure.lib.query._get_project(self.session, 'test') |
| |
| |
| pagure.lib.query.add_user_to_project( |
| self.session, project, |
| new_user='foo', |
| user='pingou', |
| access='admin' |
| ) |
| self.session.commit() |
| |
| |
| pagure.lib.query.add_token_to_user( |
| self.session, |
| project=None, |
| acls=['create_branch'], |
| username='foo') |
| mtoken = pagure.lib.query.search_token( |
| self.session, ['create_branch'], user='foo')[0] |
| |
| |
| headers = {'Authorization': 'token %s' % mtoken.id} |
| tdescription = 'my new token' |
| |
| |
| data = { |
| 'description': tdescription, |
| 'acls': ['pull_request_merge', 'pull_request_comment'] |
| } |
| output = self.app.post('/api/0/test/token/new', |
| headers=headers, data=data) |
| self.assertEqual(output.status_code, 401) |
| |
| def test_api_createapitoken_as_unauthorized_2(self): |
| """ Test accessing api_project_create_token as project user |
| with unauthorized token ACL. |
| """ |
| |
| project = pagure.lib.query._get_project(self.session, 'test') |
| |
| |
| pagure.lib.query.add_user_to_project( |
| self.session, project, |
| new_user='foo', |
| user='pingou', |
| access='commit' |
| ) |
| self.session.commit() |
| |
| |
| pagure.lib.query.add_token_to_user( |
| self.session, |
| project=None, |
| acls=['modify_project'], |
| username='foo') |
| mtoken = pagure.lib.query.search_token( |
| self.session, ['modify_project'], user='foo')[0] |
| |
| |
| headers = {'Authorization': 'token %s' % mtoken.id} |
| tdescription = 'my new token' |
| |
| |
| data = { |
| 'description': tdescription, |
| 'acls': ['pull_request_merge', 'pull_request_comment'] |
| } |
| output = self.app.post('/api/0/test/token/new', |
| headers=headers, data=data) |
| self.assertEqual(output.status_code, 401) |
| |
| |
| class PagureFlaskApiProjectConnectorTests(tests.Modeltests): |
| """ Tests for the flask API of pagure for getting connector of a project |
| """ |
| |
| maxDiff = None |
| |
| def setUp(self): |
| """ Set up the environnment, ran before every tests. """ |
| super(PagureFlaskApiProjectConnectorTests, self).setUp() |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session, project_id=None) |
| tests.create_tokens_acl( |
| self.session, 'aaabbbcccddd', 'modify_project') |
| |
| |
| def test_api_get_project_connector_as_owner(self): |
| """ Test accessing api_get_project_connector as project owner. """ |
| |
| project = pagure.lib.query._get_project(self.session, 'test') |
| |
| |
| pagure.lib.query.add_token_to_user( |
| self.session, |
| project=project, |
| acls=['pull_request_merge'], |
| username='pingou') |
| ctokens = pagure.lib.query.search_token( |
| self.session, ['pull_request_merge'], user='pingou') |
| self.assertEqual(len(ctokens), 1) |
| |
| |
| headers = {'Authorization': 'token aaabbbcccddd'} |
| output = self.app.get('/api/0/test/connector', headers=headers) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| data, |
| {"connector": { |
| "hook_token": project.hook_token, |
| "api_tokens": [ |
| {'description': t.description, |
| 'id': t.id, |
| 'expired': False} for t in ctokens] |
| }, |
| "status": "ok" |
| } |
| ) |
| |
| def test_api_get_project_connector_as_admin(self): |
| """ Test accessing api_get_project_connector as project admin """ |
| |
| project = pagure.lib.query._get_project(self.session, 'test') |
| |
| |
| pagure.lib.query.add_user_to_project( |
| self.session, project, |
| new_user='foo', |
| user='pingou', |
| access='admin' |
| ) |
| self.session.commit() |
| |
| |
| pagure.lib.query.add_token_to_user( |
| self.session, |
| project=None, |
| acls=['modify_project'], |
| username='foo') |
| mtoken = pagure.lib.query.search_token( |
| self.session, ['modify_project'], user='foo')[0] |
| |
| |
| pagure.lib.query.add_token_to_user( |
| self.session, |
| project=project, |
| acls=['pull_request_merge'], |
| username='foo') |
| ctokens = pagure.lib.query.search_token( |
| self.session, ['pull_request_merge'], user='foo') |
| self.assertEqual(len(ctokens), 1) |
| |
| |
| headers = {'Authorization': 'token %s' % mtoken.id} |
| output = self.app.get('/api/0/test/connector', headers=headers) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| data, |
| {"connector": { |
| "hook_token": project.hook_token, |
| "api_tokens": [ |
| {'description': t.description, |
| 'id': t.id, |
| 'expired': False} for t in ctokens] |
| }, |
| "status": "ok" |
| } |
| ) |
| |
| def test_api_get_project_connector_as_unauthorized(self): |
| """ Test accessing api_get_project_connector as project admin |
| but with unauthorized token ACL |
| """ |
| |
| project = pagure.lib.query._get_project(self.session, 'test') |
| |
| |
| pagure.lib.query.add_user_to_project( |
| self.session, project, |
| new_user='foo', |
| user='pingou', |
| access='admin' |
| ) |
| self.session.commit() |
| |
| |
| pagure.lib.query.add_token_to_user( |
| self.session, |
| project=None, |
| acls=['create_project'], |
| username='foo') |
| mtoken = pagure.lib.query.search_token( |
| self.session, ['create_project'], user='foo')[0] |
| |
| |
| headers = {'Authorization': 'token %s' % mtoken.id} |
| output = self.app.get('/api/0/test/connector', headers=headers) |
| self.assertEqual(output.status_code, 401) |
| |
| def test_api_get_project_connector_as_unauthorized_2(self): |
| """ Test accessing api_get_project_connector as project |
| but with unauthorized token ACL |
| """ |
| |
| project = pagure.lib.query._get_project(self.session, 'test') |
| |
| |
| pagure.lib.query.add_user_to_project( |
| self.session, project, |
| new_user='foo', |
| user='pingou', |
| access='commit' |
| ) |
| self.session.commit() |
| |
| |
| pagure.lib.query.add_token_to_user( |
| self.session, |
| project=None, |
| acls=['modify_project'], |
| username='foo') |
| mtoken = pagure.lib.query.search_token( |
| self.session, ['modify_project'], user='foo')[0] |
| |
| |
| headers = {'Authorization': 'token %s' % mtoken.id} |
| output = self.app.get('/api/0/test/connector', headers=headers) |
| self.assertEqual(output.status_code, 401) |
| |
| if __name__ == '__main__': |
| unittest.main(verbosity=2) |