x/ranger
Code Issues Proposed changes

Move ormcli tests to top level tests directory

Browse Source 
Moves ormcli/tests to orm/tests/unit/ormcli

Change-Id: Ie1c44de2e27c8938fa560d79b85eae3885e65b64
This commit is contained in:
Nicholas Jones
2017-09-15 14:26:14 -05:00
parent f28a9d276a
commit 2645bf293f
8 changed files with 51 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
orm/orm_client/ormcli/tests/__init__.py
View File

119
orm/orm_client/ormcli/tests/data/cms-add-cust.json
View File

@@ -1,119 +0,0 @@
{
"description": "Customer description",
"enabled": true,
"name": "myDomain",
"metadata": {
"my_server_name": "Apache1",
"ocx_cust": "123456889"
},
"regions": [
{
"name": "SAN1",
"type": "single",
"quotas": [
{
"compute": [
{
"instances": "10",
"injected-files": "10",
"keypairs": "10",
"ram": "10"
}
],
"storage": [
{
"gigabytes": "10",
"snapshots": "10",
"volumes": "10"
}
],
"network": [
{
"floatingip": "10",
"network": "10",
"port": "10",
"router": "10",
"subnet": "10"
}
]
}
]
},
{
"name": "AIC_MEDIUM",
"type": "group",
"quotas": [
{
"compute": [
{
"instances": "10",
"injected-files": "10",
"keypairs": "10",
"ram": "10"
}
],
"storage": [
{
"gigabytes": "10",
"snapshots": "10",
"volumes": "10"
}
],
"network": [
{
"floatingip": "10",
"network": "10",
"port": "10",
"router": "10",
"subnet": "10"
}
]
}
]
}
],
"users": [
{
"id": "userId1",
"role": [
"admin",
"other"
]
},
{
"id": "userId2",
"role": [
"storage"
]
}
],
"defaultQuotas": [
{
"compute": [
{
"instances": "10",
"injected-files": "10",
"keypairs": "10",
"ram": "10"
}
],
"storage": [
{
"gigabytes": "10",
"snapshots": "10",
"volumes": "10"
}
],
"network": [
{
"floatingip": "10",
"network": "10",
"port": "10",
"router": "10",
"subnet": "10"
}
]
}
]
}

29
orm/orm_client/ormcli/tests/data/ims-create-image.json
View File

@@ -1,29 +0,0 @@
{
"name": "abc1",
"url": "https://mirrors.it.att.com/images/image-name",
"visibility": "private",
"disk-format": "raw",
"container-format": "bare",
"min-disk": 1,
"min-ram": 1,
"tags": ["tag1", "tag2", "tag3"],
"properties": {
"property1": "value1",
"property2": "value2",
"property3": "value3"
},
"regions": [
{
"name": "rdm1",
"type": "single"
},
{
"name": "rdm3",
"type": "single"
}
],
"customers": [
"alef",
"bet"
]
}

53
orm/orm_client/ormcli/tests/test_cli_common.py
View File

@@ -1,53 +0,0 @@
import mock
from ormcli import cli_common
from unittest import TestCase
class CmsTests(TestCase):
@mock.patch.object(cli_common.requests, 'get')
def test_get_keystone_ep_sanity(self, mock_get):
my_response = mock.MagicMock()
my_response.status_code = cli_common.OK_CODE
my_response.json.return_value = {
'regions': [{'endpoints': [{
'type': 'identity', 'publicURL': 'test'}]}]}
mock_get.return_value = my_response
self.assertEqual(cli_common.get_keystone_ep('a', 'b'), 'test')
@mock.patch.object(cli_common.requests, 'get',
side_effect=cli_common.requests.exceptions
.ConnectionError())
def test_get_keystone_ep_connection_failed(self, mock_get):
self.assertIsNone(cli_common.get_keystone_ep('a', 'b'))
@mock.patch.object(cli_common.requests, 'get')
def test_get_keystone_ep_bad_status_code(self, mock_get):
my_response = mock.MagicMock()
my_response.status_code = cli_common.OK_CODE + 1
my_response.json.return_value = {
'regions': [{'endpoints': [{
'type': 'identity', 'publicURL': 'test'}]}]}
mock_get.return_value = my_response
self.assertIsNone(cli_common.get_keystone_ep('a', 'b'))
@mock.patch.object(cli_common.requests, 'get')
def test_get_keystone_ep_bad_response(self, mock_get):
my_response = mock.MagicMock()
my_response.status_code = cli_common.OK_CODE
my_response.json.return_value = {
'regions': [{'endpoinqs': [{
'type': 'identity', 'publicURL': 'test'}]}]}
mock_get.return_value = my_response
self.assertIsNone(cli_common.get_keystone_ep('a', 'b'))
my_response.json.return_value = {
'regions': [{'endpoints': [{
'type': 'identiqy', 'publicURL': 'test'}]}]}
mock_get.return_value = my_response
self.assertIsNone(cli_common.get_keystone_ep('a', 'b'))

