PR#39: Add tests for pag.client and refactor it - pag

pag

#39 Add tests for pag.client and refactor it

Merged 6 years ago by hguemar. Opened 6 years ago by lsedlar.

lsedlar/pag test-client into develop

Reduce code duplication in pag.client

Lubomír Sedlář • 6 years ago

d5b39ee

Add tests for pag.client module

Lubomír Sedlář • 6 years ago

a1f4b00

pag/client.py

file modified

+27 -61

		`@@ -6,6 +6,7 @@`

		`from pag.utils import repo_url`

		`+`
		`class PagureException(Exception):`
		`pass`

		`@@ -31,29 +32,36 @@`
		`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):`
		`@@ -61,25 +69,14 @@`
		`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`
		`@@ -93,14 +90,8 @@`
		`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')`
		`@@ -122,50 +113,25 @@`
		`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 = '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)`
		`- 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`

test/test_client.py

file added

+385

		`@@ -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',`
		`+ '<html><input id="csrf_token" value="BEEFCAFE"></html>',`
		`+ ], [`
		`+ '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',`
		`+ '<html><input id="csrf_token" value="BEEFCAFE"></html>',`
		`+ ], [`
		`+ 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',`
		`+ '<html><input id="csrf_token" value="BEEFCAFE"></html>',`
		`+ ], [`
		`+ '<html><title>Issue #123: title</title></html>',`
		`+ ])`
		`+`
		`+ 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',`
		`+ '<html><input id="csrf_token" value="BEEFCAFE"></html>',`
		`+ ], [`
		`+ 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',`
		`+ '<html><input id="csrf_token" value="BEEFCAFE"></html>',`
		`+ ], [`
		`+ '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',`
		`+ '<html><input id="csrf_token" value="BEEFCAFE"></html>',`
		`+ ], [`
		`+ 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',`
		`+ '<html><input id="csrf_token" value="BEEFCAFE"></html>',`
		`+ ], [`
		`+ '<html><span class="alert alert-info">OK</span></html>',`
		`+ ])`
		`+`
		`+ 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',`
		`+ '<html><input id="csrf_token" value="BEEFCAFE"></html>',`
		`+ ], [`
		`+ '<html><span class="alert alert-error">Boom</span></html>',`
		`+ ])`
		`+`
		`+ 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([`
		`+ '<html><input id="csrf_token" value="BEEFCAFE"></html>',`
		`+ ], [`
		`+ ('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([`
		`+ '<html><input id="csrf_token" value="BEEFCAFE"></html>',`
		`+ ], [`
		`+ 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)`
		`+ )`