| #!/usr/bin/env python3 |
| """ |
| Licensed to the Apache Software Foundation (ASF) under one |
| or more contributor license agreements. See the NOTICE file |
| distributed with this work for additional information |
| regarding copyright ownership. The ASF licenses this file |
| to you 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 os |
| import platform |
| import distro |
| import sys |
| import unittest |
| import io |
| |
| from mock.mock import patch, MagicMock |
| |
| from only_for_platform import os_distro_value |
| from ambari_commons import os_utils |
| from urllib.error import HTTPError |
| |
| import shutil |
| |
| |
| # Mock classes for reading from a file |
| class MagicFile(object): |
| def __init__(self, data): |
| self.data = data |
| |
| def read(self): |
| return self.data |
| |
| def __exit__(self, exc_type, exc_val, exc_tb): |
| pass |
| |
| def __enter__(self): |
| return self |
| |
| |
| pass |
| |
| project_dir = os.path.join( |
| os.path.abspath(os.path.dirname(__file__)), os.path.normpath("../../../../") |
| ) |
| shutil.copyfile( |
| project_dir + "/ambari-server/conf/unix/ambari.properties", "/tmp/ambari.properties" |
| ) |
| |
| # We have to use this import HACK because the filename contains a dash |
| _search_file = os_utils.search_file |
| |
| |
| def search_file_proxy(filename, searchpatch, pathsep=os.pathsep): |
| global _search_file |
| if "ambari.properties" in filename: |
| return "/tmp/ambari.properties" |
| return _search_file(filename, searchpatch, pathsep) |
| |
| |
| os_utils.search_file = search_file_proxy |
| |
| with patch.object( |
| distro, |
| "linux_distribution", |
| return_value=MagicMock(return_value=("Redhat", "6.4", "Final")), |
| ): |
| with patch("os.path.isdir", return_value=MagicMock(return_value=True)): |
| with patch("os.access", return_value=MagicMock(return_value=True)): |
| with patch.object( |
| os_utils, |
| "parse_log4j_file", |
| return_value={"ambari.log.dir": "/var/log/ambari-server"}, |
| ): |
| with patch("distro.linux_distribution", return_value=os_distro_value): |
| with patch("os.symlink"): |
| with patch.object(os_utils, "is_service_exist", return_value=True): |
| with patch("glob.glob", return_value=["/etc/init.d/postgresql-9.3"]): |
| _ambari_server_ = __import__("ambari-server") |
| with patch("builtins.open"): |
| from ambari_commons.exceptions import ( |
| FatalException, |
| NonFatalException, |
| ) |
| from ambari_server.properties import Properties |
| from ambari_server.setupSso import ( |
| setup_sso, |
| AMBARI_SSO_AUTH_ENABLED, |
| SSO_PROVIDER_URL, |
| SSO_CERTIFICATE, |
| JWT_COOKIE_NAME, |
| JWT_AUDIENCES, |
| SSO_ENABLED_SERVICES, |
| SSO_MANAGE_SERVICES, |
| ) |
| |
| |
| class TestSetupSso(unittest.TestCase): |
| @patch("ambari_server.setupSso.is_server_runing") |
| def test_sso_setup_should_fail_if_server_is_not_running(self, is_server_runing_mock): |
| out = io.StringIO() |
| sys.stdout = out |
| |
| is_server_runing_mock.return_value = (False, 0) |
| options = self._create_empty_options_mock() |
| |
| try: |
| setup_sso(options) |
| self.fail("Should fail with non-fatal exception") |
| except FatalException as e: |
| self.assertTrue("Ambari Server is not running" in e.reason) |
| pass |
| |
| sys.stdout = sys.__stdout__ |
| pass |
| |
| @patch("ambari_server.setupSso.get_silent") |
| @patch("ambari_server.setupSso.is_server_runing") |
| def test_silent_mode_is_not_allowed(self, is_server_runing_mock, get_silent_mock): |
| out = io.StringIO() |
| sys.stdout = out |
| |
| is_server_runing_mock.return_value = (True, 0) |
| get_silent_mock.return_value = True |
| options = self._create_empty_options_mock() |
| |
| try: |
| setup_sso(options) |
| self.fail("Should fail with fatal exception") |
| except NonFatalException as e: |
| self.assertTrue("setup-sso is not enabled in silent mode." in e.reason) |
| pass |
| |
| sys.stdout = sys.__stdout__ |
| pass |
| |
| @patch("ambari_server.setupSso.get_silent") |
| @patch("ambari_server.setupSso.is_server_runing") |
| def test_invalid_sso_enabled_cli_option_should_result_in_error( |
| self, is_server_runing_mock, get_silent_mock |
| ): |
| out = io.StringIO() |
| sys.stdout = out |
| |
| is_server_runing_mock.return_value = (True, 0) |
| get_silent_mock.return_value = False |
| options = self._create_empty_options_mock() |
| options.sso_enabled = "not_true_or_false" |
| |
| try: |
| setup_sso(options) |
| self.fail("Should fail with fatal exception") |
| except FatalException as e: |
| self.assertTrue("--sso-enabled should be to either 'true' or 'false'" in e.reason) |
| pass |
| |
| sys.stdout = sys.__stdout__ |
| pass |
| |
| @patch("ambari_server.setupSso.get_silent") |
| @patch("ambari_server.setupSso.is_server_runing") |
| def test_missing_sso_provider_url_cli_option_when_enabling_sso_should_result_in_error( |
| self, is_server_runing_mock, get_silent_mock |
| ): |
| out = io.StringIO() |
| sys.stdout = out |
| |
| is_server_runing_mock.return_value = (True, 0) |
| get_silent_mock.return_value = False |
| options = self._create_empty_options_mock() |
| options.sso_enabled = "true" |
| options.sso_provider_url = "" |
| |
| try: |
| setup_sso(options) |
| self.fail("Should fail with fatal exception") |
| except FatalException as e: |
| self.assertTrue("Missing option: --sso-provider-url" in e.reason) |
| pass |
| |
| sys.stdout = sys.__stdout__ |
| pass |
| |
| @patch("ambari_server.setupSso.get_silent") |
| @patch("ambari_server.setupSso.is_server_runing") |
| def test_missing_sso_public_cert_file_cli_option_when_enabling_sso_should_result_in_error( |
| self, is_server_runing_mock, get_silent_mock |
| ): |
| out = io.StringIO() |
| sys.stdout = out |
| |
| is_server_runing_mock.return_value = (True, 0) |
| get_silent_mock.return_value = False |
| options = self._create_empty_options_mock() |
| options.sso_enabled = "true" |
| options.sso_public_cert_file = "" |
| |
| try: |
| setup_sso(options) |
| self.fail("Should fail with fatal exception") |
| except FatalException as e: |
| self.assertTrue("Missing option: --sso-public-cert-file" in e.reason) |
| pass |
| |
| sys.stdout = sys.__stdout__ |
| pass |
| |
| @patch("ambari_server.setupSso.get_silent") |
| @patch("ambari_server.setupSso.is_server_runing") |
| def test_invalid_sso_provider_url_cli_option_when_enabling_sso_should_result_in_error( |
| self, is_server_runing_mock, get_silent_mock |
| ): |
| out = io.StringIO() |
| sys.stdout = out |
| |
| is_server_runing_mock.return_value = (True, 0) |
| get_silent_mock.return_value = False |
| options = self._create_empty_options_mock() |
| options.sso_enabled = "true" |
| options.sso_provider_url = "!invalidHost:invalidPort" |
| |
| try: |
| setup_sso(options) |
| self.fail("Should fail with fatal exception") |
| except FatalException as e: |
| self.assertTrue("Invalid --sso-provider-url" in e.reason) |
| pass |
| |
| options.sso_provider_url = "The SSO provider URL is https://c7402.ambari.apache.org:8443/gateway/knoxsso/api/v1/websso" |
| try: |
| setup_sso(options) |
| self.fail("Should fail with fatal exception") |
| except FatalException as e: |
| self.assertTrue("Invalid --sso-provider-url" in e.reason) |
| pass |
| |
| options.sso_provider_url = "https://c7402.ambari.apache.org:8443/gateway/knoxsso/api/v1/websso is the SSO provider URL" |
| try: |
| setup_sso(options) |
| self.fail("Should fail with fatal exception") |
| except FatalException as e: |
| self.assertTrue("Invalid --sso-provider-url" in e.reason) |
| pass |
| |
| sys.stdout = sys.__stdout__ |
| pass |
| |
| @patch("ambari_server.setupSso.perform_changes_via_rest_api") |
| @patch("ambari_server.setupSso.get_ambari_properties") |
| @patch("ambari_server.setupSso.get_silent") |
| @patch("ambari_server.setupSso.is_server_runing") |
| @patch("ambari_server.setupSso.get_json_via_rest_api") |
| @patch("builtins.open") |
| def test_all_cli_options_are_collected_when_enabling_sso( |
| self, |
| open_mock, |
| get_json_via_rest_api_mock, |
| is_server_runing_mock, |
| get_silent_mock, |
| get_ambari_properties_mock, |
| perform_changes_via_rest_api_mock, |
| ): |
| out = io.StringIO() |
| sys.stdout = out |
| |
| certificate_data = ( |
| "-----BEGIN CERTIFICATE-----\n" |
| "MIIE3DCCA8SgAwIBAgIJAKfbOMmFyOlNMA0GCSqGSIb3DQEBBQUAMIGkMQswCQYD\n" |
| "................................................................\n" |
| "dXRpbmcxFzAVBgNVBAMTDmNsb3VkYnJlYWstcmdsMSUwIwYJKoZIhvcNAQkBFhZy\n" |
| "-----END CERTIFICATE-----" |
| ) |
| mock_file = MagicFile(certificate_data) |
| open_mock.side_effect = [mock_file] |
| |
| get_json_via_rest_api_mock.return_value = (200, {}) |
| is_server_runing_mock.return_value = (True, 0) |
| get_silent_mock.return_value = False |
| |
| properties = Properties() |
| get_ambari_properties_mock.return_value = properties |
| |
| sso_enabled = "true" |
| sso_enabled_services = "Ambari, SERVICE1, SERVICE2" |
| sso_provider_url = ( |
| "https://c7402.ambari.apache.org:8443/gateway/knoxsso/api/v1/websso" |
| ) |
| sso_public_cert_file = "/test/file/path" |
| sso_jwt_cookie_name = "test_cookie" |
| sso_jwt_audience_list = "test, audience, list" |
| options = self._create_empty_options_mock() |
| options.sso_enabled = sso_enabled |
| options.sso_enabled_ambari = "true" |
| options.sso_manage_services = "true" |
| options.sso_provider_url = sso_provider_url |
| options.sso_public_cert_file = sso_public_cert_file |
| options.sso_jwt_cookie_name = sso_jwt_cookie_name |
| options.sso_jwt_audience_list = sso_jwt_audience_list |
| options.sso_enabled_services = sso_enabled_services |
| |
| setup_sso(options) |
| |
| self.assertTrue(perform_changes_via_rest_api_mock.called) |
| requestCall = perform_changes_via_rest_api_mock.call_args_list[0] |
| args, kwargs = requestCall |
| requestData = args[5] |
| self.assertTrue(isinstance(requestData, dict)) |
| ssoProperties = requestData["Configuration"]["properties"] |
| self.assertEqual(ssoProperties[AMBARI_SSO_AUTH_ENABLED], sso_enabled) |
| self.assertEqual(ssoProperties[SSO_PROVIDER_URL], sso_provider_url) |
| self.assertEqual(ssoProperties[SSO_CERTIFICATE], certificate_data) |
| self.assertEqual(ssoProperties[JWT_COOKIE_NAME], sso_jwt_cookie_name) |
| self.assertEqual(ssoProperties[JWT_AUDIENCES], sso_jwt_audience_list) |
| |
| sys.stdout = sys.__stdout__ |
| pass |
| |
| @patch("ambari_server.setupSso.perform_changes_via_rest_api") |
| @patch("ambari_server.setupSso.get_ambari_properties") |
| @patch("ambari_server.setupSso.get_silent") |
| @patch("ambari_server.setupSso.is_server_runing") |
| @patch("ambari_server.setupSso.get_json_via_rest_api") |
| @patch("builtins.open") |
| def test_only_sso_enabled_cli_option_is_collected_when_disabling_sso( |
| self, |
| open_mock, |
| get_json_via_rest_api_mock, |
| is_server_runing_mock, |
| get_silent_mock, |
| get_ambari_properties_mock, |
| perform_changes_via_rest_api_mock, |
| ): |
| out = io.StringIO() |
| sys.stdout = out |
| |
| certificate_data = ( |
| "-----BEGIN CERTIFICATE-----\n" |
| "MIIE3DCCA8SgAwIBAgIJAKfbOMmFyOlNMA0GCSqGSIb3DQEBBQUAMIGkMQswCQYD\n" |
| "................................................................\n" |
| "dXRpbmcxFzAVBgNVBAMTDmNsb3VkYnJlYWstcmdsMSUwIwYJKoZIhvcNAQkBFhZy\n" |
| "-----END CERTIFICATE-----" |
| ) |
| mock_file = MagicFile(certificate_data) |
| open_mock.side_effect = [mock_file] |
| |
| get_json_via_rest_api_mock.return_value = (200, {}) |
| |
| is_server_runing_mock.return_value = (True, 0) |
| get_silent_mock.return_value = False |
| |
| properties = Properties() |
| get_ambari_properties_mock.return_value = properties |
| |
| sso_enabled = "false" |
| sso_provider_url = "http://testHost:8080" |
| sso_public_cert_file = "/test/file/path" |
| sso_jwt_cookie_name = "test_cookie" |
| sso_jwt_audience_list = "test, audience, list" |
| options = self._create_empty_options_mock() |
| options.sso_enabled = sso_enabled |
| options.sso_provider_url = sso_provider_url |
| options.sso_public_cert_file = sso_public_cert_file |
| options.sso_jwt_cookie_name = sso_jwt_cookie_name |
| options.sso_jwt_audience_list = sso_jwt_audience_list |
| |
| setup_sso(options) |
| |
| self.assertTrue(perform_changes_via_rest_api_mock.called) |
| requestCall = perform_changes_via_rest_api_mock.call_args_list[0] |
| args, kwargs = requestCall |
| requestMethod = args[4] |
| self.assertTrue(isinstance(requestMethod, str)) |
| self.assertEqual(requestMethod, "DELETE") |
| |
| sys.stdout = sys.__stdout__ |
| pass |
| |
| @patch("ambari_server.setupSso.perform_changes_via_rest_api") |
| @patch("ambari_server.setupSso.get_YN_input") |
| @patch("ambari_server.setupSso.get_ambari_properties") |
| @patch("ambari_server.setupSso.get_silent") |
| @patch("ambari_server.setupSso.is_server_runing") |
| @patch("ambari_server.setupSso.get_json_via_rest_api") |
| @patch("builtins.open") |
| def test_sso_is_enabled_for_all_services_via_user_input( |
| self, |
| open_mock, |
| get_json_via_rest_api_mock, |
| is_server_runing_mock, |
| get_silent_mock, |
| get_ambari_properties_mock, |
| get_YN_input_mock, |
| perform_changes_via_rest_api_mock, |
| ): |
| out = io.StringIO() |
| sys.stdout = out |
| |
| certificate_data = ( |
| "-----BEGIN CERTIFICATE-----\n" |
| "MIIE3DCCA8SgAwIBAgIJAKfbOMmFyOlNMA0GCSqGSIb3DQEBBQUAMIGkMQswCQYD\n" |
| "................................................................\n" |
| "dXRpbmcxFzAVBgNVBAMTDmNsb3VkYnJlYWstcmdsMSUwIwYJKoZIhvcNAQkBFhZy\n" |
| "-----END CERTIFICATE-----" |
| ) |
| mock_file = MagicFile(certificate_data) |
| open_mock.side_effect = [mock_file] |
| |
| get_json_via_rest_api_mock.return_value = (200, {}) |
| |
| is_server_runing_mock.return_value = (True, 0) |
| get_silent_mock.return_value = False |
| get_ambari_properties_mock.return_value = Properties() |
| |
| def yn_input_side_effect(*args, **kwargs): |
| if "Manage SSO configurations" in args[0]: |
| return True |
| elif "Manage SSO configurations" in args[0]: |
| return True |
| elif "all services" in args[0]: |
| return True |
| else: |
| raise Exception( |
| "ShouldNotBeInvoked" |
| ) # only the 'Use SSO for all services' question should be asked for now |
| |
| get_YN_input_mock.side_effect = yn_input_side_effect |
| |
| sso_enabled = "true" |
| sso_provider_url = "http://testHost:8080" |
| sso_public_cert_file = "/test/file/path" |
| sso_jwt_cookie_name = "test_cookie" |
| sso_jwt_audience_list = "test, audience, list" |
| |
| options = self._create_empty_options_mock() |
| options.sso_enabled = sso_enabled |
| options.sso_enabled_ambari = "true" |
| options.sso_manage_services = "true" |
| options.sso_provider_url = sso_provider_url |
| options.sso_public_cert_file = sso_public_cert_file |
| options.sso_jwt_cookie_name = sso_jwt_cookie_name |
| options.sso_jwt_audience_list = sso_jwt_audience_list |
| |
| setup_sso(options) |
| |
| self.assertTrue(perform_changes_via_rest_api_mock.called) |
| requestCall = perform_changes_via_rest_api_mock.call_args_list[0] |
| args, kwargs = requestCall |
| requestData = args[5] |
| self.assertTrue(isinstance(requestData, dict)) |
| ssoProperties = requestData["Configuration"]["properties"] |
| self.assertEqual(ssoProperties[AMBARI_SSO_AUTH_ENABLED], sso_enabled) |
| self.assertEqual(ssoProperties[SSO_PROVIDER_URL], sso_provider_url) |
| self.assertEqual(ssoProperties[SSO_CERTIFICATE], certificate_data) |
| self.assertEqual(ssoProperties[JWT_COOKIE_NAME], sso_jwt_cookie_name) |
| self.assertEqual(ssoProperties[JWT_AUDIENCES], sso_jwt_audience_list) |
| self.assertEqual(ssoProperties[SSO_MANAGE_SERVICES], "true") |
| self.assertEqual(ssoProperties[SSO_ENABLED_SERVICES], "*") |
| |
| sys.stdout = sys.__stdout__ |
| pass |
| |
| @patch("ambari_server.setupSso.perform_changes_via_rest_api") |
| @patch("urllib.request.urlopen") |
| @patch("ambari_server.setupSso.get_cluster_name") |
| @patch("ambari_server.setupSso.get_YN_input") |
| @patch("ambari_server.setupSso.get_ambari_properties") |
| @patch("ambari_server.setupSso.get_silent") |
| @patch("ambari_server.setupSso.is_server_runing") |
| @patch("builtins.open") |
| def test_setup_sso_should_not_fail_when_sso_config_cannot_be_loaded_due_to_404_error( |
| self, |
| open_mock, |
| is_server_runing_mock, |
| get_silent_mock, |
| get_ambari_properties_mock, |
| get_YN_input_mock, |
| get_cluster_name_mock, |
| urlopen_mock, |
| perform_changes_via_rest_api_mock, |
| ): |
| out = io.StringIO() |
| sys.stdout = out |
| |
| certificate_data = ( |
| "-----BEGIN CERTIFICATE-----\n" |
| "MIIE3DCCA8SgAwIBAgIJAKfbOMmFyOlNMA0GCSqGSIb3DQEBBQUAMIGkMQswCQYD\n" |
| "................................................................\n" |
| "dXRpbmcxFzAVBgNVBAMTDmNsb3VkYnJlYWstcmdsMSUwIwYJKoZIhvcNAQkBFhZy\n" |
| "-----END CERTIFICATE-----" |
| ) |
| mock_file = MagicFile(certificate_data) |
| open_mock.side_effect = [mock_file] |
| |
| is_server_runing_mock.return_value = (True, 0) |
| get_silent_mock.return_value = False |
| get_ambari_properties_mock.return_value = Properties() |
| get_cluster_name_mock.return_value = "cluster1" |
| get_YN_input_mock.__return_value = True |
| |
| urlopen_mock.side_effect = HTTPError( |
| MagicMock(status=404), 404, "not found", None, None |
| ) |
| |
| sso_enabled = "true" |
| sso_provider_url = "http://testHost:8080" |
| sso_public_cert_file = "/test/file/path" |
| sso_jwt_cookie_name = "test_cookie" |
| sso_jwt_audience_list = "test, audience, list" |
| |
| options = self._create_empty_options_mock() |
| options.sso_enabled = sso_enabled |
| options.sso_enabled_ambari = sso_enabled |
| options.sso_manage_services = "true" |
| options.sso_provider_url = sso_provider_url |
| options.sso_public_cert_file = sso_public_cert_file |
| options.sso_jwt_cookie_name = sso_jwt_cookie_name |
| options.sso_jwt_audience_list = sso_jwt_audience_list |
| |
| setup_sso(options) |
| |
| self.assertTrue(perform_changes_via_rest_api_mock.called) |
| requestCall = perform_changes_via_rest_api_mock.call_args_list[0] |
| args, kwargs = requestCall |
| requestData = args[5] |
| self.assertTrue(isinstance(requestData, dict)) |
| ssoProperties = requestData["Configuration"]["properties"] |
| self.assertEqual(ssoProperties[AMBARI_SSO_AUTH_ENABLED], sso_enabled) |
| self.assertEqual(ssoProperties[SSO_PROVIDER_URL], sso_provider_url) |
| self.assertEqual(ssoProperties[SSO_CERTIFICATE], certificate_data) |
| self.assertEqual(ssoProperties[JWT_COOKIE_NAME], sso_jwt_cookie_name) |
| self.assertEqual(ssoProperties[JWT_AUDIENCES], sso_jwt_audience_list) |
| self.assertEqual(ssoProperties[SSO_MANAGE_SERVICES], "true") |
| self.assertEqual(ssoProperties[SSO_ENABLED_SERVICES], "*") |
| |
| @patch("urllib.request.urlopen") |
| @patch("ambari_server.setupSso.perform_changes_via_rest_api") |
| @patch("ambari_server.setupSso.get_cluster_name") |
| @patch("ambari_server.setupSso.get_YN_input") |
| @patch("ambari_server.setupSso.get_ambari_properties") |
| @patch("ambari_server.setupSso.get_silent") |
| @patch("ambari_server.setupSso.is_server_runing") |
| @patch("ambari_server.setupSso.get_json_via_rest_api") |
| @patch("builtins.open") |
| def test_sso_enabled_services_are_collected_via_user_input( |
| self, |
| open_mock, |
| get_json_via_rest_api_mock, |
| is_server_runing_mock, |
| get_silent_mock, |
| get_ambari_properties_mock, |
| get_YN_input_mock, |
| get_cluster_name_mock, |
| perform_changes_via_rest_api_mock, |
| urlopen_mock, |
| ): |
| out = io.StringIO() |
| sys.stdout = out |
| |
| certificate_data = ( |
| "-----BEGIN CERTIFICATE-----\n" |
| "MIIE3DCCA8SgAwIBAgIJAKfbOMmFyOlNMA0GCSqGSIb3DQEBBQUAMIGkMQswCQYD\n" |
| "................................................................\n" |
| "dXRpbmcxFzAVBgNVBAMTDmNsb3VkYnJlYWstcmdsMSUwIwYJKoZIhvcNAQkBFhZy\n" |
| "-----END CERTIFICATE-----" |
| ) |
| mock_file = MagicFile(certificate_data) |
| open_mock.side_effect = [mock_file] |
| |
| eligible_services = """ |
| { |
| "href": "http://c7401:8080/api/v1/clusters/cluster1/services?ServiceInfo/sso_integration_supported=true", |
| "items": [ |
| { |
| "href": "http://c7401:8080/api/v1/clusters/cluster1/services/HDFS", |
| "ServiceInfo": { |
| "cluster_name": "cluster1", |
| "service_name": "HDFS", |
| "sso_integration_supported": true, |
| "sso_integration_requires_kerberos": false, |
| "kerberos_enabled": false |
| } |
| }, |
| { |
| "href": "http://c7401:8080/api/v1/clusters/cluster1/services/ZOOKEPER", |
| "ServiceInfo": { |
| "cluster_name": "cluster1", |
| "service_name": "ZOOKEPER", |
| "sso_integration_supported": true, |
| "sso_integration_requires_kerberos": false, |
| "kerberos_enabled": false |
| } |
| } |
| ] |
| } |
| """ |
| eligible_services_json = { |
| "href": "http://c7401:8080/api/v1/clusters/cluster1/services?ServiceInfo/sso_integration_supported=true", |
| "items": [ |
| { |
| "href": "http://c7401:8080/api/v1/clusters/cluster1/services/HDFS", |
| "ServiceInfo": { |
| "cluster_name": "cluster1", |
| "service_name": "HDFS", |
| "sso_integration_supported": True, |
| "sso_integration_requires_kerberos": False, |
| "kerberos_enabled": False, |
| }, |
| }, |
| { |
| "href": "http://c7401:8080/api/v1/clusters/cluster1/services/ZOOKEPER", |
| "ServiceInfo": { |
| "cluster_name": "cluster1", |
| "service_name": "ZOOKEPER", |
| "sso_integration_supported": True, |
| "sso_integration_requires_kerberos": False, |
| "kerberos_enabled": False, |
| }, |
| }, |
| ], |
| } |
| |
| get_json_via_rest_api_mock.return_value = (200, {}) |
| get_json_via_rest_api_mock.return_value = (200, eligible_services_json) |
| |
| is_server_runing_mock.return_value = (True, 0) |
| get_silent_mock.return_value = False |
| get_ambari_properties_mock.return_value = Properties() |
| get_cluster_name_mock.return_value = "cluster1" |
| |
| def yn_input_side_effect(*args, **kwargs): |
| if "all services" in args[0]: |
| return False |
| else: |
| return True |
| |
| get_YN_input_mock.side_effect = yn_input_side_effect |
| |
| response = MagicMock() |
| response.getcode.return_value = 200 |
| response.read.return_value = eligible_services |
| urlopen_mock.return_value = response |
| |
| options = self._create_empty_options_mock() |
| options.sso_enabled = "true" |
| options.sso_enabled_ambari = "true" |
| options.sso_manage_services = "true" |
| options.sso_provider_url = "http://testHost:8080" |
| options.sso_public_cert_file = "/test/file/path" |
| options.sso_jwt_cookie_name = "test_cookie" |
| options.sso_jwt_audience_list = "test, audience, list" |
| |
| setup_sso(options) |
| |
| self.assertTrue(perform_changes_via_rest_api_mock.called) |
| requestCall = perform_changes_via_rest_api_mock.call_args_list[0] |
| args, kwargs = requestCall |
| requestData = args[5] |
| self.assertTrue(isinstance(requestData, dict)) |
| ssoProperties = requestData["Configuration"]["properties"] |
| self.assertEqual(ssoProperties[SSO_MANAGE_SERVICES], "true") |
| self.assertEqual(ssoProperties[SSO_ENABLED_SERVICES], "HDFS,ZOOKEPER") |
| |
| sys.stdout = sys.__stdout__ |
| pass |
| |
| def _create_empty_options_mock(self): |
| options = MagicMock() |
| options.sso_enabled = None |
| options.sso_enabled_ambari = None |
| options.sso_manage_services = None |
| options.sso_enabled_services = None |
| options.sso_provider_url = None |
| options.sso_public_cert_file = None |
| options.sso_jwt_cookie_name = None |
| options.sso_jwt_audience_list = None |
| return options |
