Merge pull request #164 from robinostlund/tests

Tests

.github/workflows/codecov.yaml

@@ -15,12 +15,11 @@ jobs:
       - name: Setup Python
         uses: actions/setup-python@master
         with:
-          python-version: 3.7
+          python-version: 3.9
 
       - name: Generate coverage report
         run: |
-          pip install pytest
-          pip install pytest-cov
+          pip install -r requirements-test.txt
           pytest --cov=./ --cov-report=xml
 
       # documentation: https://github.com/codecov/codecov-action
@@ -30,7 +29,8 @@ jobs:
           directory: ./coverage/reports/
           env_vars: OS,PYTHON
           fail_ci_if_error: true
-          files: ./coverage1.xml,./coverage2.xml
+          files: ./coverage.xml
           flags: unittests
           name: codecov-volkswagencarnet
-          path_to_write_report: ./coverage/codecov_report.txt
+          # path_to_write_report: ./coverage/codecov_report.txt
+          verbose: true

.github/workflows/codeql-analysis.yml

@@ -22,7 +22,7 @@ jobs:
       fail-fast: false
       matrix:
         language: ["python"]
-        python-version: ["3.6", "3.7", "3.8", "3.9", "3.10"]
+        python-version: ["3.7", "3.8", "3.9", "3.10"]
 
     steps:
       - name: Checkout repository

.github/workflows/validate.yml

@@ -50,7 +50,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.6", "3.7", "3.8", "3.9", "3.10"]
+        python-version: ["3.7", "3.8", "3.9", "3.10"]
 
     steps:
       - name: Checkout code

.gitignore

@@ -5,8 +5,11 @@ build
 dist
 *.egg-info
 *vw.conf
+.coverage
+.eggs
 .vscode
 /tests/credentials.py
-.eggs
 /venv/
 /volkswagencarnet/version.py
+/volkswagencarnet.iml
+htmlcov

pytest.ini

@@ -1,3 +1,4 @@
 [pytest]
 asyncio_mode=strict
-addopts = --doctest-modules
+addopts = -ra
+minversion = 5.4.3

requirements-test.txt

@@ -2,7 +2,9 @@ lxml
 beautifulsoup4
 aiohttp
 pyjwt
-pytest
+pytest>=7.0.0
 setuptools
 pytest-asyncio
-flake8
+flake8
+pytest-cov
+black

setup.cfg

@@ -75,3 +75,11 @@ 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
+
+[coverage:report]
+#show_missing = True
+skip_covered = False

tests/conftest.py

@@ -0,0 +1,6 @@
+import sys
+
+pytest_plugins = ["pytest_cov"]
+
+if sys.version_info >= (3, 8):
+    pytest_plugins.append("tests.fixtures.connection")

tests/fixtures/connection.py

@@ -0,0 +1,27 @@
+import os
+from pathlib import Path
+
+import pytest
+import pytest_asyncio
+from aiohttp import CookieJar, ClientSession
+
+from volkswagencarnet.vw_connection import Connection
+
+current_path = Path(os.path.dirname(os.path.realpath(__file__)))
+resource_path = os.path.join(current_path, "resources")
+
+
+@pytest_asyncio.fixture
+async def session():
+    """Client session that can be used in tests"""
+    jar = CookieJar()
+    jar.load(os.path.join(resource_path, "dummy_cookies.pickle"))
+    sess = ClientSession(headers={"Connection": "keep-alive"}, cookie_jar=jar)
+    yield sess
+    await sess.close()
+
+
+@pytest.fixture
+def connection(session):
+    """Real connection for integration tests"""
+    return Connection(session=session, username="", password="", country="DE", interval=999, fulldebug=True)

tests/vw_connection_test.py

@@ -0,0 +1,110 @@
+import logging.config
+import sys
+
+# This won't work on python versions less than 3.8
+if sys.version_info >= (3, 8):
+    from unittest import IsolatedAsyncioTestCase
+else:
+
+    class IsolatedAsyncioTestCase:
+        pass
+
+
+import unittest
+from io import StringIO
+from sys import argv
+from unittest.mock import patch
+
+import pytest
+
+import volkswagencarnet.vw_connection
+from volkswagencarnet.vw_connection import Connection
+from volkswagencarnet.vw_vehicle import Vehicle
+
+
+@pytest.mark.skipif(condition=sys.version_info < (3, 8), reason="Test incompatible with Python < 3.8")
+def test_clear_cookies(connection):
+    assert len(connection._session._cookie_jar._cookies) > 0
+    connection._clear_cookies()
+    assert len(connection._session._cookie_jar._cookies) == 0
+
+
+class CmdLineTest(IsolatedAsyncioTestCase, unittest.TestCase):
+    class FailingLoginConnection:
+        def __init__(self, sess, **kwargs):
+            self._session = sess
+
+        async def doLogin(self):
+            return False
+
+    class TwoVehiclesConnection:
+        def __init__(self, sess, **kwargs):
+            self._session = sess
+
+        async def doLogin(self):
+            return True
+
+        async def update(self):
+            return True
+
+        @property
+        def vehicles(self):
+            vehicle1 = Vehicle(None, "vin1")
+            vehicle2 = Vehicle(None, "vin2")
+            return [vehicle1, vehicle2]
+
+    @pytest.mark.asyncio
+    @patch.object(volkswagencarnet.vw_connection.logging, "basicConfig")
+    @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)
+
+    @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):
+        await volkswagencarnet.vw_connection.main()
+        assert stdout.getvalue() == ""
+
+    @pytest.mark.asyncio
+    @patch("sys.stdout", new_callable=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):
+        await volkswagencarnet.vw_connection.main()
+        assert (
+            stdout.getvalue()
+            == """Vehicle id: vin1
+Supported sensors:
+ - Force data refresh (domain:switch) - Off
+ - Request results (domain:sensor) - Unknown
+ - Requests remaining (domain:sensor) - -1
+ - Request in progress (domain:binary_sensor) - Off
+Vehicle id: vin2
+Supported sensors:
+ - Force data refresh (domain:switch) - Off
+ - Request results (domain:sensor) - Unknown
+ - Requests remaining (domain:sensor) - -1
+ - Request in progress (domain:binary_sensor) - Off
+"""
+        )

