fix test

tests/test_config.py

@@ -0,0 +1,50 @@
+import os
+import unittest
+from configparser import ConfigParser
+from unittest.mock import patch, Mock, MagicMock
+
+from psa_car_controller.common.mylogger import my_logger
+
+from psa_car_controller.psacc.repository.config_repository import ConfigRepository
+
+
+class TestUnit(unittest.TestCase):
+    def test_read_config(self):
+        from psa_car_controller.psacc.repository.config_repository import ConfigRepository
+        config = ConfigRepository.read_config()
+
+    @patch("psa_car_controller.psacc.repository.config_repository.ConfigRepository._write")
+    @patch("psa_car_controller.psacc.repository.config_repository.ConfigRepository._read_file")
+    def test_read_old_config(self, mock_read, mock_write):
+        old_config_str = """[General]
+currency = €
+
+[Electricity config]
+day price = 42
+night price = 0.1
+night hour start = 1h00
+night hour end = 4h42
+
+"""
+        mock_read.return_value = old_config_str
+        conf = ConfigRepository.read_config()
+        conf.write_config(conf)
+        config_written: ConfigParser = mock_write.call_args_list[0][0][1]
+        self.assertEqual(config_written["General"]["currency"].value, "€")
+        self.assertEqual(config_written["Electricity config"]["day price"].value, 42)
+        print(config_written)
+
+    def test_read_non_existent_config(self):
+        from psa_car_controller.psacc.repository.config_repository import ConfigRepository
+        self.assertRaises(FileNotFoundError, lambda: ConfigRepository._read_file("nonexistent.ini"))
+
+    def test_read_config_non_existent_config(self):
+        from psa_car_controller.psacc.repository.config_repository import ConfigRepository
+        result = ConfigRepository.read_config("nonexistent.ini")
+        expected_result = ConfigRepository.config_file_to_dto(ConfigRepository.get_default_config())
+        self.assertEqual(result, expected_result)
+
+
+if __name__ == '__main__':
+    my_logger(handler_level=os.environ.get("DEBUG_LEVEL", 20))
+    unittest.main()

tests/test_unit.py

