text: # --------------------------------------------------------------------------------------------
  # Copyright (c) Microsoft Corporation. All rights reserved.
  # Licensed under the MIT License. See License.txt in the project root for license information.
  # --------------------------------------------------------------------------------------------
  
  import unittest
  import base64
  from azure.cli.core.azclierror import InvalidArgumentValueError, MutuallyExclusiveArgumentError
  from azext_k8s_configuration.custom import validate_and_get_protected_settings, validate_url_with_params, validate_known_hosts
  import azext_k8s_configuration._validators as validators
  from Crypto.PublicKey import RSA, ECC, DSA
  from paramiko.ed25519key import Ed25519Key
  
  
  class TestValidateKeyTypes(unittest.TestCase):
      def test_bad_private_key(self):
          private_key_encoded = base64.b64encode("this is not a valid private key".encode('utf-8')).decode('utf-8')
          err = "Error! ssh private key provided in invalid format"
          with self.assertRaises(InvalidArgumentValueError) as cm:
              validate_and_get_protected_settings(private_key_encoded, '', '', '')
          self.assertEqual(str(cm.exception), err)
  
      def test_rsa_private_key(self):
          rsa_key = "LS0tLS1CRUdJTiBPUEVOU1NIIFBSSVZBVEUgS0VZLS0tLS0KYjNCbGJuTnphQzFyWlhrdGRqRUFBQUFBQkc1dmJtVUFBQUFFYm05dVpRQUFBQUFBQUFBQkFBQUJsd0FBQUFkemMyZ3RjbgpOaEFBQUFBd0VBQVFBQUFZRUF1bVA5M09qRHdjdlEyZHZhRlJNNWYrMEhVSnFvOFJnbmdwaGN3NFZidnd1TVNoQTZFc2FyCjFsam1CNUNnT1NGNHJqNDIvcmdxMW1hWndoSUgvckdPSElNa0lIcjFrZmNKMnBrR3ZhK1NxVm4wWUhzMjBpUW02ay92ZXQKdXdVQ2J1QjlxSU5zL2h2b0ppQ21JMUVpVWZ4VGoxRFJCUG15OXR3Qm52bW5FS1kxZ2NhT2YrS2Y1aGhCc09pd00yZnBRTwp0aTlIcHVzM1JhNXpFeElWbjJzVitpRjVvV3ZZM1JQTTlKNXFPMXRObUtOWll6TjgzbDYxMlBzRmR1Vm1QM2NUUlJtK2pzCjdzZW5jY0U0RitzU0hQMlJpMk5DU0JvZ2RJOFR5VTlzeTM3Szl3bFJ5NGZkWWI1K1o3YUZjMjhTNDdDWlo5dTRFVXdWUEYKbjU4dTUzajU0empwdXNpei9ZWmx3MG5NeEQ5SXI0aHlJZ2s0NlUzVmdHR0NPUytZVTVZT2JURGhPRG5udk5VRkg2NVhCagpEM3l6WVJuRDA3b2swQ1JUR3RCOWMzTjBFNDBjUnlPeVpEQ0l5a0FPdHZXYnBUZzdnaXA2UDc4K2pLVlFnanFwRTVQdi9ICnl1dlB6cUJoUkpWcG5VR1dvWnFlcWJhd2N5RWZwdHFLaTNtWUdVMHBBQUFGa0U5cUs3SlBhaXV5QUFBQUIzTnphQzF5YzIKRUFBQUdCQUxwai9kem93OEhMME5uYjJoVVRPWC90QjFDYXFQRVlKNEtZWE1PRlc3OExqRW9RT2hMR3E5Wlk1Z2VRb0RraAplSzQrTnY2NEt0Wm1tY0lTQi82eGpoeURKQ0I2OVpIM0NkcVpCcjJ2a3FsWjlHQjdOdElrSnVwUDczcmJzRkFtN2dmYWlECmJQNGI2Q1lncGlOUklsSDhVNDlRMFFUNXN2YmNBWjc1cHhDbU5ZSEdqbi9pbitZWVFiRG9zRE5uNlVEcll2UjZick4wV3UKY3hNU0ZaOXJGZm9oZWFGcjJOMFR6UFNlYWp0YlRaaWpXV016Zk41ZXRkajdCWGJsWmo5M0UwVVp2bzdPN0hwM0hCT0JmcgpFaHo5a1l0alFrZ2FJSFNQRThsUGJNdCt5dmNKVWN1SDNXRytmbWUyaFhOdkV1T3dtV2ZidUJGTUZUeForZkx1ZDQrZU00CjZicklzLzJHWmNOSnpNUS9TSytJY2lJSk9PbE4xWUJoZ2prdm1GT1dEbTB3NFRnNTU3elZCUit1VndZdzk4czJFWnc5TzYKSk5Ba1V4clFmWE56ZEJPTkhFY2pzbVF3aU1wQURyYjFtNlU0TzRJcWVqKy9Qb3lsVUlJNnFST1Q3L3g4cnJ6ODZnWVVTVgphWjFCbHFHYW5xbTJzSE1oSDZiYWlvdDVtQmxOS1FBQUFBTUJBQUVBQUFHQkFMaElmSXFacUZKSFRXcllyN24rays4alR3ClFtcGJvWmc1YmZSWGdhdGljaEo4ZGlXOGlNblFFRVRBcFd0OU5FZ0tqbDRrSGRuSnoyUERkZzFIN0ExaHppbkNsdzZMTTAKYUkyMGxyR2NrWWpXNDRNd3ozYmRQNHlURTllSXRiM0pmN1pNSGpqek4rSy96bWN0eWdMeXFZSzVXYTljM1JnMXdIRWFNNAplakUvNDg4M25WUmJvSFJDcjFCVi8wQVVFTTZhNisrRHpVZW9WdWdWL3RsV3RVMlJuQlZ4eCtJS0FVSDZRTHJFU2JkUkRoCkVGUEFhRWtEb3crd3dDcFpqTXBhMHdRZXBDSkhwWkJLN1pBU25EU3R3Y2RKRE4yeHZzdVNOOGg0bkN0MlZWd0xRenJKeVAKU2VjcWM3M1hIc3E3VWx6ZU5veHlTVW9KZ2JjNTZoRzhWYS9ITlhsOUtkdkFlWUVzS1l1OW5NRUprVSt3VHo1KzUvM2wwVQpxSkErb0pTVTducjYydlVKQnljbXg0SFdBcjJ6QkR2QnFBUWMzRG9LWHczeVM1Z0c5Zkc0c25OUUkxOHVRSjdOSjdndHZHClpKRU56bTNJMmFTMzl5dndWZnFIMXpXVERxU2VNeWhYeWFnTkFEcGtCVEJIMVJQR2NtTFplclFmWWx1djVVUmFNTXdRQUEKQU1BdE9oNHFwUUhidm5tQ1RVakx4dXRrWnRaRlhNa0hmSTk5NS9Nd2RvWVY1eWRKV0pUVGsyKzB1QVBIcTZEejk2b3dWbQpjUkF2WDBDOVU5d3ZRMkpnR0Y1MDZzcmgzZkVpUzM2d1ArOFd0RjZ6ODd0enJwQnpQVHIxOGRONURCOEx5L3dXRk5BVTdqClBUbXM0dHlUY1VsRXR3eEt4TXJTNC9ROUZwMWozL3JNdnNZdGVaSVgycmN4YUhkWWJDVGJtTUpZS3lVTWVXTk56NXpub1EKcFcyd2NDSmpJc1MvS1F2WmR4cHZwNWd0RXE1WlEva3FvLzJVRWd1NHhwdDNWeUNma0FBQURCQVBOSHVEU1R0ZEpJWjdzcwpaQkVwcUE4TE54b1dMQ2RURnlpRERiUnpYOWVPTldkRFQ3NklaRE9HejczNXJhZUFSM2FiY0FhaUM0dDQwTFJTNGEyN29sCm9wK1dSak9wcjVNYUtOUnk4MCt6VWw3WUlSMjErKzVnMFVnNkRnQlBEdmFJSHFSTnRsZ2gyVXdTL0cva1lOaUlEY0JiS1EKOUcvdTI4ekRIRUtNL21YYS8wYnFtSm16ZUYvY1BLdHdScFE3clFoRnAwUkdFcnZtc0l4dDl6K0ZZZUdncjFBYUVTV0ZlTApmUmZsa0lnOVBWOEl0b09GN25qK2VtMkxkNTNCS1hSUUFBQU1FQXhDTFBueHFFVEsyMW5QOXFxQVYzMEZUUkhGNW9kRHg4ClpiYnZIbjgwdEgxQjYwZjRtTGJFRm56REZFR0NwS2Rwb3dyUXR6WUhnQzNBaGNJUE9BbXFXaDg0UEFPbisreHhFanNaQkwKRWhVWmNFUndkYTMzTnJDNTVEMzZxbDBMZEsrSGRuZUFzVGZhazh0bWVlOTJWb0RxdWovNGFSMjBmUTBJUFVzMU8rWHNRNQpGWVFYQzZndExHZGRzRVFoSDF6MTh6RGtWa1UwdEhlZkJaL2pFZXBiOEZScXoxR1hpT0hGK2xBZVE2b3crS0xlcWtCcXQ4CkZxMHhGdG90SlF4VnFWQUFBQUYycHZhVzV1YVhOQVJFVlRTMVJQVUMxUVRWVkdVRFpOQVFJRAotLS0tLUVORCBPUEVOU1NIIFBSSVZBVEUgS0VZLS0tLS0K"
          protected_settings = validate_and_get_protected_settings(rsa_key, '', '', '')
          self.assertEqual('sshPrivateKey' in protected_settings, True)
          self.assertEqual(protected_settings.get('sshPrivateKey'), rsa_key)
  
      def test_dsa_private_key(self):
          key = DSA.generate(2048)
          private_key_encoded = base64.b64encode(key.export_key()).decode('utf-8')
          protected_settings = validate_and_get_protected_settings(private_key_encoded, '', '', '')
          self.assertEqual('sshPrivateKey' in protected_settings, True)
          self.assertEqual(protected_settings.get('sshPrivateKey'), private_key_encoded)
  
      def test_ecdsa_private_key(self):
          ecdsa_key = "LS0tLS1CRUdJTiBPUEVOU1NIIFBSSVZBVEUgS0VZLS0tLS0KYjNCbGJuTnphQzFyWlhrdGRqRUFBQUFBQkc1dmJtVUFBQUFFYm05dVpRQUFBQUFBQUFBQkFBQUFhQUFBQUJObFkyUnpZUwoxemFHRXlMVzVwYzNSd01qVTJBQUFBQ0c1cGMzUndNalUyQUFBQVFRUjBRc1BjWmJKeWZPaXE2a1M1d0VaeE5DbmR2YVJHCm1ETEUvVVBjakpDTDZQTVIyZmdPS2NnWlhzTEZkTUFzSnExS2d6TmNDN0ZXNGE0L0wrYTFWWUxDQUFBQXNIZ1RqTFY0RTQKeTFBQUFBRTJWalpITmhMWE5vWVRJdGJtbHpkSEF5TlRZQUFBQUlibWx6ZEhBeU5UWUFBQUJCQkhSQ3c5eGxzbko4NktycQpSTG5BUm5FMEtkMjlwRWFZTXNUOVE5eU1rSXZvOHhIWitBNHB5Qmxld3NWMHdDd21yVXFETTF3THNWYmhyajh2NXJWVmdzCklBQUFBZ0h1U3laU0NUZzJZbVNpOG9aY2c0cnVpODh0T1NUSm1aRVhkR09hdExySHNBQUFBWGFtOXBibTVwYzBCRVJWTkwKVkU5UUxWQk5WVVpRTmswQgotLS0tLUVORCBPUEVOU1NIIFBSSVZBVEUgS0VZLS0tLS0K"
          protected_settings = validate_and_get_protected_settings(ecdsa_key, '', '', '')
          self.assertEqual('sshPrivateKey' in protected_settings, True)
          self.assertEqual(protected_settings.get('sshPrivateKey'), ecdsa_key)
  
      def test_ed25519_private_key(self):
          ed25519_key = "LS0tLS1CRUdJTiBPUEVOU1NIIFBSSVZBVEUgS0VZLS0tLS0KYjNCbGJuTnphQzFyWlhrdGRqRUFBQUFBQkc1dmJtVUFBQUFFYm05dVpRQUFBQUFBQUFBQkFBQUFNd0FBQUF0emMyZ3RaVwpReU5UVXhPUUFBQUNCNjF0RzkrNGFmOTZsWGoyUStjWjJMT2JpV1liMlRtWVR6N3NSV0JDM1hVZ0FBQUtCRzFWRWZSdFZSCkh3QUFBQXR6YzJndFpXUXlOVFV4T1FBQUFDQjYxdEc5KzRhZjk2bFhqMlErY1oyTE9iaVdZYjJUbVlUejdzUldCQzNYVWcKQUFBRURRTStLcCtOSWpJVUhSUklqRFE5VDZ0U0V0SG9Ic0w1QjlwbHpCNlZ2MnluclcwYjM3aHAvM3FWZVBaRDV4bllzNQp1SlpodlpPWmhQUHV4RllFTGRkU0FBQUFGMnB2YVc1dWFYTkFSRVZUUzFSUFVDMVFUVlZHVURaTkFRSURCQVVHCi0tLS0tRU5EIE9QRU5TU0ggUFJJVkFURSBLRVktLS0tLQo="
          protected_settings = validate_and_get_protected_settings(ed25519_key, '', '', '')
          self.assertEqual('sshPrivateKey' in protected_settings, True)
          self.assertEqual(protected_settings.get('sshPrivateKey'), ed25519_key)
  
  
  class TestValidateK8sNaming(unittest.TestCase):
      def test_long_operator_namespace(self):
          operator_namespace = "thisisaverylongnamethatistoolongtobeused"
          namespace = OperatorNamespace(operator_namespace)
          err = 'Error! Invalid --operator-namespace'
          with self.assertRaises(InvalidArgumentValueError) as cm:
              validators.validate_operator_namespace(namespace)
          self.assertEqual(str(cm.exception), err)
  
      def test_long_operator_instance_name(self):
          operator_instance_name = "thisisaverylongnamethatistoolongtobeused"
          namespace = OperatorInstanceName(operator_instance_name)
          err = 'Error! Invalid --operator-instance-name'
          with self.assertRaises(InvalidArgumentValueError) as cm:
              validators.validate_operator_instance_name(namespace)
          self.assertEqual(str(cm.exception), err)
  
      def test_caps_operator_namespace(self):
          operator_namespace = 'Myoperatornamespace'
          namespace = OperatorNamespace(operator_namespace)
          err = 'Error! Invalid --operator-namespace'
          with self.assertRaises(InvalidArgumentValueError) as cm:
              validators.validate_operator_namespace(namespace)
          self.assertEqual(str(cm.exception), err)
  
      def test_caps_operator_instance_name(self):
          operator_instance_name = 'Myoperatorname'
          namespace = OperatorInstanceName(operator_instance_name)
          err = 'Error! Invalid --operator-instance-name'
          with self.assertRaises(InvalidArgumentValueError) as cm:
              validators.validate_operator_instance_name(namespace)
          self.assertEqual(str(cm.exception), err)
  
      def test_long_config_name(self):
          config_name = "thisisaverylongnamethatistoolongtobeusedthisisaverylongnamethatistoolongtobeused"
          err = 'Error! Invalid --name'
          with self.assertRaises(InvalidArgumentValueError) as cm:
              validators.validate_configuration_name(config_name)
          self.assertEqual(str(cm.exception), err)
  
      def test_valid_config_name(self):
          config_name = "this-is-a-valid-config"
          validators.validate_configuration_name(config_name)
  
      def test_caps_config_name(self):
          config_name = "ThisIsaCapsConfigName"
          err = 'Error! Invalid --name'
          with self.assertRaises(InvalidArgumentValueError) as cm:
              validators.validate_configuration_name(config_name)
          self.assertEqual(str(cm.exception), err)
  
      def test_dot_config_name(self):
          config_name = "a234567890b234567890c234567890d234567890e234567890f234567890.23"
          err = 'Error! Invalid --name'
          with self.assertRaises(InvalidArgumentValueError) as cm:
              validators.validate_configuration_name(config_name)
          self.assertEqual(str(cm.exception), err)
  
      def test_end_hyphen_config_name(self):
          config_name = "a234567890b234567890c234567890d234567890e234567890f23456789023-"
          err = 'Error! Invalid --name'
          with self.assertRaises(InvalidArgumentValueError) as cm:
              validators.validate_configuration_name(config_name)
          self.assertEqual(str(cm.exception), err)
  
  
  class TestValidateURLWithParams(unittest.TestCase):
      def test_ssh_private_key_with_ssh_url(self):
          validate_url_with_params('git@github.com:jonathan-innis/helm-operator-get-started-private.git', True, False, False)
  
      def test_ssh_known_hosts_with_ssh_url(self):
          validate_url_with_params('git@github.com:jonathan-innis/helm-operator-get-started-private.git', False, True, False)
  
      def test_https_auth_with_https_url(self):
          validate_url_with_params('https://github.com/jonathan-innis/helm-operator-get-started-private.git', False, False, True)
  
      def test_ssh_private_key_with_https_url(self):
          err = 'Error! An ssh private key cannot be used with an http(s) url'
          with self.assertRaises(MutuallyExclusiveArgumentError) as cm:
              validate_url_with_params('https://github.com/jonathan-innis/helm-operator-get-started-private.git', True, False, False)
          self.assertEqual(str(cm.exception), err)
  
      def test_ssh_known_hosts_with_https_url(self):
          err = 'Error! ssh known_hosts cannot be used with an http(s) url'
          with self.assertRaises(MutuallyExclusiveArgumentError) as cm:
              validate_url_with_params('https://github.com/jonathan-innis/helm-operator-get-started-private.git', False, True, False)
          self.assertEqual(str(cm.exception), err)
  
      def test_https_auth_with_ssh_url(self):
          err = 'Error! https auth (--https-user and --https-key) cannot be used with a non-http(s) url'
          with self.assertRaises(MutuallyExclusiveArgumentError) as cm:
              validate_url_with_params('git@github.com:jonathan-innis/helm-operator-get-started-private.git', False, False, True)
          self.assertEqual(str(cm.exception), err)
  
  
  class TestValidateKnownHosts(unittest.TestCase):
      def test_valid_known_hosts(self):
          known_hosts_raw = "ssh.dev.azure.com ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQC7Hr1oTWqNqOlzGJOfGJ4NakVyIzf1rXYd4d7wo6jBlkLvCA4odBlL0mDUyZ0/QUfTTqeu+tm22gOsv+VrVTMk6vwRU75gY/y9ut5Mb3bR5BV58dKXyq9A9UeB5Cakehn5Zgm6x1mKoVyf+FFn26iYqXJRgzIZZcZ5V6hrE0Qg39kZm4az48o0AUbf6Sp4SLdvnuMa2sVNwHBboS7EJkm57XQPVU3/QpyNLHbWDdzwtrlS+ez30S3AdYhLKEOxAG8weOnyrtLJAUen9mTkol8oII1edf7mWWbWVf0nBmly21+nZcmCTISQBtdcyPaEno7fFQMDD26/s0lfKob4Kw8H"
          known_hosts_encoded = base64.b64encode(known_hosts_raw.encode('utf-8')).decode('utf-8')
          validate_known_hosts(known_hosts_encoded)
  
      def test_valid_known_hosts_with_comment(self):
          known_hosts_raw = "ssh.dev.azure.com ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQC7Hr1oTWqNqOlzGJOfGJ4NakVyIzf1rXYd4d7wo6jBlkLvCA4odBlL0mDUyZ0/QUfTTqeu+tm22gOsv+VrVTMk6vwRU75gY/y9ut5Mb3bR5BV58dKXyq9A9UeB5Cakehn5Zgm6x1mKoVyf+FFn26iYqXJRgzIZZcZ5V6hrE0Qg39kZm4az48o0AUbf6Sp4SLdvnuMa2sVNwHBboS7EJkm57XQPVU3/QpyNLHbWDdzwtrlS+ez30S3AdYhLKEOxAG8weOnyrtLJAUen9mTkol8oII1edf7mWWbWVf0nBmly21+nZcmCTISQBtdcyPaEno7fFQMDD26/s0lfKob4Kw8H ThisIsAValidComment"
          known_hosts_encoded = base64.b64encode(known_hosts_raw.encode('utf-8')).decode('utf-8')
          validate_known_hosts(known_hosts_encoded)
  
      def test_valid_known_hosts_with_comment_own_line(self):
          known_hosts_raw = "#this is a comment on its own line\nssh.dev.azure.com ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQC7Hr1oTWqNqOlzGJOfGJ4NakVyIzf1rXYd4d7wo6jBlkLvCA4odBlL0mDUyZ0/QUfTTqeu+tm22gOsv+VrVTMk6vwRU75gY/y9ut5Mb3bR5BV58dKXyq9A9UeB5Cakehn5Zgm6x1mKoVyf+FFn26iYqXJRgzIZZcZ5V6hrE0Qg39kZm4az48o0AUbf6Sp4SLdvnuMa2sVNwHBboS7EJkm57XQPVU3/QpyNLHbWDdzwtrlS+ez30S3AdYhLKEOxAG8weOnyrtLJAUen9mTkol8oII1edf7mWWbWVf0nBmly21+nZcmCTISQBtdcyPaEno7fFQMDD26/s0lfKob4Kw8H"
          known_hosts_encoded = base64.b64encode(known_hosts_raw.encode('utf-8')).decode('utf-8')
          validate_known_hosts(known_hosts_encoded)
  
      def test_invalid_known_hosts(self):
          known_hosts_raw = "thisisabadknownhostsfilethatisaninvalidformat"
          known_hosts_encoded = base64.b64encode(known_hosts_raw.encode('utf-8')).decode('utf-8')
          err = 'Error! ssh known_hosts provided in wrong format'
          with self.assertRaises(InvalidArgumentValueError) as cm:
              validate_known_hosts(known_hosts_encoded)
          self.assertEqual(str(cm.exception), err)
  
  
  class OperatorNamespace:
      def __init__(self, operator_namespace):
          self.operator_namespace = operator_namespace
  
  
  class OperatorInstanceName:
      def __init__(self, operator_instance_name):
          self.operator_instance_name = operator_instance_name
  