Merge branch 'robinostlund:master' into master

pyproject.toml

@@ -3,7 +3,7 @@ requires = [
     "setuptools>=45",
     "wheel>=0.37.0",
     "setuptools_scm>=6.x",
-    "pytest>=5,<8",
+    "pytest>=6,<8",
 ]
 build-backend = "setuptools.build_meta"
 
@@ -13,7 +13,7 @@ write_to = "volkswagencarnet/version.py"
 
 [tool.black]
 line-length = 120
-target-version = ['py36', 'py37', 'py38', 'py39', 'py310']
+target-version = ['py37', 'py38', 'py39', 'py310']
 include = '\.pyi?$'
 extend-exclude = '''
 /(
@@ -23,3 +23,10 @@ extend-exclude = '''
   | profiling
 )/
 '''
+
+[tool.pytest.ini_options]
+minversion = "6.0"
+addopts = "-ra"
+asyncio_mode = "strict"
+testpaths = ["tests"]
+python_files = ["*_test.py"]

requirements-test.txt

@@ -6,5 +6,6 @@ flake8
 flake8-docstrings
 pre-commit
 pytest-cov
+pytest-subtests
 black
 freezegun>=1.0.0

setup.cfg

@@ -76,12 +76,6 @@ count = True
 verbose = True
 show_source = True
 
-[tool:pytest]
-minversion = 5.4.3
-addopts = -ra -q
-testpaths = tests
-python_files = *_test.py
-
 [coverage:run]
 branch = True
 omit = tests/*,volkswagencarnet/version.py  # define paths to omit

tests/vw_connection_test.py

@@ -1,18 +1,19 @@
-import logging.config
+"""Tests for main connection class."""
 import sys
 import unittest
 
-# This won't work on python versions less than 3.8
 if sys.version_info >= (3, 8):
+    # This won't work on python versions less than 3.8
     from unittest import IsolatedAsyncioTestCase
 else:
 
     class IsolatedAsyncioTestCase(unittest.TestCase):
+        """Dummy class to use instead (tests might need to skipped separately also)."""
+
         pass
 
 
 from io import StringIO
-from sys import argv
 from unittest.mock import patch
 
 import pytest
@@ -24,31 +25,44 @@ class IsolatedAsyncioTestCase(unittest.TestCase):
 
 @pytest.mark.skipif(condition=sys.version_info < (3, 8), reason="Test incompatible with Python < 3.8")
 def test_clear_cookies(connection):
+    """Check that we can clear old cookies."""
     assert len(connection._session._cookie_jar._cookies) > 0
     connection._clear_cookies()
     assert len(connection._session._cookie_jar._cookies) == 0
 
 
 class CmdLineTest(IsolatedAsyncioTestCase):
+    """Tests mostly for testing how to test..."""
+
     class FailingLoginConnection:
+        """This connection always fails login."""
+
         def __init__(self, sess, **kwargs):
+            """Init."""
             self._session = sess
 
         async def doLogin(self):
+            """Failed login attempt."""
             return False
 
     class TwoVehiclesConnection:
+        """Connection that return two vehicles."""
+
         def __init__(self, sess, **kwargs):
+            """Init."""
             self._session = sess
 
         async def doLogin(self):
+            """No-op update."""
             return True
 
         async def update(self):
+            """No-op update."""
             return True
 
         @property
         def vehicles(self):
+            """Return the vehicles."""
             vehicle1 = Vehicle(None, "vin1")
             vehicle2 = Vehicle(None, "vin2")
             return [vehicle1, vehicle2]
@@ -58,31 +72,31 @@ def vehicles(self):
     @patch("volkswagencarnet.vw_connection.Connection", spec_set=Connection, new=FailingLoginConnection)
     @pytest.mark.skipif(condition=sys.version_info < (3, 8), reason="Test incompatible with Python < 3.8")
     async def test_main_argv(self, logger_config):
-        # TODO: use patch to only change argv during the test?
-        if "-v" in argv:
-            argv.remove("-v")
-        if "-vv" in argv:
-            argv.remove("-vv")
-        # Assert default logger level is ERROR
-        await volkswagencarnet.vw_connection.main()
-        logger_config.assert_called_with(level=logging.ERROR)
-
-        # -v should be INFO
-        argv.append("-v")
-        await volkswagencarnet.vw_connection.main()
-        logger_config.assert_called_with(level=logging.INFO)
-        argv.remove("-v")
-
-        # -vv should be DEBUG
-        argv.append("-vv")
-        await volkswagencarnet.vw_connection.main()
-        logger_config.assert_called_with(level=logging.DEBUG)
+        """Test verbosity flags."""
+        from logging import ERROR
+        from logging import INFO
+        from logging import DEBUG
+
+        cases = [
+            ["none", [], ERROR],
+            ["-v", ["-v"], INFO],
+            ["-v2", ["-v2"], ERROR],
+            ["-vv", ["-vv"], DEBUG],
+        ]
+        for c in cases:
+            args = ["dummy"]
+            args.extend(c[1])
+            with patch.object(volkswagencarnet.vw_connection.sys, "argv", args), self.subTest(msg=c[0]):
+                await volkswagencarnet.vw_connection.main()
+                logger_config.assert_called_with(level=c[2])
+                logger_config.reset()
 
     @pytest.mark.asyncio
     @patch("sys.stdout", new_callable=StringIO)
     @patch("volkswagencarnet.vw_connection.Connection", spec_set=Connection, new=FailingLoginConnection)
     @pytest.mark.skipif(condition=sys.version_info < (3, 8), reason="Test incompatible with Python < 3.8")
     async def test_main_output_failed(self, stdout: StringIO):
+        """Verify empty stdout on failed login."""
         await volkswagencarnet.vw_connection.main()
         assert stdout.getvalue() == ""
 
@@ -91,6 +105,7 @@ async def test_main_output_failed(self, stdout: StringIO):
     @patch("volkswagencarnet.vw_connection.Connection", spec_set=Connection, new=TwoVehiclesConnection)
     @pytest.mark.skipif(condition=sys.version_info < (3, 8), reason="Test incompatible with Python < 3.8")
     async def test_main_output_two_vehicles(self, stdout: StringIO):
+        """Get console output for two vehicles."""
         await volkswagencarnet.vw_connection.main()
         assert (
             stdout.getvalue()

volkswagencarnet/vw_connection.py

@@ -4,6 +4,7 @@
 import base64
 import re
 import secrets
+import sys
 import time
 import logging
 import asyncio
@@ -12,7 +13,7 @@
 
 import jwt
 
-from sys import version_info, argv
+from sys import version_info
 from datetime import timedelta, datetime
 from urllib.parse import urljoin, parse_qs, urlparse
 from json import dumps as to_json
@@ -50,11 +51,11 @@
 
 
 class Connection:
-    """Connection to VW-Group Connect services"""
+    """Connection to VW-Group Connect services."""
 
     # Init connection class
     def __init__(self, session, username, password, fulldebug=False, country=COUNTRY, interval=timedelta(minutes=5)):
-        """Initialize"""
+        """Initialize."""
         self._x_client_id = None
         self._session = session
         self._session_fulldebug = fulldebug
@@ -86,7 +87,7 @@ def _clear_cookies(self):
 
     # API Login
     async def doLogin(self, tries: int = 1):
-        """Login method, clean login"""
+        """Login method, clean login."""
         _LOGGER.debug("Initiating new login")
 
         for i in range(tries):
@@ -416,7 +417,7 @@ async def _getAPITokens(self):
         return True
 
     async def terminate(self):
-        """Log out from connect services"""
+        """Log out from connect services."""
         _LOGGER.info("Initiating logout")
         await self.logout()
 
@@ -454,7 +455,7 @@ async def logout(self):
 
     # HTTP methods to API
     async def _request(self, method, url, **kwargs):
-        """Perform a query to the VW-Group API"""
+        """Perform a query to the VW-Group API."""
         _LOGGER.debug(f'HTTP {method} "{url}"')
         async with self._session.request(
             method,
@@ -912,7 +913,7 @@ async def get_sec_token(self, vin, spin, action):
 
     # Data set functions #
     async def dataCall(self, query, vin="", **data):
-        """Function to execute actions through VW-Group API."""
+        """Execute actions through VW-Group API."""
         if self.logged_in is False:
             if not await self.doLogin():
                 _LOGGER.warning(f"Login for {BRAND} account failed!")
@@ -950,7 +951,7 @@ async def dataCall(self, query, vin="", **data):
         return False
 
     async def setRefresh(self, vin):
-        """ "Force vehicle data update."""
+        """Force vehicle data update."""
         try:
             await self.set_token("vwg")
             response = await self.dataCall(
@@ -1028,14 +1029,14 @@ async def setClimater(self, vin, data, spin):
             raise
 
     async def setPreHeater(self, vin, data, spin):
-        contType = None
         """Petrol/diesel parking heater actions."""
+        content_type = None
         try:
             await self.set_token("vwg")
             if "Content-Type" in self._session_headers:
-                contType = self._session_headers["Content-Type"]
+                content_type = self._session_headers["Content-Type"]
             else:
-                contType = ""
+                content_type = ""
             self._session_headers["Content-Type"] = "application/vnd.vwg.mbb.RemoteStandheizung_v2_0_2+json"
             if "quickstop" not in data:
                 self._session_headers["x-mbbSecToken"] = await self.get_sec_token(vin=vin, spin=spin, action="heating")
@@ -1045,8 +1046,8 @@ async def setPreHeater(self, vin, data, spin):
             # Clean up headers
             self._session_headers.pop("x-mbbSecToken", None)
             self._session_headers.pop("Content-Type", None)
-            if contType:
-                self._session_headers["Content-Type"] = contType
+            if content_type:
+                self._session_headers["Content-Type"] = content_type
 
             if not response:
                 raise Exception("Invalid or no response")
@@ -1063,20 +1064,20 @@ async def setPreHeater(self, vin, data, spin):
         except Exception:
             self._session_headers.pop("x-mbbSecToken", None)
             self._session_headers.pop("Content-Type", None)
-            if contType:
-                self._session_headers["Content-Type"] = contType
+            if content_type:
+                self._session_headers["Content-Type"] = content_type
             raise
 
     async def setLock(self, vin, data, spin):
-        contType = None
         """Remote lock and unlock actions."""
+        content_type = None
         try:
             await self.set_token("vwg")
             # Prepare data, headers and fetch security token
             if "Content-Type" in self._session_headers:
-                contType = self._session_headers["Content-Type"]
+                content_type = self._session_headers["Content-Type"]
             else:
-                contType = ""
+                content_type = ""
             if "unlock" in data:
                 self._session_headers["X-mbbSecToken"] = await self.get_sec_token(vin=vin, spin=spin, action="unlock")
             else:
@@ -1088,8 +1089,8 @@ async def setLock(self, vin, data, spin):
             # Clean up headers
             self._session_headers.pop("X-mbbSecToken", None)
             self._session_headers.pop("Content-Type", None)
-            if contType:
-                self._session_headers["Content-Type"] = contType
+            if content_type:
+                self._session_headers["Content-Type"] = content_type
             if not response:
                 raise Exception("Invalid or no response")
             elif response == 429:
@@ -1106,14 +1107,14 @@ async def setLock(self, vin, data, spin):
         except:
             self._session_headers.pop("X-mbbSecToken", None)
             self._session_headers.pop("Content-Type", None)
-            if contType:
-                self._session_headers["Content-Type"] = contType
+            if content_type:
+                self._session_headers["Content-Type"] = content_type
             raise
 
     # Token handling #
     @property
     async def validate_tokens(self):
-        """Function to validate expiry of tokens."""
+        """Validate expiry of tokens."""
         idtoken = self._session_tokens["identity"]["id_token"]
         atoken = self._session_tokens["vwg"]["access_token"]
         id_exp = jwt.decode(
@@ -1144,7 +1145,7 @@ async def validate_tokens(self):
         return True
 
     async def verify_tokens(self, token, type, client="Legacy"):
-        """Function to verify JWT against JWK(s)."""
+        """Verify JWT against JWK(s)."""
         if type == "identity":
             req = await self._session.get(url="https://identity.vwgroup.io/oidc/v1/keys")
             keys = await req.json()
@@ -1180,7 +1181,7 @@ async def verify_tokens(self, token, type, client="Legacy"):
             return False
 
     async def refresh_tokens(self):
-        """Function to refresh tokens."""
+        """Refresh tokens."""
         try:
             tHeaders = {
                 "Accept-Encoding": "gzip, deflate, br",
@@ -1247,6 +1248,10 @@ def vehicles(self):
 
     @property
     def logged_in(self):
+        """
+        Return cached logged in state.
+        Not actually checking anything.
+        """
         return self._session_logged_in
 
     def vehicle(self, vin):
@@ -1262,6 +1267,7 @@ def hash_spin(self, challenge, spin):
 
     @property
     async def validate_login(self):
+        """Check that we have a valid access token."""
         try:
             if not await self.validate_tokens:
                 return False
@@ -1273,10 +1279,10 @@ async def validate_login(self):
 
 
 async def main():
-    """Main method."""
-    if "-v" in argv:
+    """Run the program."""
+    if "-v" in sys.argv:
         logging.basicConfig(level=logging.INFO)
-    elif "-vv" in argv:
+    elif "-vv" in sys.argv:
         logging.basicConfig(level=logging.DEBUG)
     else:
         logging.basicConfig(level=logging.ERROR)