+++ /dev/null
-from testapiclient.utils import command
-from testapiclient.utils import identity
-
-
-class Auth(command.Command):
- "Handle Authentication for users"
-
- def get_parser(self, prog_name):
- parser = super(Auth, self).get_parser(prog_name)
- return parser
-
- @identity.authenticate
- def take_action(self, parsed_args):
- print "Authentication has been successful!"
import json
from testapiclient.utils import command
-from testapiclient.utils import http_client as client
-from testapiclient.utils import identity
-from testapiclient.utils import url_parse as up
+from testapiclient.utils import urlparse
def pods_url():
- return up.resource_join('pods')
+ return urlparse.resource_join('pods')
def pod_url(parsed_args):
- return up.path_join(pods_url(), parsed_args.name)
+ return urlparse.path_join(pods_url(), parsed_args.name)
class PodGet(command.Lister):
"creation_date",
)
- data = client.get(up.query_by(pods_url(), 'name', parsed_args))
+ data = self.app.client_manager.get(
+ urlparse.query_by(pods_url(), 'name', parsed_args))
return self.format_output(columns, data.get('pods', []))
return parser
def take_action(self, parsed_args):
- return self.format_output(client.get(pod_url(parsed_args)))
+ return self.format_output(
+ self.app.client_manager.get(pod_url(parsed_args)))
class PodCreate(command.ShowOne):
'mode should be either "metal" or "virtual.')
return parser
- @identity.authenticate
def take_action(self, parsed_args):
- return self.format_output(client.post(pods_url(), parsed_args.pod))
+ return self.format_output(
+ self.app.client_manager.post(pods_url(), parsed_args.pod))
class PodDelete(command.Command):
help='Delete pods using name')
return parser
- @identity.authenticate
def take_action(self, parsed_args):
- return client.delete(pod_url(parsed_args))
+ return self.app.client_manager.delete(pod_url(parsed_args))
import json
from testapiclient.utils import command
-from testapiclient.utils import http_client as client
-from testapiclient.utils import identity
-from testapiclient.utils import url_parse as up
+from testapiclient.utils import urlparse
def projects_url():
- return up.resource_join('projects')
+ return urlparse.resource_join('projects')
def project_url(parsed_args):
- return up.path_join(projects_url(), parsed_args.name)
+ return urlparse.path_join(projects_url(), parsed_args.name)
class ProjectGet(command.Lister):
'creator',
'creation_date'
)
- data = client.get(up.query_by(projects_url(), 'name', parsed_args))
+ data = self.app.client_manager.get(
+ urlparse.query_by(projects_url(), 'name', parsed_args))
return self.format_output(columns, data.get('projects', []))
return parser
def take_action(self, parsed_args):
- return self.format_output(client.get(project_url(parsed_args)))
+ return self.format_output(
+ self.app.client_manager.get(project_url(parsed_args)))
class ProjectCreate(command.ShowOne):
'"description": (optional)""}')
return parser
- @identity.authenticate
def take_action(self, parsed_args):
return self.format_output(
- client.post(projects_url(), parsed_args.project))
+ self.app.client_manager.post(projects_url(), parsed_args.project))
class ProjectDelete(command.Command):
help='Delete project by name')
return parser
- @identity.authenticate
def take_action(self, parsed_args):
- return client.delete(project_url(parsed_args))
+ return self.app.client_manager.delete(project_url(parsed_args))
class ProjectPut(command.ShowOne):
'"description": (optional)""}')
return parser
- @identity.authenticate
def take_action(self, parsed_args):
return self.format_output(
- client.put(project_url(parsed_args), parsed_args.project))
+ self.app.client_manager.put(
+ project_url(parsed_args), parsed_args.project))
from cliff import app
from cliff import commandmanager
-import requests
-from testapiclient.utils import user
+from testapiclient.utils import clientmanager
class TestAPIClient(app.App):
command_manager=commandmanager.CommandManager('testapi'),
deferred_help=True,
)
- user.User.session = requests.Session()
+
+ def build_option_parser(self, description, version, argparse_kwargs=None):
+ self.LOG.debug('build_option_parser')
+ parser = super(TestAPIClient, self).build_option_parser(
+ description,
+ version,
+ argparse_kwargs)
+ parser.add_argument('-u',
+ type=str,
+ help='Username for authentication')
+ parser.add_argument('-p',
+ type=str,
+ help='Password for authentication')
+ return parser
def initialize_app(self, argv):
self.LOG.debug('initialize_app')
+ self.client_manager = clientmanager.ClientManager(self.options)
def prepare_to_run_command(self, cmd):
self.LOG.debug('prepare_to_run_command %s', cmd.__class__.__name__)
+ if self.options.u:
+ self.client_manager.auth()
def clean_up(self, cmd, result, err):
self.LOG.debug('clean_up %s', cmd.__class__.__name__)
def main(argv=sys.argv[1:]):
- myapp = TestAPIClient()
- return myapp.run(argv)
+ client = TestAPIClient()
+ return client.run(argv)
if __name__ == '__main__':
--- /dev/null
+import json
+import sys
+
+import requests
+import six
+import httplib
+
+
+class FakeResponse(requests.Response):
+
+ def __init__(self, headers=None, status_code=httplib.OK,
+ data=None, encoding=None):
+ super(FakeResponse, self).__init__()
+
+ headers = headers or {}
+
+ self.status_code = status_code
+
+ self.headers.update(headers)
+ if status_code != httplib.OK:
+ self.reason = data
+
+ self._content = json.dumps(data)
+ if not isinstance(self._content, six.binary_type):
+ self._content = self._content.encode()
+
+
+class FakeApp(object):
+
+ def __init__(self, _stdout, _log):
+ self.stdout = _stdout
+ self.client_manager = None
+ self.stdin = sys.stdin
+ self.stdout = _stdout or sys.stdout
+ self.stderr = sys.stderr
+ self.log = _log
+
+
+class FakeLog(object):
+
+ def __init__(self):
+ self.messages = {}
+
+ def debug(self, msg):
+ self.messages['debug'] = msg
+
+ def info(self, msg):
+ self.messages['info'] = msg
+
+ def warning(self, msg):
+ self.messages['warning'] = msg
+
+ def error(self, msg):
+ self.messages['error'] = msg
+
+ def critical(self, msg):
+ self.messages['critical'] = msg
+
+
+class FakeStdout(object):
+
+ def __init__(self):
+ self.content = []
+
+ def write(self, text):
+ self.content.append(text)
+
+ def make_string(self):
+ result = ''
+ for line in self.content:
+ result = result + line
+ return result
-import StringIO
-import httplib
import json
from mock import mock
-import requests
from six.moves.urllib import parse
import testtools
from testapiclient.cli import pods
+from testapiclient.tests.unit import fakes
from testapiclient.tests.unit import utils
-from testapiclient.utils import http_client
-from testapiclient.utils import user
+from testapiclient.utils import clientmanager
class PodTest(utils.TestCommand):
def setUp(self):
super(PodTest, self).setUp()
- user.User.session = requests.session()
self.base_url = parse.urljoin(self.api_url, 'pods')
self.pod_json = {
'role': 'community-ci',
def setUp(self):
super(PodGetTest, self).setUp()
- self.get_mock = mock.patch('requests.get').start()
- self.get_mock.return_value.status_code = httplib.OK
+ self.pods_rsp = {'pods': [self.pod_json]}
def test_get(self):
- pod_get = pods.PodGet(mock.Mock(), mock.Mock())
+ self.get_mock.return_value = fakes.FakeResponse(data=self.pods_rsp)
+ pod_get = pods.PodGet(self.app, mock.Mock())
args = ['-name', 'dfs']
verifies = [('name', 'dfs')]
parsed_args = self.check_parser(pod_get, args, verifies)
pod_get.take_action(parsed_args)
self.get_mock.assert_called_once_with(
- self.base_url + '?name=dfs')
+ self.base_url + '?name=dfs',
+ headers=clientmanager.ClientManager.headers)
def test_get_all(self):
- pod_get = pods.PodGet(mock.Mock(), mock.Mock())
+ self.get_mock.return_value = fakes.FakeResponse(data=self.pods_rsp)
+ pod_get = pods.PodGet(self.app, mock.Mock())
args = []
verifies = []
parsed_args = self.check_parser(pod_get, args, verifies)
pod_get.take_action(parsed_args)
self.get_mock.assert_called_once_with(
- self.base_url)
+ self.base_url,
+ headers=clientmanager.ClientManager.headers)
def test_get_one(self):
- pod_get_one = pods.PodGetOne(mock.Mock(), mock.Mock())
+ self.get_mock.return_value = fakes.FakeResponse(data=self.pod_json)
+ pod_get_one = pods.PodGetOne(self.app, mock.Mock())
args = ['def']
verifies = [('name', 'def')]
parsed_args = self.check_parser(pod_get_one, args, verifies)
pod_get_one.take_action(parsed_args)
self.get_mock.assert_called_once_with(
- self.base_url + '/def')
+ self.base_url + '/def',
+ headers=clientmanager.ClientManager.headers)
class PodCreateTest(PodTest):
def setUp(self):
super(PodCreateTest, self).setUp()
- self.post_mock = mock.patch(
- 'testapiclient.utils.user.User.session.post').start()
- self.post_mock.return_value.status_code = httplib.OK
+ self.succ_rsp = {
+ 'href': '{}/{}'.format(self.base_url, self.pod_json.get('name'))
+ }
def test_create_success(self):
- pod_create = pods.PodCreate(mock.Mock(), mock.Mock())
+ self.post_mock.return_value = fakes.FakeResponse(data=self.succ_rsp)
+ pod_create = pods.PodCreate(self.app, mock.Mock())
args = [self.pod_string]
verifies = [('pod', self.pod_json)]
parsed_args = self.check_parser(pod_create, args, verifies)
def test_create_failure(self):
with testtools.ExpectedException(Exception, 'Create failed: Error'):
- pod_create = pods.PodCreate(mock.Mock(), mock.Mock())
- self.post_mock.return_value.status_code = httplib.BAD_REQUEST
- self.post_mock.return_value.reason = "Error"
+ self.post_mock.return_value = utils.FAKE_FAILURE
+ pod_create = pods.PodCreate(self.app, mock.Mock())
args = [self.pod_string]
verifies = [('pod', self.pod_json)]
parsed_args = self.check_parser(pod_create, args, verifies)
pod_create.take_action(parsed_args)
- def test_create_unauthorized(self):
- self.mock_unautherized()
- with mock.patch('sys.stdout', new=StringIO.StringIO()) as mock_stdout:
- with mock.patch('requests.Session') as mock_sessions:
- mock_sessions().post.return_value.text = "login"
- pod_create = pods.PodCreate(mock.Mock(), mock.Mock())
- args = ['-u', 'user', '-p', 'password', self.pod_string]
- verifies = [
- ('u', 'user'),
- ('p', 'password'),
- ('pod', self.pod_json)]
- parsed_args = self.check_parser(pod_create, args, verifies)
- pod_create.take_action(parsed_args)
- self.assertEqual(mock_stdout.getvalue(),
- "Authentication has failed.\n")
-
- def test_create_authorized(self):
- pod_create = pods.PodCreate(mock.Mock(), mock.Mock())
- args = ['-u', 'user', '-p', 'password', self.pod_string]
- verifies = [
- ('u', 'user'),
- ('p', 'password'),
- ('pod', self.pod_json)
- ]
- parsed_args = self.check_parser(pod_create, args, verifies)
- pod_create.take_action(parsed_args)
- self.post_mock.assert_called_once()
-
class PodDeleteTest(PodTest):
def setUp(self):
super(PodDeleteTest, self).setUp()
- self.delete_mock = mock.patch(
- 'testapiclient.utils.user.User.session.delete').start()
- self.delete_mock.return_value.status_code = httplib.OK
def test_delete_success(self):
- pod_delete = pods.PodDelete(mock.Mock(), mock.Mock())
+ self.delete_mock.return_value = fakes.FakeResponse()
+ pod_delete = pods.PodDelete(self.app, mock.Mock())
args = ['def']
verifies = [('name', 'def')]
parsed_args = self.check_parser(pod_delete, args, verifies)
pod_delete.take_action(parsed_args)
self.delete_mock.assert_called_once_with(
self.base_url + '/def',
- data='null',
- headers=http_client.HTTPClient.headers)
+ data=None,
+ headers=clientmanager.ClientManager.headers)
def test_delete_failure(self):
with testtools.ExpectedException(Exception, 'Delete failed: Error'):
- pod_delete = pods.PodDelete(mock.Mock(), mock.Mock())
- self.delete_mock.return_value.status_code = httplib.FORBIDDEN
- self.delete_mock.return_value.reason = "Error"
+ self.delete_mock.return_value = utils.FAKE_FAILURE
+ pod_delete = pods.PodDelete(self.app, mock.Mock())
args = ['def']
verifies = [('name', 'def')]
parsed_args = self.check_parser(pod_delete, args, verifies)
pod_delete.take_action(parsed_args)
-
- def test_delete_authorized(self):
- pod_delete = pods.PodDelete(mock.Mock(), mock.Mock())
- args = ['-u', 'user', '-p', 'password', 'def']
- verifies = [('u', 'user'), ('p', 'password'), ('name', 'def')]
-
- parsed_args = self.check_parser(pod_delete, args, verifies)
- pod_delete.take_action(parsed_args)
- self.delete_mock.assert_called_once_with(
- self.base_url + '/def',
- data='null',
- headers=http_client.HTTPClient.headers)
-
- def test_create_unauthorized(self):
- self.mock_unautherized()
- with mock.patch('sys.stdout', new=StringIO.StringIO()) as mock_stdout:
- with mock.patch('requests.Session') as mock_sessions:
- mock_sessions().post.return_value.text = "login"
- pod_delete = pods.PodDelete(mock.Mock(), mock.Mock())
- args = ['-u', 'user', '-p', 'password', 'def']
- verifies = [('u', 'user'), ('p', 'password'), ('name', 'def')]
- parsed_args = self.check_parser(pod_delete, args, verifies)
- pod_delete.take_action(parsed_args)
- self.assertEqual(mock_stdout.getvalue(),
- "Authentication has failed.\n")
-import StringIO
-import httplib
import json
from mock import mock
-import requests
from six.moves.urllib import parse
import testtools
from testapiclient.cli import projects
+from testapiclient.tests.unit import fakes
from testapiclient.tests.unit import utils
-from testapiclient.utils import http_client
-from testapiclient.utils import user
+from testapiclient.utils import clientmanager
class ProjectTest(utils.TestCommand):
def setUp(self):
super(ProjectTest, self).setUp()
- user.User.session = requests.session()
self.base_url = parse.urljoin(self.api_url, 'projects')
self.project_json = {
'name': 'test_project',
def setUp(self):
super(ProjectGetTest, self).setUp()
- self.get_mock = mock.patch('requests.get').start()
- self.get_mock.return_value.status_code = httplib.OK
+ self.projects_rsp = {'projects': [self.project_json]}
def test_get(self):
- project_get = projects.ProjectGet(mock.Mock(), mock.Mock())
+ self.get_mock.return_value = fakes.FakeResponse(data=self.projects_rsp)
+ project_get = projects.ProjectGet(self.app, mock.Mock())
args = ['-name', 'dfs']
verifies = [('name', 'dfs')]
parsed_args = self.check_parser(project_get, args, verifies)
project_get.take_action(parsed_args)
self.get_mock.assert_called_once_with(
- self.base_url + '?name=dfs')
+ self.base_url + '?name=dfs',
+ headers=clientmanager.ClientManager.headers)
def test_get_all(self):
- project_get = projects.ProjectGet(mock.Mock(), mock.Mock())
+ self.get_mock.return_value = fakes.FakeResponse(data=self.projects_rsp)
+ project_get = projects.ProjectGet(self.app, mock.Mock())
args = []
verifies = []
parsed_args = self.check_parser(project_get, args, verifies)
project_get.take_action(parsed_args)
self.get_mock.assert_called_once_with(
- self.base_url)
+ self.base_url,
+ headers=clientmanager.ClientManager.headers)
def test_get_one(self):
- project_get_one = projects.ProjectGetOne(mock.Mock(), mock.Mock())
+ self.get_mock.return_value = fakes.FakeResponse(data=self.project_json)
+ project_get_one = projects.ProjectGetOne(self.app, mock.Mock())
args = ['def']
verifies = [('name', 'def')]
parsed_args = self.check_parser(project_get_one, args, verifies)
project_get_one.take_action(parsed_args)
self.get_mock.assert_called_once_with(
- self.base_url + '/def')
+ self.base_url + '/def',
+ headers=clientmanager.ClientManager.headers)
class ProjectCreateTest(ProjectTest):
def setUp(self):
super(ProjectCreateTest, self).setUp()
- self.post_mock = mock.patch(
- 'testapiclient.utils.user.User.session.post').start()
- self.post_mock.return_value.status_code = httplib.OK
def test_create_success(self):
- project_create = projects.ProjectCreate(mock.Mock(), mock.Mock())
+ succ_rsp = {
+ 'href': '{}/{}'.format(self.base_url,
+ self.project_json.get('name'))
+ }
+ self.post_mock.return_value = fakes.FakeResponse(data=succ_rsp)
+ project_create = projects.ProjectCreate(self.app, mock.Mock())
args = [self.project_string]
verifies = [('project', self.project_json)]
parsed_args = self.check_parser(project_create, args, verifies)
def test_create_failure(self):
with testtools.ExpectedException(Exception, 'Create failed: Error'):
- project_create = projects.ProjectCreate(mock.Mock(), mock.Mock())
- self.post_mock.return_value.status_code = httplib.BAD_REQUEST
- self.post_mock.return_value.reason = "Error"
+ self.post_mock.return_value = utils.FAKE_FAILURE
+ project_create = projects.ProjectCreate(self.app, mock.Mock())
args = [self.project_string]
verifies = [('project', self.project_json)]
parsed_args = self.check_parser(project_create, args, verifies)
project_create.take_action(parsed_args)
- def test_create_unauthorized(self):
- self.mock_unautherized()
- with mock.patch('sys.stdout', new=StringIO.StringIO()) as mock_stdout:
- with mock.patch('requests.Session') as mock_sessions:
- mock_sessions().post.return_value.text = "login"
- project_create = projects.ProjectCreate(
- mock.Mock(), mock.Mock())
- args = ['-u', 'user', '-p', 'password', self.project_string]
- verifies = [
- ('u', 'user'),
- ('p', 'password'),
- ('project', self.project_json)]
- parsed_args = self.check_parser(project_create, args, verifies)
- project_create.take_action(parsed_args)
- self.assertEqual(mock_stdout.getvalue(),
- "Authentication has failed.\n")
-
- def test_create_authorized(self):
- project_create = projects.ProjectCreate(mock.Mock(), mock.Mock())
- args = ['-u', 'user', '-p', 'password', self.project_string]
- verifies = [
- ('u', 'user'),
- ('p', 'password'),
- ('project', self.project_json)
- ]
- parsed_args = self.check_parser(project_create, args, verifies)
- project_create.take_action(parsed_args)
- self.post_mock.assert_called_once()
-
class ProjectDeleteTest(ProjectTest):
def setUp(self):
super(ProjectDeleteTest, self).setUp()
- self.delete_mock = mock.patch(
- 'testapiclient.utils.user.User.session.delete').start()
- self.delete_mock.return_value.status_code = httplib.OK
def test_delete_success(self):
- project_delete = projects.ProjectDelete(mock.Mock(), mock.Mock())
+ self.delete_mock.return_value = fakes.FakeResponse()
+ project_delete = projects.ProjectDelete(self.app, mock.Mock())
args = ['def']
verifies = [('name', 'def')]
parsed_args = self.check_parser(project_delete, args, verifies)
project_delete.take_action(parsed_args)
self.delete_mock.assert_called_once_with(
self.base_url + '/def',
- data='null',
- headers=http_client.HTTPClient.headers)
+ data=None,
+ headers=clientmanager.ClientManager.headers)
def test_delete_failure(self):
with testtools.ExpectedException(Exception, 'Delete failed: Error'):
- project_delete = projects.ProjectDelete(mock.Mock(), mock.Mock())
- self.delete_mock.return_value.status_code = httplib.FORBIDDEN
- self.delete_mock.return_value.reason = "Error"
+ self.delete_mock.return_value = utils.FAKE_FAILURE
+ project_delete = projects.ProjectDelete(self.app, mock.Mock())
args = ['def']
verifies = [('name', 'def')]
parsed_args = self.check_parser(project_delete, args, verifies)
project_delete.take_action(parsed_args)
- def test_delete_authorized(self):
- project_delete = projects.ProjectDelete(mock.Mock(), mock.Mock())
- args = ['-u', 'user', '-p', 'password', 'def']
- verifies = [('u', 'user'), ('p', 'password'), ('name', 'def')]
-
- parsed_args = self.check_parser(project_delete, args, verifies)
- project_delete.take_action(parsed_args)
- self.delete_mock.assert_called_once_with(
- self.base_url + '/def',
- data='null',
- headers=http_client.HTTPClient.headers)
-
- def test_delete_unauthorized(self):
- self.mock_unautherized()
- with mock.patch('sys.stdout', new=StringIO.StringIO()) as mock_stdout:
- with mock.patch('requests.Session') as mock_sessions:
- mock_sessions().post.return_value.text = "login"
- project_delete = projects.ProjectDelete(
- mock.Mock(), mock.Mock())
- args = ['-u', 'user', '-p', 'password', 'def']
- verifies = [('u', 'user'), ('p', 'password'), ('name', 'def')]
- parsed_args = self.check_parser(project_delete, args, verifies)
- project_delete.take_action(parsed_args)
- self.assertEqual(mock_stdout.getvalue(),
- "Authentication has failed.\n")
-
class ProjectPutTest(ProjectTest):
def setUp(self):
super(ProjectPutTest, self).setUp()
- self.put_mock = mock.patch(
- 'testapiclient.utils.user.User.session.put').start()
- self.put_mock.return_value.status_code = httplib.OK
def test_put_success(self):
- project_put = projects.ProjectPut(mock.Mock(), mock.Mock())
+ self.put_mock.return_value = fakes.FakeResponse(data=self.project_json)
+ project_put = projects.ProjectPut(self.app, mock.Mock())
args = ['def', self.project_string]
verifies = [('name', 'def'), ('project', self.project_json)]
parsed_args = self.check_parser(project_put, args, verifies)
def test_put_failure(self):
with testtools.ExpectedException(Exception, 'Update failed: Error'):
- project_put = projects.ProjectPut(mock.Mock(), mock.Mock())
- self.put_mock.return_value.status_code = httplib.BAD_REQUEST
- self.put_mock.return_value.reason = "Error"
+ self.put_mock.return_value = utils.FAKE_FAILURE
+ project_put = projects.ProjectPut(self.app, mock.Mock())
args = ['def', self.project_string]
verifies = [('name', 'def'), ('project', self.project_json)]
parsed_args = self.check_parser(project_put, args, verifies)
project_put.take_action(parsed_args)
-
- def test_put_unauthorized(self):
- self.mock_unautherized()
- with mock.patch('sys.stdout', new=StringIO.StringIO()) as mock_stdout:
- with mock.patch('requests.Session') as mock_sessions:
- mock_sessions().post.return_value.text = "login"
- project_put = projects.ProjectPut(mock.Mock(), mock.Mock())
- args = ['-u', 'user', '-p', 'password', 'def',
- self.project_string]
- verifies = [
- ('u', 'user'),
- ('p', 'password'),
- ('name', 'def'),
- ('project', self.project_json)]
- parsed_args = self.check_parser(project_put, args, verifies)
- project_put.take_action(parsed_args)
- self.assertEqual(mock_stdout.getvalue(),
- "Authentication has failed.\n")
-
- def test_create_authorized(self):
- project_put = projects.ProjectPut(mock.Mock(), mock.Mock())
- args = ['-u', 'user', '-p', 'password', 'def', self.project_string]
- verifies = [
- ('u', 'user'),
- ('p', 'password'),
- ('name', 'def'),
- ('project', self.project_json)
- ]
- parsed_args = self.check_parser(project_put, args, verifies)
- project_put.take_action(parsed_args)
- self.put_mock.assert_called_once()
-import testtools
+import httplib
+
from mock import mock
+import testtools
-from testapiclient.utils.user import User
+from testapiclient.tests.unit import fakes
+from testapiclient.utils import clientmanager
+
+FAKE_FAILURE = fakes.FakeResponse(status_code=httplib.FORBIDDEN, data='Error')
class ParserException(Exception):
}
self.config_mock = mock.patch.dict(
'os.environ', env_variables).start()
+ self.fake_stdout = fakes.FakeStdout()
+ self.fake_log = fakes.FakeLog()
+ self.app = fakes.FakeApp(self.fake_stdout, self.fake_log)
+ self.app.client_manager = clientmanager.ClientManager()
+ self.get_mock = mock.patch('requests.Session.get').start()
+ self.post_mock = mock.patch('requests.Session.post').start()
+ self.delete_mock = mock.patch('requests.Session.delete').start()
+ self.put_mock = mock.patch('requests.Session.put').start()
def check_parser(self, cmd, args, verify_args):
cmd_parser = cmd.get_parser('check_parser')
self.assertIn(attr, parsed_args)
self.assertEqual(value, getattr(parsed_args, attr))
return parsed_args
-
- def mock_unautherized(self):
- User.session = None
import httplib
import json
+import os
+import urllib
import requests
-from testapiclient.utils import user
-
-class HTTPClient(object):
-
- __instance = None
+class ClientManager(object):
headers = {'Content-type': 'application/json', 'Accept': 'text/plain'}
- @staticmethod
- def get_Instance():
- """ Static access method. """
- if HTTPClient.__instance is None:
- HTTPClient()
- return HTTPClient.__instance
-
- def __init__(self):
- """ Virtually private constructor. """
- if HTTPClient.__instance is not None:
- raise Exception("This class is a singleton!")
- else:
- HTTPClient.__instance = self
+ def __init__(self, cli_options=None):
+ self.cli_options = cli_options
+ self.session = requests.Session()
+
+ def auth(self):
+ hostname = '{}{}{}'.format(os.environ.get('testapi_cas_auth_url'),
+ urllib.quote(os.environ.get('testapi_url')),
+ os.environ.get('testapi_cas_signin_return'))
+ data = {
+ 'name': self.cli_options.u,
+ 'pass': self.cli_options.p,
+ 'form_id': 'user_login'
+ }
+ response = self.session.post(hostname, data)
+ if "login" in response.text:
+ raise Exception('Authenticate failed')
def get(self, url):
- return self._parse_response('Get', requests.get(url))
+ return self._parse_response('Get',
+ self._request('get', url,
+ headers=self.headers))
def post(self, url, data):
return self._parse_response('Create',
headers=self.headers))
def _request(self, method, *args, **kwargs):
- return getattr(user.User.session, method)(*args, **kwargs)
+ return getattr(self.session, method)(*args, **kwargs)
def _raise_failure(self, op, response):
raise Exception('{} failed: {}'.format(op, response.reason))
return response.json() if op != 'Delete' else None
else:
self._raise_failure(op, response)
-
-
-def _request(method, *args, **kwargs):
- return getattr(HTTPClient.get_Instance(), method)(*args, **kwargs)
-
-
-def get(url):
- return _request('get', url)
-
-
-def post(url, data):
- return _request('post', url, data)
-
-
-def put(url, data):
- return _request('put', url, data)
-
-
-def delete(url, data=None):
- return _request('delete', url, data)
+++ /dev/null
-import functools
-import os
-import urllib
-
-import requests
-
-from testapiclient.utils import user
-
-
-def _authenticate(username, password):
- session = requests.Session()
- hostname = '{}{}{}'.format(os.environ.get('testapi_cas_auth_url'),
- urllib.quote(os.environ.get('testapi_url')),
- os.environ.get('testapi_cas_signin_return'))
- data = {
- 'name': username,
- 'pass': password,
- 'form_id': 'user_login'
- }
- response = session.post(hostname, data)
- if "login" not in response.text:
- user.User.session = session
- return response
-
-
-def authenticate(xstep):
- @functools.wraps(xstep)
- def wrapper(self, parsed_args):
- if(user.User.session is None):
- username = parsed_args.u
- password = parsed_args.p
- if(username and password):
- response = _authenticate(username, password)
- if "login" in response.text:
- print "Authentication has failed."
- return
- return xstep(self, parsed_args)
- return wrapper
+++ /dev/null
-class User():
- session = None
Code Review / releng-testresults.git / commitdiff