229
orm/orm_client/ormcli/tests/test_cmscli.py
View File

@@ -1,229 +0,0 @@
from cStringIO import StringIO
import json
import mock
from ormcli import cmscli
from ormcli import ormcli
import requests
import sys
from unittest import TestCase
TJ = {'access': {'token': {'id': 'test'}}}
class CmsTests(TestCase):
def setUp(self):
out, sys.stdout, err, sys.stderr = sys.stdout, StringIO(), sys.stderr, StringIO()
self.mock_response = mock.Mock()
def respond(self, value, code, headers={}, oy=False):
# Set the response according to the parameter
if oy:
response = mock.Mock()
else:
response = self.mock_response
response.json.return_value = value
response.status_code = code
response.headers = headers
return response
def test_cmd_details(self):
# Set up the args parameter
args = mock.MagicMock()
args.custid = 'test_custid'
args.regionid = 'test_region'
args.userid = 'test_userid'
args.region = 'test_region'
args.user = 'test_user'
args.starts_with = 'test_startswith'
args.contains = 'test_contains'
subcmd_to_result = {
'create_customer': (requests.post, '',),
'delete_customer': (requests.delete, '/%s' % args.custid,),
'update_customer': (requests.put, '/%s' % args.custid,),
'add_region': (requests.post, '/%s/regions' % args.custid,),
'replace_region': (requests.put, '/%s/regions' % args.custid,),
'delete_region': (
requests.delete,
'/%s/regions/%s' % (args.custid, args.regionid),),
'add_user': (
requests.post,
'/%s/regions/%s/users' % (args.custid, args.regionid),),
'replace_user': (
requests.put,
'/%s/regions/%s/users' % (args.custid, args.regionid),),
'delete_user': (requests.delete, '/%s/regions/%s/users/%s' % (
args.custid, args.regionid, args.userid),),
'add_default_user': (requests.post, '/%s/users' % args.custid,),
'replace_default_user': (requests.put, '/%s/users' % args.custid,),
'delete_default_user': (
requests.delete, '/%s/users/%s' % (args.custid, args.userid),),
'add_metadata': (requests.post, '/%s/metadata' % args.custid,),
'replace_metadata': (requests.put, '/%s/metadata' % args.custid,),
'get_customer': (requests.get, '/%s' % args.custid,),
'list_customers': (requests.get,
'/?region=%s&user=%s&starts_with=%s'
'&contains=%s' % (args.region,
args.user, args.starts_with,
args.contains))
}
# Assert that each subcommand returns the expected details
for subcmd in subcmd_to_result:
args.subcmd = subcmd
self.assertEqual(subcmd_to_result[subcmd],
cmscli.cmd_details(args))
@mock.patch.object(cmscli.cli_common, 'get_keystone_ep',
return_value=None)
def test_get_token_keystone_ep_not_found(self, mock_get_keystone_ep):
args = mock.MagicMock()
args.username = 'test'
self.assertRaises(cmscli.ConnectionError, cmscli.get_token,
'a', args, 'c')
@mock.patch.object(cmscli.cli_common, 'get_keystone_ep')
@mock.patch.object(cmscli.requests, 'post')
def test_get_token_errors(self, mock_post, mock_get_keystone_ep):
# Bad status code
my_response = mock.MagicMock()
my_response.status_code = 201
mock_post.return_value = my_response
self.assertRaises(cmscli.ConnectionError, cmscli.get_token,
3, mock.MagicMock(), 'c')
# Post fails
mock_post.side_effect = ValueError('test')
self.assertRaises(cmscli.ConnectionError, cmscli.get_token,
3, mock.MagicMock(), 'c')
@mock.patch.object(cmscli.cli_common, 'get_keystone_ep')
@mock.patch.object(cmscli.requests, 'post')
@mock.patch.object(cmscli.requests, 'get')
def test_list_customers(self, mock_get, mock_post, mock_get_keystone_ep):
mock_post.return_value = self.respond(TJ, 200)
mock_get.return_value = self.mock_response
args = ormcli.main('orm cms list_customers t'.split())
sys.stdout.seek(0)
output = sys.stdout.read()
self.assertIn(json.dumps(TJ), output)
@mock.patch.object(cmscli.cli_common, 'get_keystone_ep')
@mock.patch.object(cmscli.requests, 'post')
@mock.patch.object(cmscli.requests, 'get')
@mock.patch.object(cmscli, 'get_token')
def test_list_customers_a(self, mock_get_token,
mock_get, mock_post, mock_get_keystone_ep):
mock_post.return_value = self.respond(TJ, 200)
mock_get.return_value = self.mock_response
mock_get.__name__ = 'a'
args = ormcli.main('orm cms --verbose list_customers t'.split())
sys.stdout.seek(0)
output = sys.stdout.read()
self.assertIn(json.dumps(TJ), output)
@mock.patch.object(cmscli.cli_common, 'get_keystone_ep')
@mock.patch.object(cmscli.requests, 'post')
@mock.patch.object(cmscli.requests, 'get')
def test_list_customers_e(self, mock_get, mock_post, mock_get_keystone_ep):
mock_post.return_value = self.respond(TJ, 200)
mock_get.side_effect = Exception('e')
with self.assertRaises(SystemExit) as cm:
args = ormcli.main('orm cms list_customers t'.split())
self.assertEqual(cm.exception.code, 1)
sys.stdout.seek(0)
output = sys.stdout.read()
self.assertIn('e', output)
@mock.patch.object(cmscli.cli_common, 'get_keystone_ep')
@mock.patch.object(cmscli.requests, 'post')
@mock.patch.object(cmscli.requests, 'get')
@mock.patch.object(cmscli, 'get_token')
@mock.patch.object(cmscli, 'globals')
def test_list_customers_errors(self, mock_globals, mock_get_token,
mock_get, mock_post,
mock_get_keystone_ep):
mock_post.return_value = self.respond(TJ, 200)
mock_get.return_value = self.respond(TJ, 204, oy=True)
with self.assertRaises(SystemExit) as cm:
args = ormcli.main('orm cms list_customers t'.split())
self.assertEqual(cm.exception.code, 0)
sys.stdout.seek(0)
output = sys.stdout.read()
self.assertEqual('', output)
def test_parsing(self):
cli = ormcli.Cli()
cli.create_parser()
cli.parse(
'orm cms --orm-base-url 12.11.10.9 --port 8832 --timeout 150 '
'add_user '
'client1 customer1 region1 '
'ormcli/tests/data/cms-add-cust.json'.split())
args = cli.args
self.assertEqual(args.orm_base_url, '12.11.10.9')
self.assertEqual(args.port, 8832)
self.assertEqual(args.timeout, 150)
@mock.patch('requests.post')
def test_timeout(self, mock_post):
mock_post.side_effect = Exception("timeout boom")
cli = ormcli.Cli()
cli.create_parser()
cli.parse(
'orm cms --faceless add_user client1 customer1 region1 '
'ormcli/tests/data/cms-add-cust.json'.split())
with self.assertRaises(SystemExit) as cm:
cli.logic()
self.assertEqual(cm.exception.code, 1)
sys.stdout.seek(0)
output = sys.stdout.read()
self.assertIn('timeout boom', output)
@mock.patch('requests.post')
@mock.patch.object(cmscli, 'get_token')
def test_no_keystone(self, mock_get_token, mock_post):
mock_post.side_effect = Exception("timeout boom")
cli = ormcli.Cli()
cli.create_parser()
cli.parse(
'orm cms add_user client1 customer1 region1 '
'ormcli/tests/data/cms-add-cust.json'.split())
with self.assertRaises(SystemExit) as cm:
cli.logic()
@mock.patch('requests.post')
@mock.patch.object(cmscli, 'get_token')
def test_response_code(self, mock_get_token, mock_post):
cli = ormcli.Cli()
cli.create_parser()
cli.parse(
'orm cms create_customer client1 '
'ormcli/tests/data/cms-add-cust.json'.split())
resp = self.respond({"access": {"token": {"id": 989}}}, 400)
mock_post.return_value = resp
with self.assertRaises(SystemExit) as cm:
cli.logic()
@mock.patch('requests.post')
def test_ok(self, mock_post):
cli = ormcli.Cli()
cli.create_parser()
cli.parse(
'orm cms create_customer client1 '
'ormcli/tests/data/cms-add-cust.json'.split())
mock_post.return_value = self.respond(
{"access": {"token": {"id": 989}}}, 200)
@mock.patch('requests.get')
@mock.patch('requests.post')
def test_list_customers(self, mock_post, mock_get):
cli = ormcli.Cli()
cli.create_parser()
cli.parse(
'orm cms list_customers --region 2 --user bob client1'.split())
resp = self.respond('{"Hi, mom"}', 200, {'X-Subject-Token': 989})
mock_post.return_value = self.respond(
{"access": {"token": {"id": 989}}}, 200)

