From 515498fa4e6f8dd7934cfb1a332796960498f00d Mon Sep 17 00:00:00 2001 From: Florian Bezannier Date: Mon, 16 Mar 2026 21:36:36 +0100 Subject: [PATCH] chore: remove assert --- .../psa/connected_car_api/rest.py | 7 +++- psa_car_controller/psa/oauth.py | 3 +- psa_car_controller/psa/otp/otp.py | 40 +++++++++---------- .../psacc/application/ecomix.py | 10 +++-- psa_car_controller/psacc/model/charge.py | 3 +- psa_car_controller/psacc/repository/db.py | 8 ++-- psa_car_controller/psacc/repository/trips.py | 4 +- psa_car_controller/web/tools/utils.py | 3 +- psa_car_controller/web/view/views.py | 24 +++++------ 9 files changed, 58 insertions(+), 44 deletions(-) diff --git a/psa_car_controller/psa/connected_car_api/rest.py b/psa_car_controller/psa/connected_car_api/rest.py index 99559bb..14bb22f 100644 --- a/psa_car_controller/psa/connected_car_api/rest.py +++ b/psa_car_controller/psa/connected_car_api/rest.py @@ -127,8 +127,11 @@ class RESTClientObject(object): (connection, read) timeouts. """ method = method.upper() - assert method in ['GET', 'HEAD', 'DELETE', 'POST', 'PUT', - 'PATCH', 'OPTIONS'] + allowed_methods = ['GET', 'HEAD', 'DELETE', 'POST', 'PUT', 'PATCH', 'OPTIONS'] + if method not in allowed_methods: + raise ValueError( + f"Invalid HTTP method: {method}. Must be one of {allowed_methods}" + ) if post_params and body: raise ValueError( diff --git a/psa_car_controller/psa/oauth.py b/psa_car_controller/psa/oauth.py index fbc7096..5949279 100644 --- a/psa_car_controller/psa/oauth.py +++ b/psa_car_controller/psa/oauth.py @@ -51,7 +51,8 @@ class OpenIdCredentialManager(CredentialManager): code_challenge=code_challenge, code_challenge_method="S256") def connect_with_code(self, code: str): - assert len(code) == 36, "Invalid code length" + if len(code) != 36: + raise ValueError(f"Invalid code length: {len(code)} (expected 36)") self._token_request({"grant_type": 'authorization_code', "code": code, "redirect_uri": self.redirect_uri, "code_verifier": self.code_verifier}, False) diff --git a/psa_car_controller/psa/otp/otp.py b/psa_car_controller/psa/otp/otp.py index 6221cb3..ddea9d5 100644 --- a/psa_car_controller/psa/otp/otp.py +++ b/psa_car_controller/psa/otp/otp.py @@ -210,14 +210,14 @@ class Otp: params.update(R) xml = self.request(params) if xml["err"] != "OK": - raise ConfigException(f"Error during activation: {xml}") + raise ConfigException(f"Error during activation finalize: {xml}") self.data.synchro(xml, self.generate_kma(self.codepin)) if self.mode == Otp.OTP_MODE: try: self.defi = str(xml["defi"]) except KeyError: - raise ConfigException from KeyError + raise ConfigException("Missing 'defi' in response") from KeyError if "J" in xml: logger.debug("Need another otp request") return Otp.OTP_TWICE @@ -228,7 +228,7 @@ class Otp: return Otp.OK if int(xml["ms_n"]) > 1: - raise NotImplementedError + raise NotImplementedError("Multiple ms_n not supported") ms_n = "0" self.challenge = xml["challenge"] @@ -251,6 +251,8 @@ class Otp: req_param.update({"id": self.data.iwid, "lastsync": self.data.iwTsync, "ms_n": 1}) req_param.update(self.get_r()) xml = self.request(req_param) + if xml["err"] != "OK": + raise ConfigException(f"Error during MS_MODE activation finalize: {xml}") self.data.synchro(xml, self.generate_kma(self.codepin)) return Otp.OK @@ -265,18 +267,17 @@ class Otp: def get_otp_code(self): self.mode = Otp.OTP_MODE otp_code = None - try: - if self.activation_start(): - res = self.activation_finalyze() - if res == Otp.OTP_TWICE: - self.mode = Otp.OTP_MODE - self.activation_start() - assert self.activation_finalyze() == Otp.OK - otp_code = self._get_otp_code() - assert otp_code is not None - logger.debug("otp code: %s", otp_code) - except AssertionError as e: - raise ConfigException("Can't get otp code") from e + if self.activation_start(): + res = self.activation_finalyze() + if res == Otp.OTP_TWICE: + self.mode = Otp.OTP_MODE + self.activation_start() + if self.activation_finalyze() != Otp.OK: + raise ConfigException("Can't get otp code (second attempt failed)") + otp_code = self._get_otp_code() + if otp_code is None: + raise ConfigException("Can't get otp code (code is None)") + logger.debug("otp code: %s", otp_code) return otp_code def __getstate__(self): @@ -332,8 +333,7 @@ def new_otp_session(smscode, codepin, old_otp_session: Otp = None, ): otp = Otp("bb8e981582b0f31353108fb020bead1c", device_id=old_otp_session.device_id) otp.smsCode = smscode otp.codepin = codepin - if otp.activation_start(): - otp.activation_finalyze() - save_otp(otp) - return otp - return None + otp.activation_start() + otp.activation_finalyze() + save_otp(otp) + return otp diff --git a/psa_car_controller/psacc/application/ecomix.py b/psa_car_controller/psacc/application/ecomix.py index f6f63d5..df589c0 100644 --- a/psa_car_controller/psacc/application/ecomix.py +++ b/psa_car_controller/psacc/application/ecomix.py @@ -64,7 +64,9 @@ class Ecomix: try: now = datetime.utcnow().replace(tzinfo=UTC) country_code = Ecomix.get_country(latitude, longitude, country_code_default) - assert country_code is not None + if country_code is None: + logger.warning("Can't find country for %s %s", latitude, longitude) + return False if country_code not in Ecomix._cache: Ecomix._cache[country_code] = [] elif len(Ecomix._cache[country_code]) > 0 and \ @@ -79,10 +81,12 @@ class Ecomix: data = res.json() value = data["carbonIntensity"] - assert isinstance(value, numbers.Number) + if not isinstance(value, numbers.Number): + logger.error("carbonIntensity invalid value: '%s'", value) + return False Ecomix._cache[country_code].append([now, value]) return True - except (AssertionError, NameError, KeyError): + except (NameError, KeyError): logger.debug("ecomix:", exc_info=True) return False else: diff --git a/psa_car_controller/psacc/model/charge.py b/psa_car_controller/psacc/model/charge.py index c87d762..e770c6f 100644 --- a/psa_car_controller/psacc/model/charge.py +++ b/psa_car_controller/psacc/model/charge.py @@ -16,7 +16,8 @@ class Charge: # pylint: disable=too-many-arguments,too-many-positional-arguments def __init__(self, start_at: datetime, stop_at: datetime = None, vin=None, start_level=None, end_level=None, co2=None, kw=None, price=None, charging_mode=None, mileage=None): - assert isinstance(start_at, datetime) + if not isinstance(start_at, datetime): + raise TypeError(f"start_at must be a datetime object, got {type(start_at)}") self.charging_mode: ChargingMode = ChargingMode(charging_mode) self.start_at = start_at self.stop_at = stop_at diff --git a/psa_car_controller/psacc/repository/db.py b/psa_car_controller/psacc/repository/db.py index 97b5d54..997a123 100644 --- a/psa_car_controller/psacc/repository/db.py +++ b/psa_car_controller/psacc/repository/db.py @@ -151,9 +151,11 @@ class Database: @staticmethod def get_db(db_file=None, force_new_conn=False) -> CustomSqliteConnection: - assert sqlite3.threadsafety == 3, \ - "SQLite is not in serialized mode (sqlite3.threadsafety != 3). " + \ - "Upgrade python to python3.11" + if sqlite3.threadsafety != 3: + raise RuntimeError( + "SQLite is not in serialized mode (sqlite3.threadsafety != 3). " + "Please upgrade python to python3.11 or ensure sqlite3 is compiled with threading support." + ) with Database.__thread_lock: if not Database.__conn or force_new_conn: if db_file is None: diff --git a/psa_car_controller/psacc/repository/trips.py b/psa_car_controller/psacc/repository/trips.py index 391cf5f..3d83491 100644 --- a/psa_car_controller/psacc/repository/trips.py +++ b/psa_car_controller/psacc/repository/trips.py @@ -64,7 +64,9 @@ class Trips(list): 'AND mileage IS NOT NULL AND Timestamp IS NOT NULL ORDER BY Timestamp', (vin,)).fetchall() if len(res) > 1: car = vehicles_list.get_car_by_vin(vin) - assert car is not None + if car is None: + logger.warning("Car with vin %s not found in vehicles list", vin) + continue trip_parser = TripParser(car) start = res[0] end = res[1] diff --git a/psa_car_controller/web/tools/utils.py b/psa_car_controller/web/tools/utils.py index 5630f69..9e5d8e9 100644 --- a/psa_car_controller/web/tools/utils.py +++ b/psa_car_controller/web/tools/utils.py @@ -72,7 +72,8 @@ def create_card(card: dict): def diff_dashtable(data, data_previous, row_id_name="row_id"): df, df_previous = DataFrame(data=data), DataFrame(data_previous) for _df in [df, df_previous]: - assert row_id_name in _df.columns + if row_id_name not in _df.columns: + raise ValueError(f"Column '{row_id_name}' not found in dataframes") _df = _df.set_index(row_id_name) mask = df.ne(df_previous) df_diff = df[mask].dropna(how="all", axis="columns").dropna(how="all", axis="rows") diff --git a/psa_car_controller/web/view/views.py b/psa_car_controller/web/view/views.py index c40cb91..67b0a76 100644 --- a/psa_car_controller/web/view/views.py +++ b/psa_car_controller/web/view/views.py @@ -199,26 +199,26 @@ def update_trips(): try: trips_by_vin = Trips.get_trips(Cars([car])) trips = trips_by_vin[car.vin] - assert len(trips) > 0 - min_date = trips[0].start_at - max_date = trips[-1].start_at - figures.get_figures(trips[0].car) - except (AssertionError, KeyError): + if len(trips) > 0: + min_date = trips[0].start_at + max_date = trips[-1].start_at + figures.get_figures(trips[0].car) + except (KeyError, IndexError): logger.debug("No trips yet") figures.get_figures(Car("vin", "vid", "brand")) - try: - chargings = Charging.get_chargings() - assert len(chargings) > 0 + figures.get_figures(Car("vin", "vid", "brand")) + chargings = Charging.get_chargings() + if len(chargings) == 0: + logger.debug("No chargings yet") + if min_date is None: + return + else: if min_date: min_date = min(min_date, chargings[0]["start_at"]) max_date = max(max_date, chargings[-1]["start_at"]) else: min_date = chargings[0]["start_at"] max_date = chargings[-1]["start_at"] - except AssertionError: - logger.debug("No chargings yet") - if min_date is None: - return # update for slider try: logger.debug("min_date:%s - max_date:%s", min_date, max_date)