



'''simple docstring'''




import json
import os
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from requests.exceptions import HTTPError

from transformers.utils import (
    CONFIG_NAME,
    FLAX_WEIGHTS_NAME,
    TF2_WEIGHTS_NAME,
    TRANSFORMERS_CACHE,
    WEIGHTS_NAME,
    cached_file,
    get_file_from_repo,
    has_file,
)


UpperCamelCase__ :			Optional[Any]												=				'''hf-internal-testing/tiny-random-bert'''
UpperCamelCase__ :			Dict												=				os.path.join(TRANSFORMERS_CACHE, '''models--hf-internal-testing--tiny-random-bert''')
UpperCamelCase__ :			str												=				'''9b8c223d42b2188cb49d29af482996f9d0f3e5a6'''






class 		_UpperCamelCase						(							unittest.TestCase					):


			'''simple docstring'''

			def UpperCamelCase__      (							self  :				Optional[Any]		):





					"""simple docstring"""






					__SCREAMING_SNAKE_CASE :	Optional[Any]					 =	cached_file(lowerCAmelCase__							,  lowerCAmelCase__		)
					# Should have downloaded the file in here
					self.assertTrue(os.path.isdir(lowerCAmelCase__		)		)
					# Cache should contain at least those three subfolders:
					for subfolder in ["blobs", "refs", "snapshots"]:
							self.assertTrue(os.path.isdir(os.path.join(lowerCAmelCase__							,  lowerCAmelCase__		)		)		)
					with open(os.path.join(lowerCAmelCase__							,  """refs"""							,  """main"""		)		) as f:
							__SCREAMING_SNAKE_CASE :	Dict					 =	f.read()
					self.assertEqual(lowerCAmelCase__							,  os.path.join(lowerCAmelCase__							,  """snapshots"""							,  lowerCAmelCase__							,  lowerCAmelCase__		)		)
					self.assertTrue(os.path.isfile(lowerCAmelCase__		)		)

					# File is cached at the same place the second time.
					__SCREAMING_SNAKE_CASE :	Any					 =	cached_file(lowerCAmelCase__							,  lowerCAmelCase__		)
					self.assertEqual(lowerCAmelCase__							,  lowerCAmelCase__		)

					# Using a specific revision to test the full commit hash.
					__SCREAMING_SNAKE_CASE :	str					 =	cached_file(lowerCAmelCase__							,  lowerCAmelCase__							,  revision="""9b8c223"""		)
					self.assertEqual(lowerCAmelCase__							,  os.path.join(lowerCAmelCase__							,  """snapshots"""							,  lowerCAmelCase__							,  lowerCAmelCase__		)		)

			def UpperCamelCase__      (							self  :				Union[str, Any]		):





					"""simple docstring"""






					with self.assertRaisesRegex(lowerCAmelCase__							,  """is not a valid model identifier"""		):
							__SCREAMING_SNAKE_CASE :	int					 =	cached_file("""tiny-random-bert"""							,  lowerCAmelCase__		)

					with self.assertRaisesRegex(lowerCAmelCase__							,  """is not a valid git identifier"""		):
							__SCREAMING_SNAKE_CASE :	List[Any]					 =	cached_file(lowerCAmelCase__							,  lowerCAmelCase__							,  revision="""aaaa"""		)

					with self.assertRaisesRegex(lowerCAmelCase__							,  """does not appear to have a file named"""		):
							__SCREAMING_SNAKE_CASE :	List[Any]					 =	cached_file(lowerCAmelCase__							,  """conf"""		)

			def UpperCamelCase__      (							self  :				Optional[Any]		):





					"""simple docstring"""






					with self.assertRaisesRegex(lowerCAmelCase__							,  """does not appear to have a file named"""		):
							__SCREAMING_SNAKE_CASE :	List[str]					 =	cached_file(lowerCAmelCase__							,  """conf"""		)

					with open(os.path.join(lowerCAmelCase__							,  """refs"""							,  """main"""		)		) as f:
							__SCREAMING_SNAKE_CASE :	Optional[int]					 =	f.read()
					self.assertTrue(os.path.isfile(os.path.join(lowerCAmelCase__							,  """.no_exist"""							,  lowerCAmelCase__							,  """conf"""		)		)		)

					__SCREAMING_SNAKE_CASE :	List[Any]					 =	cached_file(lowerCAmelCase__							,  """conf"""							,  _raise_exceptions_for_missing_entries=lowerCAmelCase__		)
					self.assertIsNone(lowerCAmelCase__		)

					__SCREAMING_SNAKE_CASE :	Optional[Any]					 =	cached_file(lowerCAmelCase__							,  """conf"""							,  local_files_only=lowerCAmelCase__							,  _raise_exceptions_for_missing_entries=lowerCAmelCase__		)
					self.assertIsNone(lowerCAmelCase__		)

					__SCREAMING_SNAKE_CASE :	Dict					 =	mock.Mock()
					__SCREAMING_SNAKE_CASE :	Dict					 =	5_0_0
					__SCREAMING_SNAKE_CASE :	str					 =	{}
					__SCREAMING_SNAKE_CASE :	Optional[int]					 =	HTTPError
					__SCREAMING_SNAKE_CASE :	Union[str, Any]					 =	{}

					# Under the mock environment we get a 500 error when trying to reach the tokenizer.
					with mock.patch("""requests.Session.request"""							,  return_value=lowerCAmelCase__		) as mock_head:
							__SCREAMING_SNAKE_CASE :	Union[str, Any]					 =	cached_file(lowerCAmelCase__							,  """conf"""							,  _raise_exceptions_for_connection_errors=lowerCAmelCase__		)
							self.assertIsNone(lowerCAmelCase__		)
							# This check we did call the fake head request
							mock_head.assert_called()

			def UpperCamelCase__      (							self  :				Optional[Any]		):





					"""simple docstring"""






					self.assertTrue(has_file("""hf-internal-testing/tiny-bert-pt-only"""							,  lowerCAmelCase__		)		)
					self.assertFalse(has_file("""hf-internal-testing/tiny-bert-pt-only"""							,  lowerCAmelCase__		)		)
					self.assertFalse(has_file("""hf-internal-testing/tiny-bert-pt-only"""							,  lowerCAmelCase__		)		)

			def UpperCamelCase__      (							self  :				List[Any]		):





					"""simple docstring"""






					self.assertIsNone(get_file_from_repo("""bert-base-cased"""							,  """ahah.txt"""		)		)

					# The function raises if the repository does not exist.
					with self.assertRaisesRegex(lowerCAmelCase__							,  """is not a valid model identifier"""		):
							get_file_from_repo("""bert-base-case"""							,  lowerCAmelCase__		)

					# The function raises if the revision does not exist.
					with self.assertRaisesRegex(lowerCAmelCase__							,  """is not a valid git identifier"""		):
							get_file_from_repo("""bert-base-cased"""							,  lowerCAmelCase__							,  revision="""ahaha"""		)

					__SCREAMING_SNAKE_CASE :	Any					 =	get_file_from_repo("""bert-base-cased"""							,  lowerCAmelCase__		)
					# The name is the cached name which is not very easy to test, so instead we load the content.
					__SCREAMING_SNAKE_CASE :	List[str]					 =	json.loads(open(lowerCAmelCase__							,  """r"""		).read()		)
					self.assertEqual(config["""hidden_size"""]							,  7_6_8		)



			def UpperCamelCase__      (							self  :				Optional[int]		):





					"""simple docstring"""






					with tempfile.TemporaryDirectory() as tmp_dir:
							__SCREAMING_SNAKE_CASE :	List[Any]					 =	Path(lowerCAmelCase__		) / """a.txt"""
							filename.touch()
							self.assertEqual(get_file_from_repo(lowerCAmelCase__							,  """a.txt"""		)							,  str(lowerCAmelCase__		)		)

							self.assertIsNone(get_file_from_repo(lowerCAmelCase__							,  """b.txt"""		)		)




