From a1f4b00e27ec17e0a8af39ea170ae7e0ade89a24 Mon Sep 17 00:00:00 2001 From: Lubomír Sedlář Date: Apr 28 2018 19:57:17 +0000 Subject: [PATCH 1/2] Add tests for pag.client module PEP8 violations in the module are fixed, one one hardcoded URL is replaced with the URL from attribute. --- diff --git a/pag/client.py b/pag/client.py index 9079801..5344e5d 100644 --- a/pag/client.py +++ b/pag/client.py @@ -6,6 +6,7 @@ import fedora.client from pag.utils import repo_url + class PagureException(Exception): pass @@ -68,7 +69,7 @@ class Pagure(fedora.client.OpenIdBaseClient): soup = bs4.BeautifulSoup(response.text, "html.parser") data = { - 'csrf_token' : soup.find(id='csrf_token').attrs['value'], + 'csrf_token': soup.find(id='csrf_token').attrs['value'], 'title': title, 'issue_content': description, 'private': private @@ -143,7 +144,7 @@ class Pagure(fedora.client.OpenIdBaseClient): return repo_url(name) def submit_pull_request(self, name, base, head, title, comment): - url = 'https://pagure.io/{name}/diff/{base}..{head}' + url = self.base_url + '/{name}/diff/{base}..{head}' url = url.format(name=name, base=base, head=head) response = self._session.get(url) diff --git a/test/test_client.py b/test/test_client.py new file mode 100644 index 0000000..d6635b0 --- /dev/null +++ b/test/test_client.py @@ -0,0 +1,385 @@ +import mock +import unittest + + +from pag import client + +BASE_URL = 'https://example.com' + + +class MockResponse(mock.Mock): + def __init__(self, r): + super(MockResponse, self).__init__() + if isinstance(r, tuple): + self.text = r[0] + self.url = r[1] + else: + self.text = r + self.url = None + + +class MockSession(mock.Mock): + def __init__(self, gets, posts=[]): + get_responses = [MockResponse(r) if r else None for r in gets] + post_responses = [MockResponse(r) if r else None for r in posts] + + super(MockSession, self).__init__() + self.get = mock.Mock(side_effect=get_responses) + self.post = mock.Mock(side_effect=post_responses) + + +def _get(url=''): + return mock.call.get(BASE_URL + url) + + +def _post(url, **kwargs): + return mock.call.post(BASE_URL + url, **kwargs) + + +class ClientTest(unittest.TestCase): + + def _make_client(self, gets, posts=[]): + self.client = client.Pagure(BASE_URL) + self.client._session = MockSession(gets, posts) + + def assertCalls(self, *calls): + print(self.client._session.mock_calls) + self.assertEqual(self.client._session.mock_calls, list(calls)) + + +class IsLoggedInTest(ClientTest): + + def test_is_logged_in(self): + self._make_client(['Do you want to logout?']) + self.assertTrue(self.client.is_logged_in) + self.assertCalls(_get()) + + def test_is_logged_in_when_not_logged_in(self): + self._make_client(['You are not logged in.']) + self.assertFalse(self.client.is_logged_in) + self.assertCalls(_get()) + + +class CreateTest(ClientTest): + + def test_success(self): + self._make_client([ + 'Checking if user can logout', + '', + ], [ + 'Ok cool created' + ]) + + url = self.client.create('name', 'description') + + # This is generated by utils.repo_url() which ignores BASE_URL + self.assertEqual(url, 'https://pagure.io/name') + + self.assertCalls( + _get(), + _get('/new'), + _post('/new', data={'name': 'name', 'description': 'description', + 'csrf_token': 'BEEFCAFE'}), + ) + + def test_not_logged_in(self): + self._make_client([ + 'Not logged in, bro', + ]) + + with self.assertRaises(client.PagureException): + self.client.create('name', 'description') + + self.assertCalls( + _get(), + ) + + def test_fail_to_get_csrf(self): + self._make_client([ + 'Checking if user can logout', + None, + ]) + + with self.assertRaises(client.PagureException): + self.client.create('name', 'description') + + self.assertCalls( + _get(), + _get('/new'), + ) + + def test_fail_request(self): + self._make_client([ + 'Checking if user can logout', + '', + ], [ + None, + ]) + + with self.assertRaises(client.PagureException): + self.client.create('name', 'description') + + self.assertCalls( + _get(), + _get('/new'), + _post('/new', data=mock.ANY), + ) + + +class CreateIssueTest(ClientTest): + + def test_success(self): + self._make_client([ + 'Checking if user can logout', + '', + ], [ + 'Issue #123: title', + ]) + + url = self.client.create_issue('foo', 'title', 'description', + private=True) + + self.assertEqual(url, BASE_URL + '/foo/issue/123') + + self.assertCalls( + _get(), + _get('/foo/new_issue'), + _post('/foo/new_issue', data={'title': 'title', + 'issue_content': 'description', + 'private': True, + 'csrf_token': 'BEEFCAFE'}), + ) + + def test_not_logged_in(self): + self._make_client([ + 'Not logged in, bro', + ]) + + with self.assertRaises(client.PagureException): + self.client.create_issue('foo', 'title', 'description', + private=True) + + self.assertCalls( + _get(), + ) + + def test_fail_to_get_csrf(self): + self._make_client([ + 'Checking if user can logout', + None, + ]) + + with self.assertRaises(client.PagureException): + self.client.create_issue('foo', 'title', 'description', + private=True) + + self.assertCalls( + _get(), + _get('/foo/new_issue'), + ) + + def test_fail_request(self): + self._make_client([ + 'Checking if user can logout', + '', + ], [ + None, + ]) + + with self.assertRaises(client.PagureException): + self.client.create_issue('foo', 'title', 'description', + private=True) + + self.assertCalls( + _get(), + _get('/foo/new_issue'), + _post('/foo/new_issue', data=mock.ANY), + ) + + +class ForkTest(ClientTest): + + def test_success(self): + self._make_client([ + 'Checking if user can logout', + '', + ], [ + 'OK cool', + ]) + + url = self.client.fork('foo') + + # This is generated by utils.repo_url() which ignores BASE_URL + # It's actually wrong... + self.assertEqual(url, 'https://pagure.io/foo') + + self.assertCalls( + _get(), + _get('/foo'), + _post('/do_fork/foo', data={'csrf_token': 'BEEFCAFE'}), + ) + + def test_not_logged_in(self): + self._make_client([ + 'Not logged in, bro', + ]) + + with self.assertRaises(client.PagureException): + self.client.fork('foo') + + self.assertCalls( + _get(), + ) + + def test_fail_to_get_csrf(self): + self._make_client([ + 'Checking if user can logout', + None, + ]) + + with self.assertRaises(client.PagureException): + self.client.fork('foo') + + self.assertCalls( + _get(), + _get('/foo'), + ) + + def test_fail_request(self): + self._make_client([ + 'Checking if user can logout', + '', + ], [ + None, + ]) + + with self.assertRaises(client.PagureException): + self.client.fork('foo') + + self.assertCalls( + _get(), + _get('/foo'), + _post('/do_fork/foo', data=mock.ANY), + ) + + +class UploadTest(ClientTest): + + def test_success(self): + self._make_client([ + 'Checking if user can logout', + '', + ], [ + 'OK', + ]) + + with mock.patch('pag.client.open', mock.mock_open()) as m: + res = self.client.upload('foo', 'file.txt') + + self.assertIsNone(res) + self.assertCalls( + _get(), + _get('/foo/upload'), + _post('/foo/upload', + data={'csrf_token': 'BEEFCAFE'}, + files={'filestream': m.return_value}) + ) + + def test_not_logged_in(self): + self._make_client([ + 'Not logged in, bro', + ]) + + with self.assertRaises(client.PagureException): + self.client.upload('foo', 'file.txt') + + self.assertCalls( + _get(), + ) + + def test_fail_to_get_csrf(self): + self._make_client([ + 'Checking if user can logout', + None, + ]) + + with self.assertRaises(client.PagureException): + self.client.upload('foo', 'file.txt') + + self.assertCalls( + _get(), + _get('/foo/upload'), + ) + + def test_fail_request(self): + self._make_client([ + 'Checking if user can logout', + '', + ], [ + 'Boom', + ]) + + with mock.patch('pag.client.open', mock.mock_open()) as m: + res = self.client.upload('foo', 'file.txt') + + self.assertEqual(res, 'Boom') + self.assertCalls( + _get(), + _get('/foo/upload'), + _post('/foo/upload', + data={'csrf_token': 'BEEFCAFE'}, + files={'filestream': m.return_value}) + ) + + +class SubmitPullRequestTest(ClientTest): + + def test_success(self): + self._make_client([ + '', + ], [ + ('Cool', 'returned url') + ]) + + res = self.client.submit_pull_request('foo', 'master', 'feature', + 'title', 'body') + + self.assertEqual(res, 'returned url') + + self.assertCalls( + _get('/foo/diff/master..feature'), + _post('/foo/diff/master..feature', + data={'csrf_token': 'BEEFCAFE', + 'branch_to': 'master', + 'title': 'title', + 'initial_comment': 'body'}) + ) + + def test_fail_to_get_csrf(self): + self._make_client([ + None, + ]) + + with self.assertRaises(client.PagureException): + self.client.submit_pull_request('foo', 'master', 'feature', + 'title', 'body') + + self.assertCalls( + _get('/foo/diff/master..feature'), + ) + + def test_fail_request(self): + self._make_client([ + '', + ], [ + None + ]) + + with self.assertRaises(client.PagureException): + self.client.submit_pull_request('foo', 'master', 'feature', + 'title', 'body') + + self.assertCalls( + _get('/foo/diff/master..feature'), + _post('/foo/diff/master..feature', data=mock.ANY) + ) From d5b39ee878da8f9c528acf88b2baa736bfbb8b1b Mon Sep 17 00:00:00 2001 From: Lubomír Sedlář Date: Apr 29 2018 07:45:11 +0000 Subject: [PATCH 2/2] Reduce code duplication in pag.client --- diff --git a/pag/client.py b/pag/client.py index 5344e5d..36b98a4 100644 --- a/pag/client.py +++ b/pag/client.py @@ -32,29 +32,36 @@ class Pagure(fedora.client.OpenIdBaseClient): response = self._session.get(self.base_url) return "logout" in response.text - def create(self, name, description): - if not self.is_logged_in: - raise PagureException('Not logged in.') - url = self.base_url + '/new' + def _get_csrf_token(self, url): response = self._session.get(url) if not bool(response): raise PagureException("Couldn't get form to get " "csrf token %r" % response) soup = bs4.BeautifulSoup(response.text, "html.parser") - data = dict( - csrf_token=soup.find(id='csrf_token').attrs['value'], - name=name, - description=description, - ) + return soup.find(id='csrf_token').attrs['value'] + def _post(self, url, data, action=''): response = self._session.post(url, data=data) if not bool(response): del data['csrf_token'] raise PagureException('Bad status code from pagure when ' - 'creating project: %r. Sent %r' % ( - response, data)) + '%s: %r. Sent %r' % ( + action, response, data)) + return response + + def create(self, name, description): + if not self.is_logged_in: + raise PagureException('Not logged in.') + url = self.base_url + '/new' + data = dict( + csrf_token=self._get_csrf_token(url), + name=name, + description=description, + ) + + self._post(url, data=data, action='creating project') return repo_url(name) def create_issue(self, repo, title, description, private=False): @@ -62,25 +69,14 @@ class Pagure(fedora.client.OpenIdBaseClient): raise PagureException('Not logged in.') url = self.base_url + '/' + repo + '/new_issue' - response = self._session.get(url) - if not bool(response): - raise PagureException("Couldn't get form to get " - "csrf token %r" % response) - - soup = bs4.BeautifulSoup(response.text, "html.parser") data = { - 'csrf_token': soup.find(id='csrf_token').attrs['value'], + 'csrf_token': self._get_csrf_token(url), 'title': title, 'issue_content': description, 'private': private } - response = self._session.post(url, data=data) - if not bool(response): - del data['csrf_token'] - raise PagureException('Bad status code from pagure when ' - 'forking project: %r. Sent %r' % ( - response, data)) + response = self._post(url, data=data, action='creating issue') soup = bs4.BeautifulSoup(response.text, "html.parser") response_title = soup.title.string @@ -94,14 +90,8 @@ class Pagure(fedora.client.OpenIdBaseClient): raise PagureException('Not logged in.') url = self.base_url + '/' + repo + '/upload' - response = self._session.get(url) - if not bool(response): - raise PagureException("Couldn't get form to get " - "csrf token %r" % response) - - soup = bs4.BeautifulSoup(response.text, "html.parser") data = { - 'csrf_token': soup.find(id='csrf_token').attrs['value'], + 'csrf_token': self._get_csrf_token(url), } files = { 'filestream': open(filepath, 'rb') @@ -123,50 +113,25 @@ class Pagure(fedora.client.OpenIdBaseClient): raise PagureException('Not logged in.') url = self.base_url + '/' + name - response = self._session.get(url) - if not bool(response): - raise PagureException("Couldn't get form to get " - "csrf token %r" % response) - - soup = bs4.BeautifulSoup(response.text, "html.parser") data = dict( - csrf_token=soup.find(id='csrf_token').attrs['value'], + csrf_token=self._get_csrf_token(url), ) url = self.base_url + '/do_fork/' + name - response = self._session.post(url, data=data) - - if not bool(response): - del data['csrf_token'] - raise PagureException('Bad status code from pagure when ' - 'forking project: %r. Sent %r' % ( - response, data)) + self._post(url, data=data, action='forking project') return repo_url(name) def submit_pull_request(self, name, base, head, title, comment): url = self.base_url + '/{name}/diff/{base}..{head}' url = url.format(name=name, base=base, head=head) - response = self._session.get(url) - if not bool(response): - raise PagureException("Couldn't get form to get " - "csrf token %r" % response) - - soup = bs4.BeautifulSoup(response.text, "html.parser") data = dict( - csrf_token=soup.find(id='csrf_token').attrs['value'], + csrf_token=self._get_csrf_token(url), branch_to=base, title=title, initial_comment=comment, ) - response = self._session.post(url, data=data) - - if not bool(response): - del data['csrf_token'] - raise PagureException('Bad status code from pagure when ' - 'creating pull request: %r. Sent %r' % ( - response, data)) - + response = self._post(url, data=data, action='creating pull request') return response.url