chore: remove requests and replace with aiohttp (#796)

* chore: remove requests and replace with aiohttp

* fix linting

pyproject.toml

@@ -23,7 +23,6 @@ exclude = [ ]
  [tool.poetry.dependencies]
  python = "^3.9.0"
  pycryptodome = ">=3.18.0"
- requests = "^2.32.3"
 
 [tool.poetry.group.dev.dependencies]
 assertpy = "^1.1"
@@ -41,7 +40,6 @@ pytest-resource-path = "^1.3.0"
 pytest-mockservers = "^0.6.0"
 pytest-sugar = "^0.9.4"
 time-machine = "^2.7.0"
-types-requests = "^2.31.0.6"
 yamllint = "^1.26.3"
 freezegun = ">=1.5.1"
 

scripts/validate_token.py

@@ -16,6 +16,7 @@
 
 """Python script for validating a Token from Switcher."""
 
+import asyncio
 from argparse import ArgumentParser, RawDescriptionHelpFormatter
 
 from aioswitcher.device.tools import validate_token
@@ -49,12 +50,12 @@
 )
 
 
-def main() -> None:
+async def main() -> None:
  """Validate the personal Token of the user."""
  try:
  args = parser.parse_args()
 
- response = validate_token(args.username, args.token)
+ response = await validate_token(args.username, args.token)
  if response:
  print("Your Personal Token is valid")
  else:
@@ -65,4 +66,4 @@ def main() -> None:
 
 
 if __name__ == "__main__":
- main()
+ asyncio.run(main())

src/aioswitcher/device/tools.py

@@ -21,7 +21,7 @@
 from logging import getLogger
 from struct import pack
 
-import requests
+import aiohttp
 from Crypto.Cipher import AES
 from Crypto.Util.Padding import unpad
 
@@ -192,26 +192,28 @@ def convert_token_to_packet(token: str) -> str:
  raise RuntimeError("convert token to packet was not successful") from ve
 
 
-def validate_token(username: str, token: str) -> bool:
- """Make API call to validate a Token by username and token."""
+async def validate_token(username: str, token: str) -> bool:
+ """Make an asynchronous API call to validate a Token by username and token."""
  request_url = "https://switcher.co.il/ValidateToken/"
  request_data = {"email": username, "token": token}
  is_token_valid = False
 
  logger.debug("calling API call for Switcher to validate the token")
- response = requests.post(request_url, data=request_data)
-
- if response.status_code == 200:
- logger.debug("request successful")
- try:
- response_json = response.json()
- result = response_json.get("result", None)
- if result is not None:
- is_token_valid = result.lower() == "true"
- except ValueError:
- logger.debug("response content is not valid JSON")
- else:
- logger.debug("request failed with status code: %s", response.status_code)
+
+ async with aiohttp.ClientSession() as session:
+ async with session.post(request_url, data=request_data) as response:
+ if response.status == 200:
+ logger.debug("request successful")
+ try:
+ response_json = await response.json()
+ result = response_json.get("result", None)
+ if result is not None:
+ is_token_valid = result.lower() == "true"
+ except aiohttp.ContentTypeError:
+ logger.debug("response content is not valid JSON")
+ else:
+ logger.debug("request failed with status code: %s", response.status)
+
  return is_token_valid
 
 

tests/test_device_tools.py

@@ -19,6 +19,7 @@
 from struct import unpack
 from unittest.mock import patch
 
+import pytest_asyncio
 from assertpy import assert_that
 from pytest import mark
 
@@ -136,25 +137,25 @@ def test_convert_token_to_packet_with_false_token_should_throw_an_error(token, e
  ).when_called_with(token).is_equal_to(response)
 
 
-@patch('requests.post')
+@pytest_asyncio.fixture
 @mark.parametrize("username, token, is_token_valid", [
  ("test@switcher.com", "zvVvd7JxtN7CgvkD1Psujw==", True)
  ])
-def test_validate_token_should_return_token_valid(mock_post, username, token, is_token_valid):
+async def test_validate_token_should_return_token_valid(mock_post, username, token, is_token_valid):
  mock_response = mock_post.return_value
  mock_response.status_code = 200
  mock_response.json.return_value = {"result": "True"}
  assert_that(tools.validate_token(username, token)).is_equal_to(is_token_valid)
 
 
-@patch('requests.post')
+@pytest_asyncio.fixture
 @mark.parametrize("username, token, is_token_valid", [
  ("test@switcher.com", "", False),
  ("test@switcher.com", "notvalidtoken", False),
  ("", "notvalidtoken", False),
  ("", "", False)
  ])
-def test_validate_token_should_return_token_invalid(mock_post, username, token, is_token_valid):
+async def test_validate_token_should_return_token_invalid(mock_post, username, token, is_token_valid):
  mock_response = mock_post.return_value
  mock_response.status_code = 200
  mock_response.json.return_value = {"result": "False"}