"Fossies" - the Fresh Open Source Software Archive  

Source code changes of the file "keystone/tests/protection/v3/test_ec2_credential.py" between
keystone-16.0.1.tar.gz and keystone-17.0.0.tar.gz

About: OpenStack Keystone (Core Service: Identity) provides an authentication and authorization service for other OpenStack services. Provides a catalog of endpoints for all OpenStack services.
The "Ussuri" series (latest release).

test_ec2_credential.py  (keystone-16.0.1):test_ec2_credential.py  (keystone-17.0.0)
skipping to change at line 13 skipping to change at line 13
# a copy of the License at # a copy of the License at
# #
# http://www.apache.org/licenses/LICENSE-2.0 # http://www.apache.org/licenses/LICENSE-2.0
# #
# Unless required by applicable law or agreed to in writing, software # Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
import http.client
from oslo_serialization import jsonutils from oslo_serialization import jsonutils
from six.moves import http_client
from keystone.common.policies import base as bp from keystone.common.policies import base as bp
from keystone.common import provider_api from keystone.common import provider_api
import keystone.conf import keystone.conf
from keystone.tests.common import auth as common_auth from keystone.tests.common import auth as common_auth
from keystone.tests import unit from keystone.tests import unit
from keystone.tests.unit import base_classes from keystone.tests.unit import base_classes
from keystone.tests.unit import ksfixtures from keystone.tests.unit import ksfixtures
from keystone.tests.unit.ksfixtures import temporaryfile from keystone.tests.unit.ksfixtures import temporaryfile
skipping to change at line 46 skipping to change at line 46
self.bootstrapper.reader_role_id, user_id=self.user_id, self.bootstrapper.reader_role_id, user_id=self.user_id,
project_id=project['id'] project_id=project['id']
) )
with self.test_client() as c: with self.test_client() as c:
r = c.post('/v3/users/%s/credentials/OS-EC2' % self.user_id, r = c.post('/v3/users/%s/credentials/OS-EC2' % self.user_id,
json={'tenant_id': project['id']}, headers=self.headers) json={'tenant_id': project['id']}, headers=self.headers)
credential_id = r.json['credential']['access'] credential_id = r.json['credential']['access']
path = '/v3/users/%s/credentials/OS-EC2/%s' % (self.user_id, credent path = '/v3/users/%s/credentials/OS-EC2/%s' % (
ial_id) self.user_id, credential_id)
r = c.get(path, headers=self.headers) r = c.get(path, headers=self.headers)
self.assertEqual( self.assertEqual(
self.user_id, r.json['credential']['user_id'] self.user_id, r.json['credential']['user_id']
) )
def test_user_can_list_their_ec2_credentials(self): def test_user_can_list_their_ec2_credentials(self):
project = unit.new_project_ref( project = unit.new_project_ref(
domain_id=CONF.identity.default_domain_id domain_id=CONF.identity.default_domain_id
) )
project = PROVIDERS.resource_api.create_project(project['id'], project) project = PROVIDERS.resource_api.create_project(project['id'], project)
skipping to change at line 86 skipping to change at line 87
) )
project = PROVIDERS.resource_api.create_project(project['id'], project) project = PROVIDERS.resource_api.create_project(project['id'], project)
PROVIDERS.assignment_api.create_grant( PROVIDERS.assignment_api.create_grant(
self.bootstrapper.reader_role_id, user_id=self.user_id, self.bootstrapper.reader_role_id, user_id=self.user_id,
project_id=project['id'] project_id=project['id']
) )
with self.test_client() as c: with self.test_client() as c:
c.post('/v3/users/%s/credentials/OS-EC2' % self.user_id, c.post('/v3/users/%s/credentials/OS-EC2' % self.user_id,
json={'tenant_id': project['id']}, headers=self.headers, json={'tenant_id': project['id']}, headers=self.headers,
expected_status_code=http_client.CREATED) expected_status_code=http.client.CREATED)
def test_user_delete_their_ec2_credentials(self): def test_user_delete_their_ec2_credentials(self):
project = unit.new_project_ref( project = unit.new_project_ref(
domain_id=CONF.identity.default_domain_id domain_id=CONF.identity.default_domain_id
) )
project = PROVIDERS.resource_api.create_project(project['id'], project) project = PROVIDERS.resource_api.create_project(project['id'], project)
PROVIDERS.assignment_api.create_grant( PROVIDERS.assignment_api.create_grant(
self.bootstrapper.reader_role_id, user_id=self.user_id, self.bootstrapper.reader_role_id, user_id=self.user_id,
project_id=project['id'] project_id=project['id']
) )
with self.test_client() as c: with self.test_client() as c:
r = c.post('/v3/users/%s/credentials/OS-EC2' % self.user_id, r = c.post('/v3/users/%s/credentials/OS-EC2' % self.user_id,
json={'tenant_id': project['id']}, headers=self.headers) json={'tenant_id': project['id']}, headers=self.headers)
credential_id = r.json['credential']['access'] credential_id = r.json['credential']['access']
c.delete('/v3/users/%s/credentials/OS-EC2/%s' % (self.user_id, crede c.delete('/v3/users/%s/credentials/OS-EC2/%s' % (
ntial_id), self.user_id, credential_id),
headers=self.headers) headers=self.headers)
def test_user_cannot_create_ec2_credentials_for_others(self): def test_user_cannot_create_ec2_credentials_for_others(self):
user = PROVIDERS.identity_api.create_user( user = PROVIDERS.identity_api.create_user(
unit.new_user_ref(domain_id=CONF.identity.default_domain_id) unit.new_user_ref(domain_id=CONF.identity.default_domain_id)
) )
project = unit.new_project_ref( project = unit.new_project_ref(
domain_id=CONF.identity.default_domain_id domain_id=CONF.identity.default_domain_id
) )
project = PROVIDERS.resource_api.create_project(project['id'], project) project = PROVIDERS.resource_api.create_project(project['id'], project)
PROVIDERS.assignment_api.create_grant( PROVIDERS.assignment_api.create_grant(
self.bootstrapper.reader_role_id, user_id=user['id'], self.bootstrapper.reader_role_id, user_id=user['id'],
project_id=project['id'] project_id=project['id']
) )
with self.test_client() as c: with self.test_client() as c:
c.post('/v3/users/%s/credentials/OS-EC2' % user['id'], c.post('/v3/users/%s/credentials/OS-EC2' % user['id'],
json={'tenant_id': project['id']}, headers=self.headers, json={'tenant_id': project['id']}, headers=self.headers,
expected_status_code=http_client.FORBIDDEN) expected_status_code=http.client.FORBIDDEN)
def test_user_cannot_delete_ec2_credentials_for_others(self): def test_user_cannot_delete_ec2_credentials_for_others(self):
user = unit.new_user_ref(domain_id=CONF.identity.default_domain_id) user = unit.new_user_ref(domain_id=CONF.identity.default_domain_id)
user_password = user['password'] user_password = user['password']
user = PROVIDERS.identity_api.create_user(user) user = PROVIDERS.identity_api.create_user(user)
project = unit.new_project_ref( project = unit.new_project_ref(
domain_id=CONF.identity.default_domain_id domain_id=CONF.identity.default_domain_id
) )
project = PROVIDERS.resource_api.create_project(project['id'], project) project = PROVIDERS.resource_api.create_project(project['id'], project)
PROVIDERS.assignment_api.create_grant( PROVIDERS.assignment_api.create_grant(
skipping to change at line 149 skipping to change at line 151
) )
with self.test_client() as c: with self.test_client() as c:
r = c.post('/v3/auth/tokens', json=user_auth) r = c.post('/v3/auth/tokens', json=user_auth)
token_id = r.headers['X-Subject-Token'] token_id = r.headers['X-Subject-Token']
headers = {'X-Auth-Token': token_id} headers = {'X-Auth-Token': token_id}
r = c.post('/v3/users/%s/credentials/OS-EC2' % user['id'], r = c.post('/v3/users/%s/credentials/OS-EC2' % user['id'],
json={'tenant_id': project['id']}, headers=headers) json={'tenant_id': project['id']}, headers=headers)
credential_id = r.json['credential']['access'] credential_id = r.json['credential']['access']
c.delete('/v3/users/%s/credentials/OS-EC2/%s' % (self.user_id, crede c.delete('/v3/users/%s/credentials/OS-EC2/%s' % (
ntial_id), self.user_id, credential_id),
headers=self.headers, expected_status_code=http_client.FORB headers=self.headers,
IDDEN) expected_status_code=http.client.FORBIDDEN)
class _SystemUserTests(object): class _SystemUserTests(object):
def test_user_can_get_ec2_credentials_for_others(self): def test_user_can_get_ec2_credentials_for_others(self):
user = unit.new_user_ref(domain_id=CONF.identity.default_domain_id) user = unit.new_user_ref(domain_id=CONF.identity.default_domain_id)
user_password = user['password'] user_password = user['password']
user = PROVIDERS.identity_api.create_user(user) user = PROVIDERS.identity_api.create_user(user)
project = unit.new_project_ref( project = unit.new_project_ref(
domain_id=CONF.identity.default_domain_id domain_id=CONF.identity.default_domain_id
) )
skipping to change at line 179 skipping to change at line 183
) )
with self.test_client() as c: with self.test_client() as c:
r = c.post('/v3/auth/tokens', json=user_auth) r = c.post('/v3/auth/tokens', json=user_auth)
token_id = r.headers['X-Subject-Token'] token_id = r.headers['X-Subject-Token']
headers = {'X-Auth-Token': token_id} headers = {'X-Auth-Token': token_id}
r = c.post('/v3/users/%s/credentials/OS-EC2' % user['id'], r = c.post('/v3/users/%s/credentials/OS-EC2' % user['id'],
json={'tenant_id': project['id']}, headers=headers) json={'tenant_id': project['id']}, headers=headers)
credential_id = r.json['credential']['access'] credential_id = r.json['credential']['access']
path = '/v3/users/%s/credentials/OS-EC2/%s' % (self.user_id, credent path = '/v3/users/%s/credentials/OS-EC2/%s' % (
ial_id) self.user_id, credential_id)
c.get(path, headers=self.headers, expected_status_code=http_client.O c.get(path, headers=self.headers,
K) expected_status_code=http.client.OK)
class _SystemReaderAndMemberTests(object): class _SystemReaderAndMemberTests(object):
def test_user_cannot_list_ec2_credentials_for_others(self): def test_user_cannot_list_ec2_credentials_for_others(self):
user = unit.new_user_ref(domain_id=CONF.identity.default_domain_id) user = unit.new_user_ref(domain_id=CONF.identity.default_domain_id)
user_password = user['password'] user_password = user['password']
user = PROVIDERS.identity_api.create_user(user) user = PROVIDERS.identity_api.create_user(user)
project = unit.new_project_ref( project = unit.new_project_ref(
domain_id=CONF.identity.default_domain_id domain_id=CONF.identity.default_domain_id
) )
skipping to change at line 374 skipping to change at line 380
) )
with self.test_client() as c: with self.test_client() as c:
r = c.post('/v3/auth/tokens', json=user_auth) r = c.post('/v3/auth/tokens', json=user_auth)
token_id = r.headers['X-Subject-Token'] token_id = r.headers['X-Subject-Token']
headers = {'X-Auth-Token': token_id} headers = {'X-Auth-Token': token_id}
r = c.post('/v3/users/%s/credentials/OS-EC2' % user['id'], r = c.post('/v3/users/%s/credentials/OS-EC2' % user['id'],
json={'tenant_id': project['id']}, headers=headers) json={'tenant_id': project['id']}, headers=headers)
credential_id = r.json['credential']['access'] credential_id = r.json['credential']['access']
c.delete('/v3/users/%s/credentials/OS-EC2/%s' % (self.user_id, crede c.delete('/v3/users/%s/credentials/OS-EC2/%s' % (
ntial_id), self.user_id, credential_id),
headers=self.headers) headers=self.headers)
class ProjectAdminTests(base_classes.TestCaseWithBootstrap, class ProjectAdminTests(base_classes.TestCaseWithBootstrap,
common_auth.AuthTestMixin, common_auth.AuthTestMixin,
_UserEC2CredentialTests, _UserEC2CredentialTests,
_SystemReaderAndMemberTests): _SystemReaderAndMemberTests):
def _override_policy(self): def _override_policy(self):
# TODO(cmurphy): Remove this once the deprecated policies in # TODO(cmurphy): Remove this once the deprecated policies in
# keystone.common.policies.ec2_credential have been removed. This is # keystone.common.policies.ec2_credential have been removed. This is
# only here to make sure we test the new policies instead of the # only here to make sure we test the new policies instead of the
# deprecated ones. Oslo.policy will OR deprecated policies with new # deprecated ones. Oslo.policy will OR deprecated policies with new
# policies to maintain compatibility and give operators a chance to # policies to maintain compatibility and give operators a chance to
# update permissions or update policies without breaking users. This # update permissions or update policies without breaking users. This
# will cause these specific tests to fail since we're trying to correct # will cause these specific tests to fail since we're trying to correct
# this broken behavior with better scope checking. # this broken behavior with better scope checking.
reader_or_cred_owner = bp.SYSTEM_READER_OR_CRED_OWNER
reader_or_owner = bp.RULE_SYSTEM_READER_OR_OWNER
admin_or_cred_owner = bp.SYSTEM_ADMIN_OR_CRED_OWNER
with open(self.policy_file_name, 'w') as f: with open(self.policy_file_name, 'w') as f:
overridden_policies = { overridden_policies = {
'identity:ec2_get_credential': bp.SYSTEM_READER_OR_CRED_OWNER, 'identity:ec2_get_credential': reader_or_cred_owner,
'identity:ec2_list_credentials': bp.RULE_SYSTEM_READER_OR_OWNER, 'identity:ec2_list_credentials': reader_or_owner,
'identity:ec2_create_credential': bp.SYSTEM_ADMIN_OR_CRED_OWNER, 'identity:ec2_create_credential': admin_or_cred_owner,
'identity:ec2_update_credential': bp.SYSTEM_ADMIN_OR_CRED_OWNER, 'identity:ec2_update_credential': admin_or_cred_owner,
'identity:ec2_delete_credential': bp.SYSTEM_ADMIN_OR_CRED_OWNER 'identity:ec2_delete_credential': admin_or_cred_owner
} }
f.write(jsonutils.dumps(overridden_policies)) f.write(jsonutils.dumps(overridden_policies))
def setUp(self): def setUp(self):
super(ProjectAdminTests, self).setUp() super(ProjectAdminTests, self).setUp()
self.loadapp() self.loadapp()
self.policy_file = self.useFixture(temporaryfile.SecureTempFile()) self.policy_file = self.useFixture(temporaryfile.SecureTempFile())
self.policy_file_name = self.policy_file.file_name self.policy_file_name = self.policy_file.file_name
self.useFixture( self.useFixture(
 End of changes. 11 change blocks. 
22 lines changed or deleted 25 lines changed or added

Home  |  About  |  Features  |  All  |  Newest  |  Dox  |  Diffs  |  RSS Feeds  |  Screenshots  |  Comments  |  Imprint  |  Privacy  |  HTTP(S)