This file is indexed.

/usr/lib/python2.7/dist-packages/keystoneclient/tests/unit/v3/test_endpoint_filter.py is in python-keystoneclient 1:3.2.0-4.

This file is owned by root:root, with mode 0o644.

The actual contents of the file can be viewed below.

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
# Copyright 2014 OpenStack Foundation
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.

import uuid

from keystoneclient.tests.unit.v3 import utils


class EndpointTestUtils(object):
    """Mixin class with shared methods between Endpoint Filter & Policy."""

    def new_ref(self, **kwargs):
        # copied from CrudTests as we need to create endpoint and project
        # refs for our tests. EndpointFilter is not exactly CRUD API.
        kwargs.setdefault('id', uuid.uuid4().hex)
        kwargs.setdefault('enabled', True)
        return kwargs

    def new_endpoint_ref(self, **kwargs):
        # copied from EndpointTests as we need endpoint refs for our tests
        kwargs = self.new_ref(**kwargs)
        kwargs.setdefault('interface', 'public')
        kwargs.setdefault('region', uuid.uuid4().hex)
        kwargs.setdefault('service_id', uuid.uuid4().hex)
        kwargs.setdefault('url', uuid.uuid4().hex)
        return kwargs


class EndpointFilterTests(utils.ClientTestCase, EndpointTestUtils):
    """Test project-endpoint associations (a.k.a. EndpointFilter Extension).

    Endpoint filter provides associations between service endpoints and
    projects. These assciations are then used to create ad-hoc catalogs for
    each project-scoped token request.

    """

    def setUp(self):
        super(EndpointFilterTests, self).setUp()
        self.manager = self.client.endpoint_filter

    def new_project_ref(self, **kwargs):
        # copied from ProjectTests as we need project refs for our tests
        kwargs = self.new_ref(**kwargs)
        kwargs.setdefault('domain_id', uuid.uuid4().hex)
        kwargs.setdefault('name', uuid.uuid4().hex)
        return kwargs

    def test_add_endpoint_to_project_via_id(self):
        endpoint_id = uuid.uuid4().hex
        project_id = uuid.uuid4().hex

        self.stub_url('PUT',
                      [self.manager.OS_EP_FILTER_EXT, 'projects', project_id,
                       'endpoints', endpoint_id],
                      status_code=201)

        self.manager.add_endpoint_to_project(project=project_id,
                                             endpoint=endpoint_id)

    def test_add_endpoint_to_project_via_obj(self):
        project_ref = self.new_project_ref()
        endpoint_ref = self.new_endpoint_ref()
        project = self.client.projects.resource_class(self.client.projects,
                                                      project_ref,
                                                      loaded=True)
        endpoint = self.client.endpoints.resource_class(self.client.endpoints,
                                                        endpoint_ref,
                                                        loaded=True)

        self.stub_url('PUT',
                      [self.manager.OS_EP_FILTER_EXT,
                       'projects', project_ref['id'],
                       'endpoints', endpoint_ref['id']],
                      status_code=201)

        self.manager.add_endpoint_to_project(project=project,
                                             endpoint=endpoint)

    def test_delete_endpoint_from_project(self):
        endpoint_id = uuid.uuid4().hex
        project_id = uuid.uuid4().hex

        self.stub_url('DELETE',
                      [self.manager.OS_EP_FILTER_EXT, 'projects', project_id,
                       'endpoints', endpoint_id],
                      status_code=201)

        self.manager.delete_endpoint_from_project(project=project_id,
                                                  endpoint=endpoint_id)

    def test_check_endpoint_in_project(self):
        endpoint_id = uuid.uuid4().hex
        project_id = uuid.uuid4().hex

        self.stub_url('HEAD',
                      [self.manager.OS_EP_FILTER_EXT, 'projects', project_id,
                       'endpoints', endpoint_id],
                      status_code=201)

        self.manager.check_endpoint_in_project(project=project_id,
                                               endpoint=endpoint_id)

    def test_list_endpoints_for_project(self):
        project_id = uuid.uuid4().hex
        endpoints = {'endpoints': [self.new_endpoint_ref(),
                                   self.new_endpoint_ref()]}
        self.stub_url('GET',
                      [self.manager.OS_EP_FILTER_EXT, 'projects', project_id,
                       'endpoints'],
                      json=endpoints,
                      status_code=200)

        endpoints_resp = self.manager.list_endpoints_for_project(
            project=project_id)

        expected_endpoint_ids = [
            endpoint['id'] for endpoint in endpoints['endpoints']]
        actual_endpoint_ids = [endpoint.id for endpoint in endpoints_resp]
        self.assertEqual(expected_endpoint_ids, actual_endpoint_ids)

    def test_list_projects_for_endpoint(self):
        endpoint_id = uuid.uuid4().hex
        projects = {'projects': [self.new_project_ref(),
                                 self.new_project_ref()]}
        self.stub_url('GET',
                      [self.manager.OS_EP_FILTER_EXT, 'endpoints', endpoint_id,
                       'projects'],
                      json=projects,
                      status_code=200)

        projects_resp = self.manager.list_projects_for_endpoint(
            endpoint=endpoint_id)

        expected_project_ids = [
            project['id'] for project in projects['projects']]
        actual_project_ids = [project.id for project in projects_resp]
        self.assertEqual(expected_project_ids, actual_project_ids)