from io import StringIO import json import mock from orm.orm_client.ormcli import ormcli from orm.orm_client.ormcli import rmscli import requests import sys from unittest import TestCase TJ = {'access': {'token': {'id': 'test'}}} 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.ranger_agent_version = '4' args.clli = '5' args.regionname = '6' args.osversion = '7' args.location_type = '8' args.domain_name = 'default' args.state = '9' args.country = '10' args.city = '11' args.street = '12' args.zip = '13' args.clcp_name = '14' list_region_url = '/?type=%s&status=%s&metadata=%s'\ '&ranger_agent_version=%s&clli=%s®ionname=%s'\ '&osversion=%s&location_type=%s&domain_name=%s'\ '&state=%s'\ '&country=%s&city=%s&street=%s&zip=%s&vlcp_name=%s' 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, list_region_url % (args.type, args.status, args.metadata, args.ranger_agent_version, args.clli, args.regionname, args.osversion, args.location_type, args.domain_name, args.state, args.country, args.city, args.street, args.zip, args.clcp_name)) } for subcmd in subcmd_to_result: args.subcmd = subcmd self.assertEqual(subcmd_to_result[subcmd], rmscli.cmd_details(args)) @mock.patch.object(rmscli, 'validate_args') @mock.patch.object(rmscli.requests, 'post') @mock.patch.object(rmscli.requests, 'get') @mock.patch.object(rmscli.cli_common, 'get_token') @mock.patch.object(rmscli, 'globals') def test_list_regions(self, mock_globals, mock_get_token, mock_get, mock_post, mock_validate_args): mock_post.return_value = self.respond(TJ, 200) mock_get.return_value = self.mock_response args = ormcli.main('orm rms list_regions t'.split()) sys.stdout.seek(0) output = sys.stdout.read() self.assertIn(json.dumps(TJ), output) @mock.patch.object(rmscli, 'validate_args') @mock.patch.object(rmscli.requests, 'post') @mock.patch.object(rmscli.requests, 'get') @mock.patch.object(rmscli.cli_common, 'get_token') @mock.patch.object(rmscli, 'globals') def test_list_regions_a(self, mock_globals, mock_get_token, mock_get, mock_post, mock_validate_args): mock_post.return_value = self.respond(TJ, 200) mock_get.return_value = self.mock_response mock_get.__name__ = 'a' args = ormcli.main('orm rms --verbose list_regions t'.split()) sys.stdout.seek(0) output = sys.stdout.read() self.assertIn(json.dumps(TJ), output) @mock.patch.object(rmscli, 'validate_args') @mock.patch.object(rmscli.requests, 'post') @mock.patch.object(rmscli.requests, 'get') def test_list_regions_e(self, mock_get, mock_post, mock_validate_args): mock_post.return_value = self.respond(TJ, 200) mock_get.side_effect = Exception('e') with self.assertRaises(SystemExit) as cm: args = ormcli.main('orm rms list_regions t'.split()) self.assertEqual(cm.exception.code, 1) sys.stdout.seek(0) output = sys.stdout.read() self.assertIn('e', output) @mock.patch('requests.get') @mock.patch('requests.post') def test_list_regions_with_filters(self, mock_post, mock_get): cli = ormcli.Cli() cli.create_parser() cli.parse( 'orm rms list_regions --city StLouis --zip 63101 client1'.split()) resp = self.respond('{"Howdy, mate"}', 200, {'X-Subject-Token': 989}) mock_post.return_value = self.respond( {"access": {"token": {"id": 989}}}, 200) @mock.patch.object(rmscli, 'validate_args') @mock.patch.object(rmscli.requests, 'post') @mock.patch.object(rmscli.requests, 'get') @mock.patch.object(rmscli.cli_common, 'get_token') @mock.patch.object(rmscli, 'globals') def test_list_regions_errors(self, mock_globals, mock_get_token, mock_get, mock_post, mock_validate_args): mock_post.return_value = self.respond(TJ, 200) mock_get.return_value = self.respond(TJ, 204) mock_get.__name__ = 'test' with self.assertRaises(SystemExit) as cm: args = ormcli.main('orm rms list_regions 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) with self.assertRaises(SystemExit) as cm: args = ormcli.main('orm rms --faceless list_regions t'.split()) self.assertEqual(cm.exception.code, 1) sys.stdout.seek(0) output = sys.stdout.read() self.assertIn('API error:', output) @mock.patch('requests.get') def test_one_zone(self, mock_get): cli = ormcli.Cli() cli.create_parser() cli.parse( 'orm rms --faceless --rms-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", "ranger_agent_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('"ranger_agent_version": "aic3.0"', output)