@@ -0,0 +1,309 @@
+# flake8: noqa
+import json
+import os
+import unittest
+from datetime import datetime, timedelta
+from unittest.mock import MagicMock
+
+import reverse_geocode
+from pytz import UTC
+
+from psa_car_controller import psa
+from psa_car_controller.common import utils
+from psa_car_controller.common.mylogger import my_logger
+
+from psa_car_controller.common.utils import parse_hour, RateLimitException, rate_limit
+from psa_car_controller.psa.otp.otp import load_otp, save_otp
+from psa_car_controller.psa.setup.app_decoder import get_content_from_apk, GITHUB_USER, GITHUB_REPO
+from psa_car_controller.psa.setup.github import github_file_need_to_be_downloaded
+from psa_car_controller.psa.connected_car_api import ApiClient
+from psa_car_controller.psacc.application.charge_control import ChargeControls
+from psa_car_controller.psacc.application.charging import Charging
+from psa_car_controller.psacc.application.ecomix import Ecomix
+from psa_car_controller.psacc.application.psa_client import PSAClient
+from psa_car_controller.psacc.model.car import Car, Cars
+from psa_car_controller.psacc.repository import config_repository
+from psa_car_controller.psacc.repository.car_model import CarModelRepository
+from psa_car_controller.psacc.repository.config_repository import ConfigRepository
+from psa_car_controller.psacc.repository.db import Database
+from psa_car_controller.psacc.repository.trips import Trips
+from psa_car_controller.psacc.utils.utils import get_temp
+from tests.utils import DATA_DIR, record_position, latitude, longitude, date0, date1, date2, date3, record_charging, \
+    vehicule_list, get_new_test_db, get_date, date4
+from psa_car_controller.web.figures import get_figures, get_battery_curve_fig, get_altitude_fig
+from deepdiff import DeepDiff
+
+
+def compare_dict(result, expected):
+    diff = DeepDiff(expected, result)
+    if diff != {}:
+        raise AssertionError(str(diff))
+    return True
+
+
+dummy_value = 0
+
+
+def callback_test():
+    global dummy_value
+    dummy_value += 1
+
+
+class TestUnit(unittest.TestCase):
+    def __init__(self, methodName='runTest'):
+        super().__init__(methodName)
+        self.test_online = os.environ.get("TEST_ONLINE", "0") == "1"
+        self.vehicule_list = vehicule_list
+
+    def test_car(self):
+        car1 = Car("VRAAAAAAA", "1sdfdksnfk222", "Peugeot", "208", 46, 0)
+        car2 = Car("VR3UHZKX", "1sdfdksnfk222", "Peugeot")
+        cars = Cars([car1, car2])
+        cars.save_cars(name=DATA_DIR + "test_car.json")
+        Cars.load_cars(name=DATA_DIR + "test_car.json")
+
+    def test_otp_config(self):
+        otp_config = load_otp(filename=DATA_DIR + "otp_test.bin")
+        assert otp_config is not None
+        save_otp(otp_config, filename=DATA_DIR + "otp_test2.bin")
+
+    def test_mypsacc(self):
+        if self.test_online:
+            myp = PSAClient.load_config("config.json")
+            myp.refresh_token()
+            myp.get_vehicles()
+            car = myp.vehicles_list[0]
+            myp.abrp.abrp_enable_vin.add(car.vin)
+            myp.get_vehicle_info(myp.vehicles_list[0].vin)
+            myp.abrp.call(car, 22.1)
+            assert isinstance(get_temp(str(latitude), str(longitude), myp.weather_api), float)
+
+    def test_car_model(self):
+        assert CarModelRepository().find_model_by_vin("VR3UHZKXZL").name == "e-208"
+        assert CarModelRepository().find_model_by_vin("VR3UKZKXZM").name == "e-2008"
+        assert CarModelRepository().find_model_by_vin("VXKUHZKXZL").name == "corsa-e"
+
+    def test_c02_signal_cache(self):
+        start = datetime.utcnow().replace(tzinfo=UTC) - timedelta(minutes=30)
+        end = datetime.utcnow().replace(tzinfo=UTC)
+        Ecomix._cache = {'FR': [[start - timedelta(days=1), 100],
+                                [start + timedelta(minutes=1), 10],
+                                [start + timedelta(minutes=2), 20],
+                                [start + timedelta(minutes=3), 30]]}
+        assert Ecomix.get_co2_from_signal_cache(start, end, "FR") == 20
+        Ecomix.clean_cache()
+        self.assertEqual(3, len(Ecomix._cache["FR"]))
+
+    def test_c02_signal(self):
+        if self.test_online:
+            key = "d186c74bfbcd1da8"
+            Ecomix.co2_signal_key = key
+            def_country = "FR"
+            Ecomix.get_data_from_co2_signal(latitude, longitude, def_country)
+            res = Ecomix.get_co2_from_signal_cache(datetime.utcnow().replace(tzinfo=UTC) - timedelta(minutes=5),
+                                                   datetime.now(), def_country)
+            assert isinstance(res, float)
+
+    def test_charge_control(self):
+        charge_control = ChargeControls()
+        charge_control.file_name = "test_charge_control.json"
+        charge_control.save_config(force=True)
+
+    def test_battery_curve(self):
+        vin = self.vehicule_list[0].vin
+        get_new_test_db()
+        record_charging()
+        charge = Database.get_last_charge(vin)
+        self.assertEqual(charge.stop_at, date4)
+        res = Database.get_battery_curve(Database.get_db(), date0, date4, vin)
+        self.assertEqual(4, len(res))
+
+    def test_set_charge_price(self):
+        vin = self.vehicule_list[0].vin
+        get_new_test_db()
+        record_charging()
+        charge = Database.get_last_charge(vin)
+        charge.price = 42
+        conn = Database.get_db()
+        Database.set_chargings_price(conn, charge)
+        conn.close()
+        res = Database.get_charge(vin, charge.start_at)
+        self.assertEqual(charge.price, res.price)
+
+    def test_sdk(self):
+        res = {
+            'lastPosition': {'type': 'Feature', 'geometry': {'type': 'Point', 'coordinates': [9.65457, 49.96119, 21]},
+                             'properties': {'updatedAt': '2021-03-29T05:16:10Z', 'heading': 126,
+                                            'type': 'Estimated'}}, 'preconditionning': {
+                'airConditioning': {'updatedAt': '2021-04-01T16:17:01Z', 'status': 'Disabled', 'programs': [
+                    {'enabled': False, 'slot': 1, 'recurrence': 'Daily', 'start': 'PT21H40M',
+                     'occurence': {'day': ['Sat']}}]}},
+            'energy': [{'updatedAt': '2021-02-23T22:29:03Z', 'type': 'Fuel', 'level': 0},
+                       {'updatedAt': '2021-04-01T16:17:01Z', 'type': 'Electric', 'level': 70, 'autonomy': 192,
+                        'charging': {'plugged': True, 'status': 'InProgress', 'remainingTime': 'PT0S',
+                                     'chargingRate': 20, 'chargingMode': 'Slow', 'nextDelayedTime': 'PT21H30M'}}],
+            'createdAt': '2021-04-01T16:17:01Z',
+            'battery': {'voltage': 99, 'current': 0, 'createdAt': '2021-04-01T16:17:01Z'},
+            'kinetic': {'createdAt': '2021-03-29T05:16:10Z', 'moving': False},
+            'privacy': {'createdAt': '2021-04-01T16:17:01Z', 'state': 'None'},
+            'service': {'type': 'Electric', 'updatedAt': '2021-02-23T21:10:29Z'}, '_links': {'self': {
+                'href': 'https://api.groupe-psa.com/connectedcar/v4/user/vehicles/myid/status'},
+                'vehicles': {
+                    'href': 'https://api.groupe-psa.com/connectedcar/v4/user/vehicles/myid'}},
+            'timed.odometer': {'createdAt': None, 'mileage': 1107.1}, 'updatedAt': '2021-04-01T16:17:01Z'}
+        api = ApiClient()
+        status: psa.connected_car_api.models.status.Status = api._ApiClient__deserialize(res, "Status")
+        geocode_res = reverse_geocode.search([(status.last_position.geometry.coordinates[:2])[::-1]])[0]
+        assert geocode_res["country_code"] == "DE"
+        get_new_test_db()
+        car = Car("XX", "vid", "Peugeot")
+        car.status = status
+        myp = PSAClient.load_config(DATA_DIR + "config.json")
+        myp.record_info(car)
+        assert "features" in json.loads(Database.get_recorded_position())
+        # electric should be first
+        assert car.status.energy[0].type == 'Electric'
+
+    def test_record_position_charging(self):
+        get_new_test_db()
+        config_repository.CONFIG_FILENAME = DATA_DIR + "config.ini"
+        car = self.vehicule_list[0]
+        record_position()
+        utils.get_positions = MagicMock(
+            return_value=[{'dataset': 'srtm30m', 'elevation': 40.0, 'location': {'lat': 47.2183, 'lng': -1.60362}}])
+        Database.add_altitude_to_db(Database.get_db())
+        data = json.loads(Database.get_recorded_position())
+        assert data["features"][1]["geometry"]["coordinates"] == [float(longitude), float(latitude)]
+        trips = Trips.get_trips(self.vehicule_list)[car.vin]
+        trip = trips[0]
+        map(trip.add_temperature, [10, 13, 15])
+        res = trip.get_info()
+        assert compare_dict(res, {'consumption_km': 24.21052631578947,
+                                  'start_at': date0,
+                                  'consumption_by_temp': None,
+                                  'positions': {'lat': [latitude], 'long': [longitude]},
+                                  'duration': 40.0, 'speed_average': 28.5, 'distance': 19.0, 'mileage': 30.0,
+                                  'altitude_diff': 2, 'id': 1, 'consumption': 4.6})
+
+        Charging.elec_price = ConfigRepository.read_config(DATA_DIR + "config.ini").Electricity_config
+        start_level = 40
+        end_level = 85
+        Charging.record_charging(car, "InProgress", date0, start_level, latitude, longitude, None, "slow", 20, 60)
+        Charging.record_charging(car, "InProgress", date1, 70, latitude, longitude, "FR", "slow", 20, 60)
+        Charging.record_charging(car, "InProgress", date1, 70, latitude, longitude, "FR", "slow", 20, 60)
+        Charging.record_charging(car, "InProgress", date2, 80, latitude, longitude, "FR", "slow", 20, 60)
+        Charging.record_charging(car, "Stopped", date3, end_level, latitude, longitude, "FR", "slow", 20, 60)
+        chargings = Charging.get_chargings()
+        co2 = chargings[0]["co2"]
+        assert isinstance(co2, float)
+        assert compare_dict(chargings, [{'start_at': date0,
+                                         'stop_at': date3,
+                                         'VIN': 'VR3UHZKX',
+                                         'start_level': 40,
+                                         'end_level': 85,
+                                         'co2': co2,
+                                         'kw': 20.7,
+                                         'price': 3.84,
+                                         'charging_mode': 'slow'}])
+        assert get_figures(car)
+        row = {"start_at": date0.strftime('%Y-%m-%dT%H:%M:%S.000Z'),
+               "stop_at": date3.strftime('%Y-%m-%dT%H:%M:%S.000Z'), "start_level": start_level, "end_level": end_level}
+        battery_curve_fix = get_battery_curve_fig(row, car)
+        assert battery_curve_fix is not None
+        assert get_altitude_fig(trip) is not None
+
+    def test_fuel_car(self):
+        get_new_test_db()
+        config_repository.CONFIG_FILENAME = DATA_DIR + "config.ini"
+        car = self.vehicule_list[1]
+        Database.record_position(None, car.vin, 11, latitude, longitude, 22, date0, 40, 30, False)
+        Database.record_position(None, car.vin, 20, latitude, longitude, 22, date1, 35, 29, False)
+        Database.record_position(None, car.vin, 30, latitude, longitude, 22, date2, 30, 28, False)
+        trips = Trips.get_trips(self.vehicule_list)
+        res = trips[car.vin].get_trips_as_dict()
+        assert compare_dict(res, [{'consumption_km': 6.947368421052632,
+                                   'start_at': date0,
+                                   'consumption_by_temp': None,
+                                   'positions': {'lat': [latitude],
+                                                 'long': [longitude]},
+                                   'duration': 40.0,
+                                   'speed_average': 28.5,
+                                   'distance': 19.0,
+                                   'mileage': 30.0,
+                                   'altitude_diff': 0,
+                                   'id': 1,
+                                   'consumption': 1.32,
+                                   'consumption_fuel_km': 4.53}])
+
+    def test_none_mileage(self):
+        get_new_test_db()
+        config_repository.CONFIG_FILENAME = DATA_DIR + "config.ini"
+        car = self.vehicule_list[1]
+        Database.record_position(None, car.vin, None, latitude, longitude, 22, get_date(1), 40, 30, False)
+        Database.record_position(None, car.vin, None, latitude, longitude, 22, get_date(2), 35, 29, False)
+        Database.record_position(None, car.vin, None, latitude, longitude, 22, get_date(3), 30, 28, False)
+        start = get_date(4)
+        end = get_date(6)
+        Database.record_position(None, car.vin, 11, latitude, longitude, 22, start, 40, 30, False)
+        Database.record_position(None, car.vin, 20, latitude, longitude, 22, get_date(5), 35, 29, False)
+        Database.record_position(None, car.vin, 30, latitude, longitude, 22, end, 30, 28, False)
+        trips = Trips.get_trips(self.vehicule_list)
+        res = trips[car.vin].get_trips_as_dict()
+        assert compare_dict(res, [{'consumption_km': 6.947368421052632,
+                                   'start_at': start,
+                                   'consumption_by_temp': None,
+                                   'positions': {'lat': [latitude],
+                                                 'long': [longitude]},
+                                   'duration': 120.0,
+                                   'speed_average': 9.5,
+                                   'distance': 19.0,
+                                   'mileage': 30.0,
+                                   'altitude_diff': 0,
+                                   'id': 1,
+                                   'consumption': 1.32,
+                                   'consumption_fuel_km': 4.53}])
+
+    def test_db_callback(self):
+        old_dummy_value = dummy_value
+        get_new_test_db()
+        Database.set_db_callback(callback_test)
+        assert old_dummy_value == dummy_value
+        Database.record_position(None, "xx", 11, latitude, longitude - 0.05, None, date0, 40, None, False)
+        assert old_dummy_value != dummy_value
+
+    def test_parse_hour(self):
+        expected_res = [[2, 0, 0], [3, 14, 0], [0, 0, 2], [0, 30, 0]]
+        assert expected_res == [parse_hour(h) for h in ["PT2H", "PT3H14", "PT2S", "PT30M"]]
+
+    def test_rate_limit(self):
+        @rate_limit(2, 10)
+        def test_fct():
+            pass
+
+        test_fct()
+        test_fct()
+        try:
+            test_fct()
+            raise Exception("It should have raise RateLimitException")
+        except RateLimitException:
+            pass
+
+    def test_parse_apk(self):
+        filename = "mypeugeot.apk"
+        try:
+            os.remove(filename)
+        except FileNotFoundError:
+            pass
+        assert get_content_from_apk(filename, "FR")
+        assert github_file_need_to_be_downloaded(GITHUB_USER, GITHUB_REPO, "", filename) is False
+
+    def test_file_need_to_be_updated(self):
+        filename = "mypeugeot.apk"
+        with open(filename, "w") as f:
+            f.write(" ")
+        assert github_file_need_to_be_downloaded(GITHUB_USER, GITHUB_REPO, "", filename) is True
+
+
+if __name__ == '__main__':
+    my_logger(handler_level=os.environ.get("DEBUG_LEVEL", 20))
+    unittest.main()