172
orm/orm_client/ormcli/tests/test_fmscli.py
View File

@@ -1,172 +0,0 @@
from cStringIO import StringIO
import json
import mock
from ormcli import fmscli
from ormcli import ormcli
import requests
import sys
from unittest import TestCase
TJ = {'access': {'token': {'id': 'test'}}}
class FmsTests(TestCase):
def setUp(self):
out, sys.stdout, err, sys.stderr = sys.stdout, StringIO(), sys.stderr, StringIO()
self.mock_response = mock.Mock()
def respond(self, value, code, headers={}, oy=False):
# Set the response according to the parameter
if oy:
response = mock.Mock()
else:
response = self.mock_response
response.json.return_value = value
response.status_code = code
response.headers = headers
return response
def test_cmd_details(self):
# Set up the args parameter
args = mock.MagicMock()
args.flavorid = 'test_flavorid'
args.regionid = 'test_region'
args.region = 'test_region'
args.tagname = 'test_tagname'
args.eskeyname = 'test_eskeyname'
args.visibility = 'test_visibility'
args.tenant = 'test_tenant'
args.series = 'test_series'
args.starts_with = 'test_startswith'
args.contains = 'test_contains'
args.alias = 'test_alias'
list_flavors_url = '/?visibility=%s&region=%s&tenant=%s&series=%s' \
'&starts_with=%s&contains=%s&alias=%s'
subcmd_to_result = {
'create_flavor': (requests.post, '',),
'delete_flavor': (requests.delete, '/%s' % args.flavorid,),
'add_region': (requests.post, '/%s/regions' % args.flavorid,),
'add_tags': (requests.post, '/%s/tags' % args.flavorid,),
'replace_tags': (requests.put, '/%s/tags' % args.flavorid,),
'delete_tag': (
requests.delete,
'/%s/tags/%s' % (args.flavorid, args.tagname),),
'delete_all_tags': (requests.delete, '/%s/tags' % args.flavorid,),
'get_tags': (requests.get, '/%s/tags' % args.flavorid,),
'delete_region': (requests.delete, '/%s/regions/%s' % (
args.flavorid, args.regionid),),
'add_tenant': (requests.post, '/%s/tenants' % args.flavorid,),
'delete_tenant': (requests.delete, '/%s/tenants/%s' % (
args.flavorid, args.tenantid),),
'get_flavor': (requests.get, '/%s' % args.flavorid,),
'get_extra_specs': (
requests.get, '/%s/os_extra_specs' % args.flavorid,),
'delete_all_extra_specs': (
requests.delete, '/%s/os_extra_specs' % args.flavorid,),
'delete_extra_spec': (requests.delete, '/%s/os_extra_specs/%s' % (
args.flavorid, args.eskeyname),),
'add_extra_specs': (
requests.post, '/%s/os_extra_specs' % args.flavorid,),
'list_flavors': (requests.get,
list_flavors_url % (args.visibility, args.region,
args.tenant, args.series,
args.starts_with,
args.contains, args.alias))
}
# Assert that each subcommand returns the expected details
for subcmd in subcmd_to_result:
args.subcmd = subcmd
self.assertEqual(subcmd_to_result[subcmd],
fmscli.cmd_details(args))
@mock.patch.object(fmscli.cli_common, 'get_keystone_ep',
return_value=None)
def test_get_token_keystone_ep_not_found(self, mock_get_keystone_ep):
args = mock.MagicMock()
args.username = 'test'
self.assertRaises(fmscli.ConnectionError, fmscli.get_token,
'a', args, 'c')
@mock.patch.object(fmscli.cli_common, 'get_keystone_ep')
@mock.patch.object(fmscli.requests, 'post')
def test_get_token_errors(self, mock_post, mock_get_keystone_ep):
# Bad status code
my_response = mock.MagicMock()
my_response.status_code = 201
mock_post.return_value = my_response
self.assertRaises(fmscli.ConnectionError, fmscli.get_token,
3, mock.MagicMock(), 'c')
# Post fails
mock_post.side_effect = ValueError('test')
self.assertRaises(fmscli.ConnectionError, fmscli.get_token,
3, mock.MagicMock(), 'c')
@mock.patch.object(fmscli.cli_common, 'get_keystone_ep')
@mock.patch.object(fmscli.requests, 'post')
@mock.patch.object(fmscli.requests, 'get')
@mock.patch.object(fmscli, 'get_token')
@mock.patch.object(fmscli, 'globals')
def test_list_flavors(self, mock_globals, mock_get_token,
mock_get, mock_post, mock_get_keystone_ep):
mock_post.return_value = self.respond(TJ, 200)
mock_get.return_value = self.mock_response
args = ormcli.main('orm fms list_flavors t'.split())
sys.stdout.seek(0)
output = sys.stdout.read()
self.assertIn(json.dumps(TJ), output)
@mock.patch.object(fmscli.cli_common, 'get_keystone_ep')
@mock.patch.object(fmscli.requests, 'post')
@mock.patch.object(fmscli.requests, 'get')
@mock.patch.object(fmscli, 'get_token')
@mock.patch.object(fmscli, 'globals')
def test_list_flavors_a(self, mock_globals, mock_get_token,
mock_get, mock_post, mock_get_keystone_ep):
mock_post.return_value = self.respond(TJ, 200)
mock_get.return_value = self.mock_response
mock_get.__name__ = 'a'
args = ormcli.main('orm fms --verbose list_flavors t'.split())
sys.stdout.seek(0)
output = sys.stdout.read()
self.assertIn(json.dumps(TJ), output)
@mock.patch.object(fmscli.cli_common, 'get_keystone_ep')
@mock.patch.object(fmscli.requests, 'post')
@mock.patch.object(fmscli.requests, 'get')
def test_list_flavors_e(self, mock_get, mock_post, mock_get_keystone_ep):
mock_post.return_value = self.respond(TJ, 200)
mock_get.side_effect = Exception('e')
with self.assertRaises(SystemExit) as cm:
args = ormcli.main('orm fms list_flavors t'.split())
self.assertEqual(cm.exception.code, 1)
sys.stdout.seek(0)
output = sys.stdout.read()
self.assertIn('e', output)
@mock.patch.object(fmscli.cli_common, 'get_keystone_ep')
@mock.patch.object(fmscli.requests, 'post')
@mock.patch.object(fmscli.requests, 'get')
@mock.patch.object(fmscli, 'get_token')
@mock.patch.object(fmscli, 'globals')
def test_list_flavors_errors(self, mock_globals, mock_get_token,
mock_get, mock_post,
mock_get_keystone_ep):
mock_post.return_value = self.respond(TJ, 200)
mock_get.return_value = self.respond(TJ, 204, oy=True)
with self.assertRaises(SystemExit) as cm:
args = ormcli.main('orm fms list_flavors t'.split())
self.assertEqual(cm.exception.code, 0)
sys.stdout.seek(0)
output = sys.stdout.read()
self.assertEqual('', output)
mock_get.return_value = self.respond(TJ, 404, oy=True)
with self.assertRaises(SystemExit) as cm:
args = ormcli.main('orm fms --faceless list_flavors t'.split())
self.assertEqual(cm.exception.code, 1)
sys.stdout.seek(0)
output = sys.stdout.read()
self.assertIn('API error:', output)

