Brian Waldon | f20e4ed | 2011-10-27 22:04:15 -0400 | [diff] [blame] | 1 | # vim: tabstop=4 shiftwidth=4 softtabstop=4 |
| 2 | |
| 3 | # Copyright 2011 OpenStack, LLC |
| 4 | # All Rights Reserved. |
| 5 | # |
| 6 | # Licensed under the Apache License, Version 2.0 (the "License"); you may |
| 7 | # not use this file except in compliance with the License. You may obtain |
| 8 | # a copy of the License at |
| 9 | # |
| 10 | # http://www.apache.org/licenses/LICENSE-2.0 |
| 11 | # |
| 12 | # Unless required by applicable law or agreed to in writing, software |
| 13 | # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT |
| 14 | # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the |
| 15 | # License for the specific language governing permissions and limitations |
| 16 | # under the License. |
| 17 | |
| 18 | """Functional test case against the OpenStack Nova API server""" |
| 19 | |
| 20 | import httplib2 |
| 21 | import json |
| 22 | import os |
| 23 | import time |
| 24 | import uuid |
| 25 | |
| 26 | from kong import keystone |
| 27 | from kong import tests |
| 28 | |
| 29 | |
| 30 | class TestKeystoneAuth(tests.FunctionalTest): |
| 31 | |
| 32 | def setUp(self): |
| 33 | super(TestKeystoneAuth, self).setUp() |
| 34 | |
| 35 | api_version = self.keystone['apiver'] |
| 36 | if api_version != 'v2.0': |
| 37 | raise ValueError("Must use Identity API v2.0") |
| 38 | |
| 39 | args = (self.keystone['service_host'], |
| 40 | self.keystone['service_port'], |
| 41 | api_version) |
| 42 | |
| 43 | self.base_url = "http://%s:%s/%s/tokens" % args |
| 44 | |
| 45 | self.user = self.keystone['user'] |
| 46 | self.password = self.keystone['password'] |
| 47 | self.tenant_id = self.keystone['tenant_id'] |
| 48 | |
| 49 | def test_can_get_token(self): |
| 50 | headers = {'content-type': 'application/json'} |
| 51 | |
| 52 | body = { |
| 53 | "auth": { |
| 54 | "passwordCredentials":{ |
| 55 | "username": self.user, |
| 56 | "password": self.password, |
| 57 | }, |
| 58 | "tenantId": self.tenant_id, |
| 59 | }, |
| 60 | } |
| 61 | |
| 62 | http = httplib2.Http() |
| 63 | response, content = http.request(self.base_url, 'POST', |
| 64 | headers=headers, |
| 65 | body=json.dumps(body)) |
| 66 | |
| 67 | self.assertEqual(response.status, 200) |
| 68 | res_body = json.loads(content) |
| 69 | self.assertTrue(res_body['access']['token']['id']) |
| 70 | test_can_get_token.tags = ['auth'] |
| 71 | |
| 72 | def test_bad_user(self): |
| 73 | headers = {'content-type': 'application/json'} |
| 74 | |
| 75 | body = { |
| 76 | "auth": { |
| 77 | "passwordCredentials": { |
| 78 | "username": str(uuid.uuid4()), |
| 79 | "password": self.password, |
| 80 | }, |
| 81 | "tenantId": self.tenant_id, |
| 82 | }, |
| 83 | } |
| 84 | |
| 85 | http = httplib2.Http() |
| 86 | response, content = http.request(self.base_url, 'POST', |
| 87 | headers=headers, |
| 88 | body=json.dumps(body)) |
| 89 | |
| 90 | self.assertEqual(response.status, 401) |
| 91 | test_bad_user.tags = ['auth'] |
| 92 | |
| 93 | def test_bad_password(self): |
| 94 | headers = {'content-type': 'application/json'} |
| 95 | |
| 96 | body = { |
| 97 | "auth": { |
| 98 | "passwordCredentials": { |
| 99 | "username": self.user, |
| 100 | "password": str(uuid.uuid4()), |
| 101 | }, |
| 102 | "tenantId": self.tenant_id, |
| 103 | }, |
| 104 | } |
| 105 | |
| 106 | http = httplib2.Http() |
| 107 | response, content = http.request(self.base_url, 'POST', |
| 108 | headers=headers, |
| 109 | body=json.dumps(body)) |
| 110 | |
| 111 | self.assertEqual(response.status, 401) |
| 112 | test_bad_password.tags = ['auth'] |
| 113 | |
| 114 | def test_bad_tenant_id(self): |
| 115 | headers = {'content-type': 'application/json'} |
| 116 | |
| 117 | body = { |
| 118 | "auth": { |
| 119 | "passwordCredentials": { |
| 120 | "username": self.user, |
| 121 | "password": self.password, |
| 122 | }, |
| 123 | "tenantId": str(uuid.uuid4()), |
| 124 | }, |
| 125 | } |
| 126 | |
| 127 | http = httplib2.Http() |
| 128 | response, content = http.request(self.base_url, 'POST', |
| 129 | headers=headers, |
| 130 | body=json.dumps(body)) |
| 131 | |
| 132 | self.assertEqual(response.status, 401) |
| 133 | test_bad_tenant_id.tags = ['auth'] |
| 134 | |
| 135 | |
| 136 | |
| 137 | class TestKeystoneAuthWithNova(tests.FunctionalTest): |
| 138 | |
| 139 | def setUp(self): |
| 140 | super(TestKeystoneAuthWithNova, self).setUp() |
| 141 | args = (self.nova['host'], self.nova['port'], |
| 142 | self.nova['ver'], self.nova['project']) |
| 143 | self.base_url = "http://%s:%s/%s/%s" % args |
| 144 | |
| 145 | self.keystone_api = keystone.API(self.keystone['service_host'], |
| 146 | self.keystone['service_port']) |
| 147 | |
| 148 | def _get_token(self): |
| 149 | user = self.keystone['user'] |
| 150 | password = self.keystone['password'] |
| 151 | tenant_id = self.keystone['tenant_id'] |
| 152 | return self.keystone_api.get_token(user, password, tenant_id) |
| 153 | |
| 154 | def test_good_token(self): |
| 155 | http = httplib2.Http() |
| 156 | url = '%s/flavors' % self.base_url |
| 157 | headers = {'x-auth-token': self._get_token()} |
| 158 | response, content = http.request(url, 'GET', headers=headers) |
| 159 | self.assertEqual(response.status, 200) |
| 160 | test_good_token.tags = ['nova', 'auth'] |
| 161 | |
| 162 | def test_bad_token(self): |
| 163 | http = httplib2.Http() |
| 164 | url = '%s/flavors' % self.base_url |
| 165 | headers = {'x-auth-token': str(uuid.uuid4())} |
| 166 | response, content = http.request(url, 'GET', headers=headers) |
| 167 | self.assertEqual(response.status, 401) |
| 168 | test_bad_token.tags = ['nova', 'auth'] |
| 169 | |
| 170 | def test_no_token(self): |
| 171 | http = httplib2.Http() |
| 172 | url = '%s/flavors' % self.base_url |
| 173 | headers = {'x-auth-token': str(uuid.uuid4())} |
| 174 | response, content = http.request(url, 'GET', headers=headers) |
| 175 | self.assertEqual(response.status, 401) |
| 176 | test_no_token.tags = ['nova', 'auth'] |
| 177 | |
| 178 | def test_no_header(self): |
| 179 | http = httplib2.Http() |
| 180 | url = '%s/flavors' % self.base_url |
| 181 | response, content = http.request(url, 'GET') |
| 182 | self.assertEqual(response.status, 401) |
| 183 | test_no_header.tags = ['nova', 'auth'] |