text: #
  # 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.
  #
  """
  This module contains integration with Azure File Share.
  
  Cloud variant of a SMB file share. Make sure that a Airflow connection of
  type `wasb` exists. Authorization can be done by supplying a login (=Storage account name)
  and password (=Storage account key), or login and SAS token in the extra field
  (see connection `azure_fileshare_default` for an example).
  """
  
  import json
  import unittest
  from unittest import mock
  
  import pytest
  from azure.storage.file import Directory, File
  
  from airflow.models import Connection
  from airflow.providers.microsoft.azure.hooks.fileshare import AzureFileShareHook
  from airflow.utils import db
  
  
  class TestAzureFileshareHook(unittest.TestCase):
      def setUp(self):
          db.merge_conn(
              Connection(
                  conn_id='azure_fileshare_test_key',
                  conn_type='azure_file_share',
                  login='login',
                  password='key',
              )
          )
          db.merge_conn(
              Connection(
                  conn_id='azure_fileshare_extras',
                  conn_type='azure_fileshare',
                  login='login',
                  extra=json.dumps(
                      {
                          'extra__azure_fileshare__sas_token': 'token',
                          'extra__azure_fileshare__protocol': 'http',
                      }
                  ),
              )
          )
          db.merge_conn(
              # Neither password nor sas_token present
              Connection(
                  conn_id='azure_fileshare_missing_credentials',
                  conn_type='azure_fileshare',
                  login='login',
              )
          )
          db.merge_conn(
              Connection(
                  conn_id='azure_fileshare_extras_deprecated',
                  conn_type='azure_fileshare',
                  login='login',
                  extra=json.dumps(
                      {
                          'sas_token': 'token',
                      }
                  ),
              )
          )
          db.merge_conn(
              Connection(
                  conn_id='azure_fileshare_extras_deprecated_empty_wasb_extra',
                  conn_type='azure_fileshare',
                  login='login',
                  password='password',
                  extra=json.dumps(
                      {
                          'extra__azure_fileshare__shared_access_key': '',
                      }
                  ),
              )
          )
  
          db.merge_conn(
              Connection(
                  conn_id='azure_fileshare_extras_wrong',
                  conn_type='azure_fileshare',
                  login='login',
                  extra=json.dumps(
                      {
                          'wrong_key': 'token',
                      }
                  ),
              )
          )
  
      def test_key_and_connection(self):
          from azure.storage.file import FileService
  
          hook = AzureFileShareHook(azure_fileshare_conn_id='azure_fileshare_test_key')
          assert hook.conn_id == 'azure_fileshare_test_key'
          assert hook._conn is None
          print(hook.get_conn())
          assert isinstance(hook.get_conn(), FileService)
  
      def test_sas_token(self):
          from azure.storage.file import FileService
  
          hook = AzureFileShareHook(azure_fileshare_conn_id='azure_fileshare_extras')
          assert hook.conn_id == 'azure_fileshare_extras'
          assert isinstance(hook.get_conn(), FileService)
  
      def test_deprecated_sas_token(self):
          from azure.storage.file import FileService
  
          with pytest.warns(DeprecationWarning):
              hook = AzureFileShareHook(azure_fileshare_conn_id='azure_fileshare_extras_deprecated')
              assert hook.conn_id == 'azure_fileshare_extras_deprecated'
              assert isinstance(hook.get_conn(), FileService)
  
      def test_deprecated_wasb_connection(self):
          from azure.storage.file import FileService
  
          with pytest.warns(DeprecationWarning):
              hook = AzureFileShareHook(
                  azure_fileshare_conn_id='azure_fileshare_extras_deprecated_empty_wasb_extra'
              )
              assert hook.conn_id == 'azure_fileshare_extras_deprecated_empty_wasb_extra'
              assert isinstance(hook.get_conn(), FileService)
  
      def test_wrong_extras(self):
          hook = AzureFileShareHook(azure_fileshare_conn_id='azure_fileshare_extras_wrong')
          assert hook.conn_id == 'azure_fileshare_extras_wrong'
          with pytest.raises(TypeError, match=".*wrong_key.*"):
              hook.get_conn()
  
      def test_missing_credentials(self):
          hook = AzureFileShareHook(azure_fileshare_conn_id='azure_fileshare_missing_credentials')
          assert hook.conn_id == 'azure_fileshare_missing_credentials'
          with pytest.raises(ValueError, match=".*account_key or sas_token.*"):
              hook.get_conn()
  
      @mock.patch('airflow.providers.microsoft.azure.hooks.fileshare.FileService', autospec=True)
      def test_check_for_file(self, mock_service):
          mock_instance = mock_service.return_value
          mock_instance.exists.return_value = True
          hook = AzureFileShareHook(azure_fileshare_conn_id='azure_fileshare_extras')
          assert hook.check_for_file('share', 'directory', 'file', timeout=3)
          mock_instance.exists.assert_called_once_with('share', 'directory', 'file', timeout=3)
  
      @mock.patch('airflow.providers.microsoft.azure.hooks.fileshare.FileService', autospec=True)
      def test_check_for_directory(self, mock_service):
          mock_instance = mock_service.return_value
          mock_instance.exists.return_value = True
          hook = AzureFileShareHook(azure_fileshare_conn_id='azure_fileshare_extras')
          assert hook.check_for_directory('share', 'directory', timeout=3)
          mock_instance.exists.assert_called_once_with('share', 'directory', timeout=3)
  
      @mock.patch('airflow.providers.microsoft.azure.hooks.fileshare.FileService', autospec=True)
      def test_load_file(self, mock_service):
          mock_instance = mock_service.return_value
          hook = AzureFileShareHook(azure_fileshare_conn_id='azure_fileshare_extras')
          hook.load_file('path', 'share', 'directory', 'file', max_connections=1)
          mock_instance.create_file_from_path.assert_called_once_with(
              'share', 'directory', 'file', 'path', max_connections=1
          )
  
      @mock.patch('airflow.providers.microsoft.azure.hooks.fileshare.FileService', autospec=True)
      def test_load_string(self, mock_service):
          mock_instance = mock_service.return_value
          hook = AzureFileShareHook(azure_fileshare_conn_id='azure_fileshare_extras')
          hook.load_string('big string', 'share', 'directory', 'file', timeout=1)
          mock_instance.create_file_from_text.assert_called_once_with(
              'share', 'directory', 'file', 'big string', timeout=1
          )
  
      @mock.patch('airflow.providers.microsoft.azure.hooks.fileshare.FileService', autospec=True)
      def test_load_stream(self, mock_service):
          mock_instance = mock_service.return_value
          hook = AzureFileShareHook(azure_fileshare_conn_id='azure_fileshare_extras')
          hook.load_stream('stream', 'share', 'directory', 'file', 42, timeout=1)
          mock_instance.create_file_from_stream.assert_called_once_with(
              'share', 'directory', 'file', 'stream', 42, timeout=1
          )
  
      @mock.patch('airflow.providers.microsoft.azure.hooks.fileshare.FileService', autospec=True)
      def test_list_directories_and_files(self, mock_service):
          mock_instance = mock_service.return_value
          hook = AzureFileShareHook(azure_fileshare_conn_id='azure_fileshare_extras')
          hook.list_directories_and_files('share', 'directory', timeout=1)
          mock_instance.list_directories_and_files.assert_called_once_with('share', 'directory', timeout=1)
  
      @mock.patch('airflow.providers.microsoft.azure.hooks.fileshare.FileService', autospec=True)
      def test_list_files(self, mock_service):
          mock_instance = mock_service.return_value
          mock_instance.list_directories_and_files.return_value = [
              File("file1"),
              File("file2"),
              Directory("dir1"),
              Directory("dir2"),
          ]
          hook = AzureFileShareHook(azure_fileshare_conn_id='azure_fileshare_extras')
          files = hook.list_files('share', 'directory', timeout=1)
          assert files == ["file1", 'file2']
          mock_instance.list_directories_and_files.assert_called_once_with('share', 'directory', timeout=1)
  
      @mock.patch('airflow.providers.microsoft.azure.hooks.fileshare.FileService', autospec=True)
      def test_create_directory(self, mock_service):
          mock_instance = mock_service.return_value
          hook = AzureFileShareHook(azure_fileshare_conn_id='azure_fileshare_extras')
          hook.create_directory('share', 'directory', timeout=1)
          mock_instance.create_directory.assert_called_once_with('share', 'directory', timeout=1)
  
      @mock.patch('airflow.providers.microsoft.azure.hooks.fileshare.FileService', autospec=True)
      def test_get_file(self, mock_service):
          mock_instance = mock_service.return_value
          hook = AzureFileShareHook(azure_fileshare_conn_id='azure_fileshare_extras')
          hook.get_file('path', 'share', 'directory', 'file', max_connections=1)
          mock_instance.get_file_to_path.assert_called_once_with(
              'share', 'directory', 'file', 'path', max_connections=1
          )
  
      @mock.patch('airflow.providers.microsoft.azure.hooks.fileshare.FileService', autospec=True)
      def test_get_file_to_stream(self, mock_service):
          mock_instance = mock_service.return_value
          hook = AzureFileShareHook(azure_fileshare_conn_id='azure_fileshare_extras')
          hook.get_file_to_stream('stream', 'share', 'directory', 'file', max_connections=1)
          mock_instance.get_file_to_stream.assert_called_once_with(
              'share', 'directory', 'file', 'stream', max_connections=1
          )
  
      @mock.patch('airflow.providers.microsoft.azure.hooks.fileshare.FileService', autospec=True)
      def test_create_share(self, mock_service):
          mock_instance = mock_service.return_value
          hook = AzureFileShareHook(azure_fileshare_conn_id='azure_fileshare_extras')
          hook.create_share('my_share')
          mock_instance.create_share.assert_called_once_with('my_share')
  
      @mock.patch('airflow.providers.microsoft.azure.hooks.fileshare.FileService', autospec=True)
      def test_delete_share(self, mock_service):
          mock_instance = mock_service.return_value
          hook = AzureFileShareHook(azure_fileshare_conn_id='azure_fileshare_extras')
          hook.delete_share('my_share')
          mock_instance.delete_share.assert_called_once_with('my_share')
  