199
orm/orm_client/ormcli/tests/test_imscli.py
View File

@@ -1,199 +0,0 @@
from cStringIO import StringIO
import mock
from ormcli import imscli
from ormcli.imscli import cmd_data
from ormcli import ormcli
import sys
from unittest import TestCase
class ImsTests(TestCase):
def setUp(self):
out, sys.stdout, err, sys.stderr = sys.stdout, StringIO(), sys.stderr, StringIO()
self.mock_response = mock.Mock()
def respond(self, value, code, headers={}):
self.mock_response.json.return_value = value
self.mock_response.status_code = code
self.mock_response.headers = headers
return self.mock_response
def test_error_with_empty_args(self):
with self.assertRaises(SystemExit) as cm:
args = ormcli.main([])
self.assertEqual(cm.exception.code, 2)
sys.stderr.seek(0)
output = sys.stderr.read()
self.assertIn('too few arguments', output)
def test_help_command(self):
with self.assertRaises(SystemExit) as cm:
args = ormcli.main('orm --help'.split())
self.assertEqual(cm.exception.code, 0)
sys.stdout.seek(0)
output = sys.stdout.read()
self.assertIn('usage:', output)
self.assertIn('optional arguments:', output)
self.assertIn('<service>', output)
self.assertIn('ims', output)
self.assertIn('Image Management', output)
def test_ims_help_command(self):
with self.assertRaises(SystemExit) as cm:
args = ormcli.main('orm ims --help'.split())
self.assertEqual(cm.exception.code, 0)
sys.stdout.seek(0)
output = sys.stdout.read()
self.assertIn('usage:', output)
self.assertIn('timeout', output)
self.assertIn('optional arguments:', output)
self.assertIn('orm ims', output)
@mock.patch.object(imscli, 'cli_common')
@mock.patch('requests.put')
@mock.patch('requests.post')
@mock.patch.object(imscli, 'get_token')
@mock.patch.object(imscli, 'globals')
def test_timeout(self, mock_globals, mock_get_token,
mock_post, mock_put, mock_common):
mock_post.side_effect = Exception("timeout boom")
cli = ormcli.Cli()
cli.create_parser()
cli.parse(
'orm ims create_image client1 '
'ormcli/tests/data/ims-create-image.json'.split())
with self.assertRaises(SystemExit) as cm:
cli.logic()
self.assertEqual(cm.exception.code, 1)
sys.stdout.seek(0)
output = sys.stdout.read()
self.assertIn('timeout boom', output)
@mock.patch('requests.post')
@mock.patch.object(imscli, 'get_token')
@mock.patch.object(imscli, 'globals')
def test_no_keystone(self, mock_globals, mock_get_token, mock_post):
mock_post.side_effect = Exception("timeout boom")
cli = ormcli.Cli()
cli.create_parser()
globals()['auth_region'] = 'test'
cli.parse(
'orm ims create_image client1 '
'ormcli/tests/data/ims-create-image.json'.split())
with self.assertRaises(SystemExit) as cm:
cli.logic()
# @mock.patch.object(imscli, 'cli_common')
# @mock.patch('requests.post')
# def test_response_code(self, mock_post, mock_common):
# cli = ormcli.Cli()
# cli.create_parser()
# cli.parse(
# 'orm ims create_image client1 '
# 'ormcli/tests/data/ims-create-image.json'.split())
# resp = self.respond({"access": {"token": {"id": 989}}}, 200)
# mock_post.return_value = resp
# cli.logic()
# sys.stdout.seek(0)
# output = sys.stdout.read()
# # The response json should be printed, but since we mock post() only
# # once, the response would be the same response received
# # from Keystone
# self.assertIn('{\'access\': {\'token\': {\'id\': 989}}}', output)
# @mock.patch.object(imscli, 'cli_common')
# @mock.patch('requests.get')
# @mock.patch('requests.post')
# def test_get_image_sanity(self, mock_post, mock_get, mock_common):
# mock_post.return_value = self.respond(
# {"access": {"token": {"id": 989}}}, 201)
# cli = ormcli.Cli()
# cli.create_parser()
# cli.parse('orm ims get_image client1 test'.split())
# mock_get.return_value = self.respond(
# {"access": {"token": {"id": 989}}}, 200)
# cli.logic()
# sys.stdout.seek(0)
# output = sys.stdout.read()
# self.assertIn('{\'access\': {\'token\': {\'id\': 989}}}', output)
# @mock.patch.object(imscli, 'cli_common')
# @mock.patch('requests.get')
# @mock.patch('requests.post')
# def test_list_images(self, mock_post, mock_get, mock_common):
# mock_post.return_value = self.respond(
# {"access": {"token": {"id": 989}}}, 201)
# cli = ormcli.Cli()
# cli.create_parser()
# cli.parse(
# 'orm ims list_images client1 --visibility public --region a '
# '--tenant b'.split())
# resp = self.respond({"access": {"token": {"id": 989}}}, 200)
# mock_get.return_value = self.respond(
# {"access": {"token": {"id": 989}}}, 200)
# cli.logic()
# sys.stdout.seek(0)
# output = sys.stdout.read()
# self.assertIn('{\'access\': {\'token\': {\'id\': 989}}}', output)
@mock.patch.object(imscli, 'cli_common')
@mock.patch('requests.get')
@mock.patch('requests.post')
@mock.patch.object(imscli, 'get_token')
@mock.patch.object(imscli, 'globals')
def test_list_images_bad_request(self, mock_get_token, mock_globals,
mock_post, mock_get, mock_common):
mock_post.return_value = self.respond(
{"access": {"token": {"id": 989}}}, 201)
cli = ormcli.Cli()
cli.create_parser()
cli.parse(
'orm ims list_images client1 --visibility public --region a '
'--customer b'.split())
resp = self.respond({"access": {"token": {"id": 989}}}, 200)
with self.assertRaises(SystemExit) as cm:
cli.logic()
self.assertEqual(cm.exception.code, 1)
sys.stdout.seek(0)
output = sys.stdout.read()
self.assertIn('API error', output)
def test_cmd_data_enable(self):
my_args = mock.MagicMock()
my_args.subcmd = 'enable'
cm_data = cmd_data(my_args)
self.assertEqual("{\n \"enabled\": true\n}", cm_data)
def test_cmd_data_disable(self):
my_args = mock.MagicMock()
my_args.subcmd = 'disable'
cm_data = cmd_data(my_args)
self.assertEqual("{\n \"enabled\": false\n}", cm_data)
def test_cmd_data_no_data_file(self):
my_args = mock.MagicMock()
my_args.subcmd = 'xyz'
my_args.datafile.read.return_value = "123"
cm_data = cmd_data(my_args)
self.assertEqual("{}", cm_data)
def test_cmd_data_from_data_file(self):
my_args = MyDataFile()
cm_data = cmd_data(my_args)
self.assertEqual("123", cm_data)
class MyDataFile(object):
def __init__(self):
self.subcmd = '1'
self.datafile = FakeDataFIle()
def __iter__(self):
return iter(['datafile'])
class FakeDataFIle(object):
def read(self):
return '123'