tests/utils.py

@@ -0,0 +1,56 @@
+# flake8: noqa
+
+import os
+from datetime import timedelta, datetime
+
+import pytz
+
+
+from psa_car_controller.psacc.application.charging import Charging
+from psa_car_controller.psacc.model.car import Cars, Car
+from psa_car_controller.psacc.repository.db import Database
+
+DATA_DIR = os.path.dirname(os.path.realpath(__file__)) + "/data/"
+latitude = 47.2183
+longitude = -1.55362
+date3 = datetime.utcnow().replace(2021, 3, 1, 12, 00, 00, 00, tzinfo=pytz.UTC)
+date2 = date3 - timedelta(minutes=20)
+date1 = date3 - timedelta(minutes=40)
+date0 = date3 - timedelta(minutes=60)
+date4 = date3 + timedelta(minutes=1)
+
+vehicule_list = Cars()
+vehicule_list.extend(
+    [Car("VR3UHZKX", "vid", "Peugeot"), Car("VXXXXX", "XXXX", "Peugeot", label="SUV 3008 Hybrid 225")])
+car = vehicule_list[0]
+DB_DIR = DATA_DIR + "tmp.db"
+
+
+def get_new_test_db():
+    try:
+        os.remove(DATA_DIR + "tmp.db")
+    except FileNotFoundError:
+        pass
+    Database.DEFAULT_DB_FILE = DB_DIR
+    Database.db_initialized = False
+    conn = Database.get_db()
+    return conn
+
+
+def record_position():
+    Database.record_position(None, car.vin, 11, latitude, longitude - 0.05, None, date0, 40, None, False)
+    Database.record_position(None, car.vin, 20, latitude, longitude, 32, date1, 35, None, False)
+    Database.record_position(None, car.vin, 30, latitude, longitude, 42, date2, 30, None, False)
+    Database.record_position(None, car.vin, None, latitude, longitude, 42, date2, 30, None, False)
+
+
+def record_charging():
+    Charging.record_charging(car, "InProgress", date0, 50, latitude, longitude, "FR", "slow", 20, 60)
+    Charging.record_charging(car, "InProgress", date1, 75, latitude, longitude, "FR", "slow", 20, 60)
+    Charging.record_charging(car, "InProgress", date2, 85, latitude, longitude, "FR", "slow", 20, 60)
+    Charging.record_charging(car, "InProgress", date3, 90, latitude, longitude, "FR", "slow", 20, 60)
+    Charging.record_charging(car, "Stopped", date4, 91, latitude, longitude, "FR", "slow", 20, 60)
+
+
+def get_date(offset):
+    return date3 + timedelta(minutes=60 * offset)