diff options
author | Michal Jagiello <michal.jagiello@t-mobile.pl> | 2020-03-09 14:34:14 +0000 |
---|---|---|
committer | Michal Jagiello <michal.jagiello@t-mobile.pl> | 2020-03-11 09:30:15 +0000 |
commit | fff342d0769e1c9f4e7900fd66b91e82c6966ac3 (patch) | |
tree | 5144bb755abdbe3a5595cc80e23d38d3bbb2da70 /ms/py-executor/resource_resolution/tests/authorization_interceptor_test.py | |
parent | 88929cf6c21023328644fc637627371733b085d8 (diff) |
PyExecutor ResourceResoluton helper class.
Create a class to call workflow execution requests to gRPC server.
Create an interceptor to use header authorization for gRPC calls.
Issue-ID: CCSDK-1989
Signed-off-by: Michal Jagiello <michal.jagiello@t-mobile.pl>
Change-Id: Ia449a089e02e7a12e31bee5e3b7debee506d8426
Diffstat (limited to 'ms/py-executor/resource_resolution/tests/authorization_interceptor_test.py')
-rw-r--r-- | ms/py-executor/resource_resolution/tests/authorization_interceptor_test.py | 50 |
1 files changed, 50 insertions, 0 deletions
diff --git a/ms/py-executor/resource_resolution/tests/authorization_interceptor_test.py b/ms/py-executor/resource_resolution/tests/authorization_interceptor_test.py new file mode 100644 index 000000000..4b03f0b36 --- /dev/null +++ b/ms/py-executor/resource_resolution/tests/authorization_interceptor_test.py @@ -0,0 +1,50 @@ +"""Copyright 2020 Deutsche Telekom. + +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. +""" + +from unittest.mock import MagicMock, _Call + +import pytest + +from resource_resolution.authorization import AuthTokenInterceptor, NewClientCallDetails + + +def test_resource_resolution_auth_token_interceptor(): + """Test AuthTokenInterceptor class. + + - Checks if it's correctly set default value. + - Checks if it's correctly set passed values. + - Checks if it's correctly checked if all header characters are lowercase. + - Checks if continuation function is called with headers setted + """ + interceptor: AuthTokenInterceptor = AuthTokenInterceptor("test_token", header="header") + assert interceptor.token == "test_token" + assert interceptor.header == "header" + + interceptor: AuthTokenInterceptor = AuthTokenInterceptor("test_token") + assert interceptor.token == "test_token" + assert interceptor.header == "authorization" + + with pytest.raises(ValueError): + AuthTokenInterceptor("test_token", header="Auth") + + continuation_mock: MagicMock = MagicMock() + client_call_details: MagicMock = MagicMock() + request_iterator: MagicMock = MagicMock() + + interceptor.intercept_stream_stream(continuation_mock, client_call_details, request_iterator) + continuation_mock.assert_called_once() + client_call_details_argument: _Call = continuation_mock.call_args_list[0][0][0] # Get NewClientCallDetails instance + assert isinstance(client_call_details_argument, NewClientCallDetails) + assert client_call_details_argument.metadata[0] == (interceptor.header, interceptor.token) |