• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2016 Google LLC
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7#      http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14
15import google.auth
16import google.auth.credentials
17import google.auth.jwt
18import google.auth.transport.grpc
19from google.oauth2 import service_account
20
21from google.cloud import pubsub_v1
22
23
24def test_grpc_request_with_regular_credentials(http_request):
25    credentials, project_id = google.auth.default()
26    credentials = google.auth.credentials.with_scopes_if_required(
27        credentials, scopes=["https://www.googleapis.com/auth/pubsub"]
28    )
29
30
31    # Create a pub/sub client.
32    client = pubsub_v1.PublisherClient(credentials=credentials)
33
34    # list the topics and drain the iterator to test that an authorized API
35    # call works.
36    list_topics_iter = client.list_topics(project="projects/{}".format(project_id))
37    list(list_topics_iter)
38
39
40def test_grpc_request_with_regular_credentials_and_self_signed_jwt(http_request):
41    credentials, project_id = google.auth.default()
42
43    # At the time this test is being written, there are no GAPIC libraries
44    # that will trigger the self-signed JWT flow. Manually create the self-signed
45    # jwt on the service account credential to check that the request
46    # succeeds.
47    credentials = credentials.with_scopes(
48        scopes=[], default_scopes=["https://www.googleapis.com/auth/pubsub"]
49    )
50    credentials._create_self_signed_jwt(audience="https://pubsub.googleapis.com/")
51
52    # Create a pub/sub client.
53    client = pubsub_v1.PublisherClient(credentials=credentials)
54
55    # list the topics and drain the iterator to test that an authorized API
56    # call works.
57    list_topics_iter = client.list_topics(project="projects/{}".format(project_id))
58    list(list_topics_iter)
59
60    # Check that self-signed JWT was created and is being used
61    assert credentials._jwt_credentials is not None
62    assert credentials._jwt_credentials.token == credentials.token
63
64
65def test_grpc_request_with_jwt_credentials():
66    credentials, project_id = google.auth.default()
67    audience = "https://pubsub.googleapis.com/google.pubsub.v1.Publisher"
68    credentials = google.auth.jwt.Credentials.from_signing_credentials(
69        credentials, audience=audience
70    )
71
72    # Create a pub/sub client.
73    client = pubsub_v1.PublisherClient(credentials=credentials)
74
75    # list the topics and drain the iterator to test that an authorized API
76    # call works.
77    list_topics_iter = client.list_topics(project="projects/{}".format(project_id))
78    list(list_topics_iter)
79
80
81def test_grpc_request_with_on_demand_jwt_credentials():
82    credentials, project_id = google.auth.default()
83    credentials = google.auth.jwt.OnDemandCredentials.from_signing_credentials(
84        credentials
85    )
86
87    # Create a pub/sub client.
88    client = pubsub_v1.PublisherClient(credentials=credentials)
89
90    # list the topics and drain the iterator to test that an authorized API
91    # call works.
92    list_topics_iter = client.list_topics(project="projects/{}".format(project_id))
93    list(list_topics_iter)
94