tests/vw_utilities_test.py

@@ -0,0 +1,104 @@
+import unittest
+from datetime import datetime, timezone, timedelta
+from json import JSONDecodeError
+from unittest import mock
+from unittest.mock import DEFAULT
+
+from volkswagencarnet.vw_utilities import camel2slug, is_valid_path, obj_parser, json_loads, read_config
+
+
+class UtilitiesTest(unittest.TestCase):
+    def test_camel_to_slug(self):
+        data = {"foo": "foo", "fooBar": "foo_bar", "XYZ": "x_y_z", "B4R": "b4_r"}  # Should this actually be "b_4_r"? =)
+        for v in data:
+            res = camel2slug(v)
+            self.assertEqual(data[v], res)
+
+    def test_is_valid_path(self):
+        data = {
+            "None": [None, None, True],
+            "a in a": [{"a": 1}, "a", True],
+            "b in a": [{"a": 1}, "b", False],
+            "a.b in a.b": [{"a": {"b": 7}}, "a.b", True],
+            "list": [[1, "a", None], "a", TypeError],
+        }
+
+        for v in data:
+            with self.subTest():
+                try:
+                    if isinstance(data[v][2], bool):
+                        self.assertEqual(
+                            data[v][2],
+                            is_valid_path(data[v][0], data[v][1]),
+                            msg=f"Path validation error for {data[v][1]} in {data[v][0]}",
+                        )
+                    else:
+                        with self.assertRaises(data[v][2]):
+                            is_valid_path(data[v][0], data[v][1])
+                except Exception as e:
+                    if isinstance(e, AssertionError):
+                        raise
+                    self.fail(f"Wrong exception? Got {type(e)} but expected {data[v][2]}")
+
+    def test_obj_parser(self):
+        data = {
+            "int": [0, AttributeError],
+            "dict": [{"foo": "bar"}, {"foo": "bar"}],
+            "dict with time": [
+                {"foo": "2001-01-01T23:59:59Z"},
+                {"foo": datetime(2001, 1, 1, 23, 59, 59, tzinfo=timezone.utc)},
+            ],
+            "dict with timezone": [
+                {"foo": "2001-01-01T23:59:59+0200"},
+                {"foo": datetime(2001, 1, 1, 23, 59, 59, tzinfo=timezone(timedelta(hours=2)))},
+            ],
+        }
+        for v in data:
+            if isinstance(data[v][1], dict):
+                res = obj_parser(data[v][0])
+                self.assertEqual(data[v][1], res)
+            else:
+                with self.assertRaises(data[v][1]):
+                    obj_parser(data[v][0])
+
+    def test_json_loads(self):
+        expected = {"foo": {"bar": "baz"}}
+        actual = json_loads('{"foo":  {\n"bar":\t"baz"}}')
+        self.assertEqual(expected, actual)
+
+        self.assertEqual(42, json_loads("42"))
+
+        with self.assertRaises(JSONDecodeError):
+            json_loads("{[}")
+        with self.assertRaises(TypeError):
+            json_loads(42)
+
+    def test_read_config_success(self):
+        """successfully read configuration from a file"""
+        read_data = """
+# Comment
+foo: bar
+"""
+        mock_open = mock.mock_open(read_data=read_data)
+        with mock.patch("builtins.open", mock_open):
+            res = read_config()
+            self.assertEqual({"foo": "bar"}, res)
+
+    def test_read_config_error(self):
+        """success on second file, but parse error"""
+        read_data = """
+foo: bar
+baz
+"""
+        mock_open = mock.mock_open(read_data=read_data)
+        mock_open.side_effect = [IOError, DEFAULT]
+        with mock.patch("builtins.open", mock_open):
+            with self.assertRaises(ValueError):
+                read_config()
+
+    def test_read_config_not_found(self):
+        """empty config on no file found"""
+        mock_open = mock.mock_open()
+        mock_open.side_effect = IOError
+        with mock.patch("builtins.open", mock_open):
+            self.assertEqual({}, read_config())

volkswagencarnet/vw_connection.py

@@ -94,7 +94,7 @@ async def doLogin(self, tries: int = 1):
             if self._session_logged_in:
                 break
             _LOGGER.info("Something failed")
-            await asyncio.sleep(random * 5)
+            await asyncio.sleep(random() * 5)
 
         if not self._session_logged_in:
             return False
@@ -1026,7 +1026,6 @@ async def setClimater(self, vin, data, spin):
         except:
             self._session_headers.pop("X-securityToken", None)
             raise
-        return False
 
     async def setPreHeater(self, vin, data, spin):
         contType = None