diff options
-rw-r--r-- | README.markdown | 4 | ||||
-rw-r--r-- | pylibchorus/__init__.py | 26 | ||||
-rw-r--r-- | pylibchorus/chorus_api.py | 142 | ||||
-rw-r--r-- | pylibchorus/tests/chorus_client_tests.py | 151 |
4 files changed, 92 insertions, 231 deletions
diff --git a/README.markdown b/README.markdown index 1d4d9da..32c87b6 100644 --- a/README.markdown +++ b/README.markdown @@ -20,9 +20,9 @@ usage. ['/tmp/alpine.cfg'] >>> with pylibchorus.ChorusSession(config) as session: ... print(session.sid) - ... ok, sid = pylibchorus.check_login_status(session) + ... ok, json = pylibchorus.get('/sessions', session) ... print(ok) - ... print(sid) + ... print(json['response']['session_id']) ... 49fb8e27e591d9ccb6f0006334eedb8e4d8004a5 200 diff --git a/pylibchorus/__init__.py b/pylibchorus/__init__.py index 079ae14..4c61171 100644 --- a/pylibchorus/__init__.py +++ b/pylibchorus/__init__.py @@ -2,12 +2,10 @@ '''PyLibChorus -- Python Chorus API Library''' import logging -from pylibchorus.chorus_api import login -from pylibchorus.chorus_api import logout -from pylibchorus.chorus_api import check_login_status -from pylibchorus.chorus_api import create_workfile -from pylibchorus.chorus_api import update_workfile_version -from pylibchorus.chorus_api import delete_workfile +from pylibchorus.chorus_api import get +from pylibchorus.chorus_api import post +from pylibchorus.chorus_api import put +from pylibchorus.chorus_api import delete LOG = logging.getLogger(__name__) @@ -17,26 +15,26 @@ class ChorusSession(object): def __init__(self, config): self.config = config - self.sid = None - self.cookies = None + self.sid = '' + self.cookies = {} def __enter__(self): '''create session and return sid and cookies''' LOG.debug("Opening Chorus Session") - code, json, cookies = login( - self.config.get('alpine', 'username'), - self.config.get('alpine', 'password'), - self) + data = { + 'username': self.config.get('alpine', 'username'), + 'password': self.config.get('alpine', 'password'), + } + code, json = post('/sessions', self, data) if code != 201: raise RuntimeError("Chorus Session Login Failed") self.sid = json['response']['session_id'] - self.cookies = dict(cookies) return self def __exit__(self, _type, _value, _traceback): '''Close chorus session''' LOG.debug("Closing Chorus Session") - logout(self) + delete('/sessions', self) diff --git a/pylibchorus/chorus_api.py b/pylibchorus/chorus_api.py index 8319aff..18f5096 100644 --- a/pylibchorus/chorus_api.py +++ b/pylibchorus/chorus_api.py @@ -9,55 +9,37 @@ LOG = logging.Logger(name=__name__) CONTENT_TYPE = 'application/x-www-form-urlencoded' JSON_CONTENT_TYPE = 'application/json' -def login(username, password, session): - '''POST login request to chorus server''' - return _perform_http_method_( - session.config.get('alpine', 'host'), _login_(username, password)) - -def logout(session): - '''DELETE login request to chorus server''' - return _perform_http_method_( - session.config.get('alpine', 'host'), - _logout_(session.sid, session.cookies)) - -#pylint: disable=C0103 -def check_login_status(session): - '''GET login request to chorus server''' - ok, json, _ = _perform_http_method_( +def get(url, session): + '''Perform GET request using current session''' + isok, json, cookies = _perform_http_method_( session.config.get('alpine', 'host'), - _check_login_(session.sid, session.cookies)) - return (ok, json['response']['session_id'],) + _get_(url, session.sid, session.cookies)) + session.cookies = cookies + return (isok, json,) -#pylint: disable=C0103 -def create_workfile(workspace_id, workfile_name, session): - '''POST new workfile to workspace''' - ok, json, _ = _perform_http_method_( +def post(url, session, data=None): + '''Perform POST request using current session''' + isok, json, cookies = _perform_http_method_( session.config.get('alpine', 'host'), - _create_workfile_(workspace_id, - workfile_name, - session.sid, - session.cookies)) - return (ok, json['response']['id'], json['response']['user_modified_at'],) + _post_(url, session.sid, session.cookies, data=data)) + session.cookies = cookies + return (isok, json,) -#pylint: disable=C0103 -def update_workfile_version(userid, workfile_id, workfile, session): - '''POST new workfile version''' - ok, json, _ = _perform_http_method_( +def put(url, session, data=None): + '''Perform PUT request using current session''' + isok, json, cookies = _perform_http_method_( session.config.get('alpine', 'host'), - _update_workfile_version_(userid, - workfile_id, - workfile, - session.sid, - session.cookies)) - return (ok, json['response']['id'], json['response']['user_modified_at'],) + _put_(url, session.sid, session.cookies, data=data)) + session.cookies = cookies + return (isok, json,) -#pylint: disable=C0103 -def delete_workfile(workfile_id, session): - '''DELETE workfile''' - ok, _, _ = _perform_http_method_( +def delete(url, session): + '''Perform DELETE request using current session''' + isok, json, cookies = _perform_http_method_( session.config.get('alpine', 'host'), - _delete_workfile_(workfile_id, session.sid, session.cookies)) - return ok + _delete_(url, session.sid, session.cookies)) + session.cookies = cookies + return (isok, json,) def _get_url_(host, endpoint=""): '''Return the host and path for the chorus instance''' @@ -67,6 +49,7 @@ def _perform_http_method_(host, request_data): '''Perform IO operation to Chorus Server using request_data object''' methods = {'GET': requests.get, 'POST': requests.post, + 'PUT': requests.put, 'DELETE': requests.delete,} method = methods[request_data['method']] response = method(_get_url_(host, request_data['url']), @@ -77,81 +60,42 @@ def _perform_http_method_(host, request_data): LOG.info("Request: %s status code: %d", request_data['url'], response.status_code) - return (response.status_code, response.json(), response.cookies,) + return (response.status_code, response.json(), dict(response.cookies),) -def _login_(username, password): - '''Create Request Data for ChorusSession''' +def _get_(url, sid, cookies): + '''Create GET request data''' return { - 'data': { - 'username': username, - 'password': password, - }, - 'headers': { - 'content-type': CONTENT_TYPE, - }, + 'data': None, 'params': { - 'session_id': '', + 'session_id': sid, }, - 'cookies': None, - 'url': '/sessions?session_id=', - 'method': 'POST', - } - -def _check_login_(_, cookies): - '''Create request data for check login check''' - return { - 'data': None, - 'params': None, 'headers': { 'content-type': CONTENT_TYPE, }, 'cookies': cookies, - 'url': '/sessions', + 'url': url, 'method': 'GET', } -def _logout_(sid, cookies): - '''Create request data for ChorusSession''' +def _post_(url, sid, cookies, data): + '''Create POST request data''' return { - 'data': None, - 'headers': { - 'content-type': CONTENT_TYPE, - }, + 'data': data, 'params': { 'session_id': sid, }, - 'cookies': cookies, - 'url': '/sessions', - 'method': 'DELETE', - } - -def _create_workfile_(workspace_id, workfile_name, sid, cookies): - '''Create request data for workfile creation''' - return { - 'data': { - 'workspace_id': workspace_id, - 'file_name': workfile_name, - }, 'headers': { 'content-type': CONTENT_TYPE, }, - 'params': { - 'session_id': sid, - }, 'cookies': cookies, - 'url': '/workspaces/%s/workfiles' % workspace_id, + 'url': url, 'method': 'POST', } -def _update_workfile_version_(userid, workfile_id, workfile, sid, cookies): - '''Create request data to update a workfile''' +def _put_(url, sid, cookies, data): + '''Create PUT request data''' return { - 'data': { - 'owner_id': userid, - 'modifier_id': userid, - 'commit_message': 'git commit', - 'content': workfile, - }, + 'data': data, 'params': { 'session_id': sid, }, @@ -159,12 +103,12 @@ def _update_workfile_version_(userid, workfile_id, workfile, sid, cookies): 'content-type': CONTENT_TYPE, }, 'cookies': cookies, - 'url': '/workfiles/%s/versions' % workfile_id, - 'method': 'POST', + 'url': url, + 'method': 'PUT', } -def _delete_workfile_(workfile_id, sid, cookies): - '''Create request data to delete a workfile''' +def _delete_(url, sid, cookies): + '''Create DELETE request data''' return { 'data': None, 'params': { @@ -174,6 +118,6 @@ def _delete_workfile_(workfile_id, sid, cookies): 'content-type': CONTENT_TYPE, }, 'cookies': cookies, - 'url': '/workfiles/%s' % workfile_id, + 'url': url, 'method': 'DELETE', } diff --git a/pylibchorus/tests/chorus_client_tests.py b/pylibchorus/tests/chorus_client_tests.py index 5103b70..6eb4b38 100644 --- a/pylibchorus/tests/chorus_client_tests.py +++ b/pylibchorus/tests/chorus_client_tests.py @@ -2,12 +2,10 @@ '''Chorus Client Test Cases''' import logging -from pylibchorus.chorus_api import _login_ -from pylibchorus.chorus_api import _logout_ -from pylibchorus.chorus_api import _check_login_ -from pylibchorus.chorus_api import _create_workfile_ -from pylibchorus.chorus_api import _update_workfile_version_ -from pylibchorus.chorus_api import _delete_workfile_ +from pylibchorus.chorus_api import _get_ +from pylibchorus.chorus_api import _post_ +from pylibchorus.chorus_api import _put_ +from pylibchorus.chorus_api import _delete_ import sys if sys.version_info[0] == 2 and sys.version_info[1] == 6: import unittest2 as unittest @@ -43,134 +41,55 @@ def check_params(testcase, params, expected_sid): class ChorusSessionTests(unittest.TestCase): '''ChorusSession Test Case''' - def test_login_returns_request_data(self): - '''Test _login_ returns request data''' - actual = _login_('chorusadmin', 'secret') - check_request_structure(self, actual) - self.assertIsNotNone(actual['data']) - self.assertIsNotNone(actual['headers']) - self.assertIsNotNone(actual['params']) - self.assertIsNone(actual['cookies']) - self.assertIsNotNone(actual['url']) - self.assertIsNotNone(actual['method']) - data = actual['data'] - self.assertIn('username', data) - self.assertIn('password', data) - self.assertEquals(data['username'], 'chorusadmin') - self.assertEquals(data['password'], 'secret') - check_header(self, actual['headers']) - params = actual['params'] - self.assertIn('session_id', params) - self.assertEquals(params['session_id'], '') - self.assertEquals('/sessions?session_id=', actual['url']) - self.assertEquals('POST', actual['method']) - - - #pylint: disable=C0103 - def test_logout_returns_request_data(self): - '''Test _logout_ returns correct request data''' - sid = 'foobar' - cookies = {'session_id': sid} - actual = _logout_(sid, cookies) - check_request_structure(self, actual) - self.assertIsNone(actual['data']) - self.assertIsNotNone(actual['headers']) - self.assertIsNotNone(actual['params']) - self.assertIsNotNone(actual['cookies']) - self.assertIsNotNone(actual['url']) - self.assertIsNotNone(actual['method']) - headers = actual['headers'] - self.assertIn('content-type', headers) - self.assertEquals('application/x-www-form-urlencoded', - headers['content-type']) - params = actual['params'] - self.assertIn('session_id', params) - self.assertEquals(sid, params['session_id']) - self.assertEquals(cookies, actual['cookies']) - self.assertEquals('/sessions', actual['url']) - self.assertEquals('DELETE', actual['method']) - - #pylint: disable=C0103 - def test_check_login_returns_request_data(self): - '''Test _check_login_ returns correct request data''' + def test_get_returns_request_data(self): + '''Test _get_ returns correct request data''' + url = '/' sid = 'foobar' cookies = {'session_id': sid} - actual = _check_login_(sid, cookies) + actual = _get_(url, sid, cookies) check_request_structure(self, actual) + check_params(self, actual['params'], sid) self.assertIsNone(actual['data']) - self.assertIsNotNone(actual['headers']) - self.assertIsNone(actual['params']) - self.assertIsNotNone(actual['cookies']) - self.assertIsNotNone(actual['url']) - self.assertIsNotNone(actual['method']) - check_header(self, actual['headers']) - self.assertEquals(cookies, actual['cookies']) - self.assertEquals('/sessions', actual['url']) + self.assertEquals(url, actual['url']) self.assertEquals('GET', actual['method']) - #pylint: disable=C0103 - def test_create_workfile_returns_request_data(self): - '''Test _create_workfile_ returns correct request data''' - workspace_id = 1 - workfile_name = 'foo' + def test_post_returns_request_data(self): + '''Test _post_ returns correct request data''' + url = '/workfiles/42' sid = 'foobar' cookies = {'session_id': sid} - actual = _create_workfile_(workspace_id, workfile_name, sid, cookies) + post_data = {'foo': 'bar'} + actual = _post_(url, sid, cookies, data=post_data) check_request_structure(self, actual) - self.assertIsNotNone(actual['data']) - self.assertIsNotNone(actual['headers']) - self.assertIsNotNone(actual['params']) - self.assertIsNotNone(actual['cookies']) - self.assertIsNotNone(actual['url']) - self.assertIsNotNone(actual['method']) - data = actual['data'] - self.assertIn('workspace_id', data) - self.assertIn('file_name', data) - self.assertEquals(workspace_id, data['workspace_id']) - self.assertEquals(workfile_name, data['file_name']) - check_header(self, actual['headers']) check_params(self, actual['params'], sid) - self.assertEquals(cookies, actual['cookies']) - self.assertEquals('/workspaces/%d/workfiles' % workspace_id, - actual['url']) + self.assertIsNotNone(actual['data']) + self.assertIn('foo', actual['data']) + self.assertEquals('bar', actual['data']['foo']) + self.assertEquals(url, actual['url']) self.assertEquals('POST', actual['method']) - #pylint: disable=C0103 - def test_update_workfile_returns_request_data(self): - '''Test _update_workfile_version_ returns correct request data''' - userid = 1 - workfile_id = 1 - workfile = 'some long string that looks like code, somewhere' + def test_put_returns_request_data(self): + '''Test _put_ returns correct request data''' + url = '/workfiles/42/version/0' sid = 'foobar' cookies = {'session_id': sid} - actual = _update_workfile_version_( - userid, - workfile_id, - workfile, - sid, - cookies) + put_data = {'foo': 'bar'} + actual = _put_(url, sid, cookies, data=put_data) check_request_structure(self, actual) check_params(self, actual['params'], sid) - data = actual['data'] - self.assertIn('owner_id', data) - self.assertIn('modifier_id', data) - self.assertIn('commit_message', data) - self.assertIn('content', data) - self.assertEquals(data['owner_id'], userid) - self.assertEquals(data['modifier_id'], userid) - self.assertEquals(data['commit_message'], 'git commit') - self.assertEquals(data['content'], workfile) - self.assertEquals('/workfiles/1/versions', actual['url']) - self.assertEquals('POST', actual['method']) + self.assertIsNotNone(actual['data']) + self.assertIn('foo', actual['data']) + self.assertEquals('bar', actual['data']['foo']) + self.assertEquals(url, actual['url']) + self.assertEquals('PUT', actual['method']) - def test_delete_workfile_returs_rquest_data(self): - '''Test _delete_workfile_ returns correct request data''' - workfile_id = 1 + #pylint: disable=C0103 + def test_delete_returns_request_data(self): + '''Test _delete_ returns correct request data''' + url = '/workfiles/42/versions/0' sid = 'foobar' cookies = {'session_id': sid} - actual = _delete_workfile_(workfile_id, sid, cookies) + actual = _delete_(url, sid, cookies) check_request_structure(self, actual) - check_params(self, actual['params'], sid) self.assertIsNone(actual['data']) - self.assertEquals('/workfiles/1', actual['url']) - self.assertEquals('DELETE', actual['method']) + check_params(self, actual['params'], sid) |