188
orm/orm_client/ormcli/tests/test_rmscli.py
View File

@@ -1,188 +0,0 @@
from cStringIO import StringIO
import mock
from ormcli import ormcli
from ormcli import rmscli
import requests
import sys
from unittest import TestCase
class RmsTests(TestCase):
def setUp(self):
out, sys.stdout, err, sys.stderr = sys.stdout, StringIO(), sys.stderr, StringIO()
self.mock_response = mock.Mock()
def respond(self, value, code, headers={}):
self.mock_response.json.return_value = value
self.mock_response.status_code = code
self.mock_response.headers = headers
return self.mock_response
def test_cmd_details(self):
args = mock.MagicMock()
args.get_group = 'test_get_group'
args.list_groups = 'test_list_groups'
args.create_group = 'test_create_group'
args.update_group = 'test_update_group'
args.region_name_or_id = 'test_region_name_or_id'
args.type = '1'
args.status = '2'
args.metadata = '3'
args.aicversion = '4'
args.clli = '5'
args.regionname = '6'
args.osversion = '7'
args.valet = '8'
args.state = '9'
args.country = '10'
args.city = '11'
args.street = '12'
args.zip = '13'
args.vlcp_name = '14'
AAAAAAA = '/?type=%s&status=%s&metadata=%s&aicversion=%s&clli=%s'\
'&regionname=%s&osversion=%s&valet=%s&state=%s&country=%s'\
'&city=%s&street=%s&zip=%s&vlcp_name=%s' % (args.type,
args.status,
args.metadata,
args.aicversion,
args.clli,
args.regionname,
args.osversion,
args.valet,
args.state,
args.country,
args.city,
args.street,
args.zip,
args.vlcp_name,)
subcmd_to_result = {'get_region': (requests.get,
'/%s' % args.region_name_or_id),
'get_group': (requests.get, '/%s' % args.group_id),
'list_groups': (requests.get, '/'),
'create_group': (requests.post, '/'),
'update_group': (
requests.put, '/%s' % args.group_id),
'list_regions': (requests.get, AAAAAAA)
}
for subcmd in subcmd_to_result:
args.subcmd = subcmd
self.assertEqual(subcmd_to_result[subcmd],
rmscli.cmd_details(args))
def test_parsing(self):
cli = ormcli.Cli()
cli.create_parser()
cli.parse(
'orm rms --orm-base-url 12.11.10.9 --port 8832 --timeout 150 '
'list_regions --type big '.split())
args = cli.args
self.assertEqual(args.orm_base_url, '12.11.10.9')
self.assertEqual(args.port, 8832)
self.assertEqual(args.type, 'big')
self.assertEqual(args.timeout, 150)
@mock.patch('requests.get')
def test_timeout(self, mock_get):
cli = ormcli.Cli()
cli.create_parser()
cli.parse(
'orm rms --faceless --orm-base-url 12.11.10.9 --port 8832'
' --timeout 1 get_region x'.split())
mock_get.side_effect = Exception("timeout boom")
with self.assertRaises(SystemExit) as cm:
cli.logic()
self.assertEqual(cm.exception.code, 1)
sys.stdout.seek(0)
output = sys.stdout.read()
self.assertIn('timeout boom', output)
@mock.patch('requests.get')
def test_one_zone(self, mock_get):
cli = ormcli.Cli()
cli.create_parser()
cli.parse(
'orm rms --faceless --orm-base-url 12.11.10.9 --port 8832'
' --timeout 150 get_region zoneone'.split())
resp = self.respond(
{
"clli": "n/a",
"name": "SNA 1",
"enabled": 1,
"state": "functional",
"aic_version": "aic3.0",
"endpoints": [
{
"type": "horizon",
"publicurl": "http://horizon1.com"
},
{
"type": "identity",
"publicurl": "http://identity1.com"
},
{
"type": "ord",
"publicurl": "http://ord1.com"
}
],
"id": "SNA1",
"metadata": []
}, 200,
{'X-Subject-Token': 989})
mock_get.return_value = resp
cli.logic()
sys.stdout.seek(0)
output = sys.stdout.read()
self.assertIn('"aic_version": "aic3.0"', output)
# def test_error_with_wrong_port(self):
# args = self.parser.parse_args('--port 1111'.split())
# with self.assertRaises(SystemExit) as cm:
# rmscli.rmscli_logic(args)
# self.assertEqual(cm.exception.code, 1)
# sys.stdout.seek(0)
# output = sys.stdout.read()
# self.assertIn('Connection refused', output)
# def test_help_command(self):
# with self.assertRaises(SystemExit) as cm:
# args = self.parser.parse_args(['--help'])
# self.assertEqual(cm.exception.code, 0)
# sys.stdout.seek(0)
# output = sys.stdout.read()
# self.assertIn('usage:', output)
# self.assertIn('timeout', output)
# self.assertIn('optional arguments:', output)
# self.assertIn('--host', output)
# @mock.patch('requests.get')
# def test_timeout(self, mock_get):
# args = self.parser.parse_args('--host 1.1.1.1 --timeout
# 1000'.split())
# mock_get.side_effect = Exception("HTTPConnectionPool(
# host='1.1.1.1', port=8080): Max retries exceeded with url: /lcp (
# Caused by ConnectTimeoutError(
# <requests.packages.urllib3.connection.HTTPConnection object at
# 0x7f9469c1a310>, 'Connection to 1.1.1.1 timed out. (connect
# timeout=1.0)'))")
# with self.assertRaises(SystemExit) as cm:
# rmscli.rmscli_logic(args)
# self.assertEqual(cm.exception.code, 1)
# sys.stdout.seek(0)
# output = sys.stdout.read()
# self.assertIn('ConnectTimeoutError', output)
# learn how to mock 'real' request.get
# @mock.patch('rmscli.rmscli.rmscli.requests.get', autospec=True)
# def test_bad_status(self, mock_get):
# args = self.parser.parse_args([])
# mock_get.return_value = Response({},500)
# with self.assertRaises(SystemExit) as cm:
# rmscli.rmscli_logic(args)
# self.assertEqual(cm.exception.code, 1)
# sys.stdout.seek(0)
# output = sys.stdout.read()
# self.assertIn('GET', output)