| |
| |
| """ |
| (c) 2015 - Copyright Red Hat Inc |
| |
| Authors: |
| Pierre-Yves Chibon <pingou@pingoured.fr> |
| |
| """ |
| |
| from __future__ import unicode_literals, absolute_import |
| |
| import unittest |
| import shutil |
| import sys |
| import os |
| |
| import json |
| from mock import patch, MagicMock |
| |
| sys.path.insert( |
| 0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..") |
| ) |
| |
| import pagure.api |
| import pagure.flask_app |
| import pagure.lib.query |
| import tests |
| |
| |
| class PagureFlaskApitests(tests.SimplePagureTest): |
| """ Tests for flask API controller of pagure """ |
| |
| maxDiff = None |
| |
| def test_api_doc(self): |
| """ Test the API documentation page. """ |
| print(dir(self.app)) |
| output = self.app.get("/api/0/") |
| output_text = output.get_data(as_text=True) |
| self.assertIn("<title> API | pagure - Pagure</title>\n", output_text) |
| self.assertIn( |
| " Pagure API Reference\n </h2>\n", output_text |
| ) |
| |
| def test_api_doc_authenticated(self): |
| """ Test the API documentation page. """ |
| user = tests.FakeUser(username="foo") |
| with tests.user_set(self.app.application, user): |
| output = self.app.get("/api/0/") |
| output_text = output.get_data(as_text=True) |
| self.assertIn( |
| "<title> API | pagure - Pagure</title>\n", output_text |
| ) |
| self.assertIn( |
| " Pagure API Reference\n </h2>\n", output_text |
| ) |
| |
| def test_api_get_request_data(self): |
| data = {"foo": "bar"} |
| |
| |
| self._app.teardown_request_funcs = {} |
| with self._app.test_request_context( |
| "/api/0/version", method="POST", data=data |
| ): |
| self.assertEqual(pagure.api.get_request_data()["foo"], "bar") |
| data = json.dumps(data) |
| with self._app.test_request_context( |
| "/api/0/version", data=data, content_type="application/json" |
| ): |
| self.assertEqual(pagure.api.get_request_data()["foo"], "bar") |
| |
| def test_api_version_old_url(self): |
| """ Test the api_version function. """ |
| output = self.app.get("/api/0/version") |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(data["version"], pagure.__api_version__) |
| self.assertEqual(sorted(data.keys()), ["version"]) |
| |
| def test_api_version_new_url(self): |
| """ Test the api_version function at its new url. """ |
| output = self.app.get("/api/0/-/version") |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(data["version"], pagure.__api_version__) |
| self.assertEqual(sorted(data.keys()), ["version"]) |
| |
| def test_api_project_tags(self): |
| """ Test the api_project_tags function. """ |
| tests.create_projects(self.session) |
| |
| output = self.app.get("/api/0/foo/tags/") |
| self.assertEqual(output.status_code, 404) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(set(data.keys()), set(["output", "error"])) |
| self.assertEqual(data["output"], "notok") |
| self.assertEqual(data["error"], "Project not found") |
| |
| output = self.app.get("/api/0/test/tags/") |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(sorted(data.keys()), ["tags", "total_tags"]) |
| self.assertEqual(data["tags"], []) |
| self.assertEqual(data["total_tags"], 0) |
| |
| |
| item = pagure.lib.model.Issue( |
| id=1, |
| uid="foobar", |
| project_id=1, |
| title="issue", |
| content="a bug report", |
| user_id=1, |
| ) |
| self.session.add(item) |
| self.session.commit() |
| item = pagure.lib.model.TagColored( |
| tag="tag1", tag_color="DeepBlueSky", project_id=1 |
| ) |
| self.session.add(item) |
| self.session.commit() |
| item = pagure.lib.model.TagIssueColored( |
| issue_uid="foobar", tag_id=item.id |
| ) |
| self.session.add(item) |
| self.session.commit() |
| |
| output = self.app.get("/api/0/test/tags/") |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(sorted(data.keys()), ["tags", "total_tags"]) |
| self.assertEqual(data["tags"], ["tag1"]) |
| self.assertEqual(data["total_tags"], 1) |
| |
| output = self.app.get("/api/0/test/tags/?pattern=t") |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(sorted(data.keys()), ["tags", "total_tags"]) |
| self.assertEqual(data["tags"], ["tag1"]) |
| self.assertEqual(data["total_tags"], 1) |
| |
| output = self.app.get("/api/0/test/tags/?pattern=p") |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(sorted(data.keys()), ["tags", "total_tags"]) |
| self.assertEqual(data["tags"], []) |
| self.assertEqual(data["total_tags"], 0) |
| |
| def test_api_groups(self): |
| """ Test the api_groups function. """ |
| |
| |
| item = pagure.lib.model.PagureGroup( |
| group_name="group1", |
| group_type="user", |
| display_name="User group", |
| user_id=1, |
| ) |
| self.session.add(item) |
| |
| item = pagure.lib.model.PagureGroup( |
| group_name="rel-eng", |
| group_type="user", |
| display_name="Release engineering group", |
| user_id=1, |
| ) |
| self.session.add(item) |
| self.session.commit() |
| |
| output = self.app.get("/api/0/groups") |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(data["groups"], ["group1", "rel-eng"]) |
| self.assertEqual( |
| sorted(data.keys()), ["groups", "pagination", "total_groups"] |
| ) |
| self.assertEqual(data["total_groups"], 2) |
| |
| output = self.app.get("/api/0/groups?pattern=re") |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(data["groups"], ["rel-eng"]) |
| self.assertEqual( |
| sorted(data.keys()), ["groups", "pagination", "total_groups"] |
| ) |
| self.assertEqual(data["total_groups"], 1) |
| |
| def test_api_whoami_unauth(self): |
| """ Test the api_whoami function. """ |
| |
| output = self.app.post("/api/0/-/whoami") |
| self.assertEqual(output.status_code, 401) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| data, |
| { |
| "error": "Invalid or expired token. Please visit " |
| "http://localhost.localdomain/settings#nav-api-tab to get or " |
| "renew your API token.", |
| "error_code": "EINVALIDTOK", |
| }, |
| ) |
| |
| def test_api_whoami_invalid_auth(self): |
| """ Test the api_whoami function with an invalid token. """ |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session) |
| |
| headers = {"Authorization": "token invalid"} |
| |
| output = self.app.post("/api/0/-/whoami", headers=headers) |
| self.assertEqual(output.status_code, 401) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| data, |
| { |
| "error": "Invalid or expired token. Please visit " |
| "http://localhost.localdomain/settings#nav-api-tab to get or " |
| "renew your API token.", |
| "error_code": "EINVALIDTOK", |
| "errors": "Invalid token", |
| }, |
| ) |
| |
| def test_api_whoami_auth(self): |
| """ Test the api_whoami function with a valid token. """ |
| tests.create_projects(self.session) |
| tests.create_tokens(self.session) |
| |
| headers = {"Authorization": "token aaabbbcccddd"} |
| |
| output = self.app.post("/api/0/-/whoami", headers=headers) |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(data, {"username": "pingou"}) |
| |
| def test_api_error_codes(self): |
| """ Test the api_error_codes endpoint. """ |
| output = self.app.get("/api/0/-/error_codes") |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(len(data), 37) |
| self.assertEqual( |
| sorted(data.keys()), |
| sorted( |
| [ |
| "EDATETIME", |
| "EDBERROR", |
| "EGITERROR", |
| "EINVALIDISSUEFIELD", |
| "EINVALIDISSUEFIELD_LINK", |
| "EINVALIDPERPAGEVALUE", |
| "EINVALIDPRIORITY", |
| "EINVALIDREQ", |
| "EINVALIDTOK", |
| "EISSUENOTALLOWED", |
| "EMODIFYPROJECTNOTALLOWED", |
| "ENEWPROJECTDISABLED", |
| "ENOCODE", |
| "ENOCOMMENT", |
| "ENOCOMMIT", |
| "ENOGROUP", |
| "ENOISSUE", |
| "ENOPRCLOSE", |
| "ENOPROJECT", |
| "ENOPROJECTS", |
| "ENOPRSTATS", |
| "ENOREQ", |
| "ENOSIGNEDOFF", |
| "ENOTASSIGNED", |
| "ENOTASSIGNEE", |
| "ENOTHIGHENOUGH", |
| "ENOTMAINADMIN", |
| "ENOUSER", |
| "EPRCONFLICTS", |
| "EPRNOTALLOWED", |
| "EPRSCORE", |
| "EPULLREQUESTSDISABLED", |
| "ETIMESTAMP", |
| "ETRACKERDISABLED", |
| "ETRACKERREADONLY", |
| "EUBLOCKED", |
| "EREBASENOTALLOWED", |
| ] |
| ), |
| ) |
| |
| @patch("pagure.lib.tasks.get_result") |
| def test_api_task_status(self, get_result): |
| async_result = MagicMock() |
| async_result.status = "running" |
| async_result.ready.return_value = False |
| get_result.return_value = async_result |
| output = self.app.get("/api/0/task/123abc/status/") |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual(data, {"ready": False, "status": "running"}) |
| |
| async_result = MagicMock() |
| async_result.status = "finished" |
| async_result.ready.return_value = True |
| async_result.successful.return_value = True |
| get_result.return_value = async_result |
| output = self.app.get("/api/0/task/123abc/status/") |
| self.assertEqual(output.status_code, 200) |
| data = json.loads(output.get_data(as_text=True)) |
| self.assertEqual( |
| data, {"ready": True, "status": "finished", "successful": True} |
| ) |
| |
| |
| if __name__ == "__main__": |
| unittest.main(verbosity=2) |