'''simple docstring'''




from unittest.mock import Mock, patch

from file_transfer.send_file import send_file


@patch("""socket.socket"""      )
@patch("""builtins.open"""      )
def        lowerCAmelCase_     (			_lowerCamelCase:     Dict ,   _lowerCamelCase:     List[Any]      ):
		# ===== initialization =====
		__SCREAMING_SNAKE_CASE :	Optional[int]					 =	Mock()
		__SCREAMING_SNAKE_CASE :	Optional[int]					 =	conn, Mock()
		__SCREAMING_SNAKE_CASE :	str					 =	iter([1, None]      )
		__SCREAMING_SNAKE_CASE :	Any					 =	lambda _lowerCamelCase      : next(_lowerCamelCase      )

		# ===== invoke =====
		send_file(filename="""mytext.txt""" ,   testing=_lowerCamelCase      )

		# ===== ensurance =====
		sock.assert_called_once()
		sock.return_value.bind.assert_called_once()
		sock.return_value.listen.assert_called_once()
		sock.return_value.accept.assert_called_once()
		conn.recv.assert_called_once()

		file.return_value.__enter__.assert_called_once()
		file.return_value.__enter__.return_value.read.assert_called()

		conn.send.assert_called_once()
		conn.close.assert_called_once()
		sock.return_value.shutdown.assert_called_once()
		sock.return_value.close.assert_called_once()