From cb90ed017e247e818fa176605d66fee986cb8657 Mon Sep 17 00:00:00 2001 From: magikh0e <35289504+magikh0e@users.noreply.github.com> Date: Fri, 5 Jun 2026 18:52:24 -1000 Subject: [PATCH] Fix CI: ruff, black, and stale VehicleStatus tests The test workflow started running on master (#22) and surfaced three failures: - ruff UP031: cognito_timestamp used percent formatting; switched to an f-string with identical output. - black: reformatted device_srp.py, auth.py, client.py and the device tests to satisfy black --check (client.py was never black-compliant; CI just never ran black on master before). - pytest: test_from_dict_full and test_get_vehicle_status_success fed a flat dict the model never reads (VehicleStatus.from_dict parses the nested last_known_state.controller), so they asserted against unparsed values. Rewrote both to the real nested payload and the values the model actually produces. ruff check, black --check and pytest (79 passed) all pass locally. --- drone_mobile/auth.py | 16 ++++------------ drone_mobile/client.py | 4 +--- drone_mobile/device_srp.py | 30 +++++++++++++++++++----------- tests/test_client.py | 21 +++++++++++++++------ tests/test_device_remembering.py | 18 +++++++++++------- tests/test_device_srp.py | 8 ++------ tests/test_models.py | 26 +++++++++++++------------- 7 files changed, 65 insertions(+), 58 deletions(-) diff --git a/drone_mobile/auth.py b/drone_mobile/auth.py index 3b6ed45..0bc5ac6 100644 --- a/drone_mobile/auth.py +++ b/drone_mobile/auth.py @@ -399,9 +399,7 @@ def _respond_to_device_srp(self, challenge: dict) -> AuthToken: ) device_key = device["DeviceKey"] - username = ( - challenge.get("ChallengeParameters", {}).get("USERNAME") or self.username - ) + username = challenge.get("ChallengeParameters", {}).get("USERNAME") or self.username srp = DeviceSRP(device["DeviceGroupKey"], device_key, device["DevicePassword"]) try: @@ -550,9 +548,7 @@ def _cognito_request(self, target: str, payload: dict) -> dict: ) return response.json() if response.content else {} - def _remember_device( - self, access_token: str, device_key: str, device_group_key: str - ) -> None: + def _remember_device(self, access_token: str, device_key: str, device_group_key: str) -> None: """Confirm and remember a newly-issued Cognito device. Replicates the app's "Don't ask again on this device": generate a device @@ -564,9 +560,7 @@ def _remember_device( if existing and existing.get("DeviceKey") == device_key: return - device_password, verifier_config = generate_device_verifier( - device_group_key, device_key - ) + device_password, verifier_config = generate_device_verifier(device_group_key, device_key) self._cognito_request( "ConfirmDevice", { @@ -654,9 +648,7 @@ def _parse_auth_response(self, response: dict) -> AuthToken: ndm.get("DeviceGroupKey", ""), ) except Exception as e: # noqa: BLE001 - never block login on this - _LOGGER.warning( - "Could not remember device (will rely on refresh token): %s", e - ) + _LOGGER.warning("Could not remember device (will rely on refresh token): %s", e) expires_in = auth_result["ExpiresIn"] expires_at = datetime.now() + timedelta(seconds=expires_in - TOKEN_EXPIRY_MARGIN) diff --git a/drone_mobile/client.py b/drone_mobile/client.py index 0c9029f..5540882 100644 --- a/drone_mobile/client.py +++ b/drone_mobile/client.py @@ -369,9 +369,7 @@ def set_features( self.auth.authenticate(force_refresh=True) return self.set_features(vehicle_id, features, _retry=False) elif response.status_code == 429: - raise RateLimitError( - "API rate limit exceeded", response.status_code, response.json() - ) + raise RateLimitError("API rate limit exceeded", response.status_code, response.json()) else: raise APIError( f"Failed to update features: {response.text}", diff --git a/drone_mobile/device_srp.py b/drone_mobile/device_srp.py index 5805e2b..51ba573 100644 --- a/drone_mobile/device_srp.py +++ b/drone_mobile/device_srp.py @@ -72,9 +72,7 @@ def generate_device_verifier(device_group_key: str, device_key: str): verifier_hex = _pad_hex(pow(_G, x, _BIG_N)) return device_password, { - "PasswordVerifier": base64.standard_b64encode( - bytes.fromhex(verifier_hex) - ).decode("utf-8"), + "PasswordVerifier": base64.standard_b64encode(bytes.fromhex(verifier_hex)).decode("utf-8"), "Salt": base64.standard_b64encode(bytes.fromhex(salt_hex)).decode("utf-8"), } @@ -112,13 +110,23 @@ def cognito_timestamp() -> str: """ days = ("Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun") months = ( - "Jan", "Feb", "Mar", "Apr", "May", "Jun", - "Jul", "Aug", "Sep", "Oct", "Nov", "Dec", + "Jan", + "Feb", + "Mar", + "Apr", + "May", + "Jun", + "Jul", + "Aug", + "Sep", + "Oct", + "Nov", + "Dec", ) now = datetime.datetime.now(datetime.timezone.utc) - return "%s %s %d %02d:%02d:%02d UTC %d" % ( - days[now.weekday()], months[now.month - 1], now.day, - now.hour, now.minute, now.second, now.year, + return ( + f"{days[now.weekday()]} {months[now.month - 1]} {now.day} " + f"{now.hour:02d}:{now.minute:02d}:{now.second:02d} UTC {now.year}" ) @@ -177,6 +185,6 @@ def process_challenge( + base64.standard_b64decode(secret_block) + timestamp.encode("utf-8") ) - return base64.standard_b64encode( - hmac.new(key, message, hashlib.sha256).digest() - ).decode("utf-8") + return base64.standard_b64encode(hmac.new(key, message, hashlib.sha256).digest()).decode( + "utf-8" + ) diff --git a/tests/test_client.py b/tests/test_client.py index 261846a..b64ce20 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -166,11 +166,19 @@ def test_get_vehicle_status_success(self, mock_get, client, mock_auth): mock_response = Mock() mock_response.status_code = 200 mock_response.json.return_value = { - "vehicle_id": "123", - "device_key": "device_123", - "is_running": True, - "is_locked": False, - "battery_percent": 85, + "results": [ + { + "id": "123", + "device_key": "device_123", + "last_known_state": { + "controller": { + "engine_on": True, + "armed": False, + "main_battery_voltage": 12.6, + }, + }, + } + ] } mock_get.return_value = mock_response @@ -178,7 +186,8 @@ def test_get_vehicle_status_success(self, mock_get, client, mock_auth): assert status.vehicle_id == "123" assert status.is_running is True - assert status.battery_percent == 85 + assert status.is_locked is False + assert status.battery_voltage == 12.6 @patch("drone_mobile.client.requests.Session.post") def test_send_command_success(self, mock_post, client, mock_auth): diff --git a/tests/test_device_remembering.py b/tests/test_device_remembering.py index bce47a3..51f5702 100644 --- a/tests/test_device_remembering.py +++ b/tests/test_device_remembering.py @@ -112,12 +112,14 @@ def test_mfa_username_falls_back_to_email(mock_post, tmp_path, plain_tokens): @patch("drone_mobile.auth.requests.post") -def test_device_confirmed_and_remembered(mock_post, tmp_path, totp_challenge_sub, tokens_with_device): +def test_device_confirmed_and_remembered( + mock_post, tmp_path, totp_challenge_sub, tokens_with_device +): mock_post.side_effect = [ - _resp(200, totp_challenge_sub), # InitiateAuth -> challenge - _resp(200, tokens_with_device), # RespondToAuthChallenge -> tokens + device + _resp(200, totp_challenge_sub), # InitiateAuth -> challenge + _resp(200, tokens_with_device), # RespondToAuthChallenge -> tokens + device _resp(200, {"UserConfirmationNecessary": True}), # ConfirmDevice - _resp(200, {}), # UpdateDeviceStatus + _resp(200, {}), # UpdateDeviceStatus ] auth = AuthenticationManager( "user@example.com", "pw", token_dir=tmp_path, mfa_callback=lambda _c: "123456" @@ -144,7 +146,9 @@ def test_device_confirmed_and_remembered(mock_post, tmp_path, totp_challenge_sub @patch("drone_mobile.auth.requests.post") -def test_confirm_failure_does_not_break_login(mock_post, tmp_path, totp_challenge_sub, tokens_with_device): +def test_confirm_failure_does_not_break_login( + mock_post, tmp_path, totp_challenge_sub, tokens_with_device +): """A ConfirmDevice rejection is swallowed: the user still gets logged in.""" mock_post.side_effect = [ _resp(200, totp_challenge_sub), @@ -213,9 +217,9 @@ def test_device_srp_login_skips_mfa(mock_post, tmp_path, plain_tokens): }, } mock_post.side_effect = [ - _resp(200, device_srp), # InitiateAuth -> DEVICE_SRP_AUTH + _resp(200, device_srp), # InitiateAuth -> DEVICE_SRP_AUTH _resp(200, password_verifier), # DEVICE_SRP_AUTH -> DEVICE_PASSWORD_VERIFIER - _resp(200, plain_tokens), # DEVICE_PASSWORD_VERIFIER -> tokens + _resp(200, plain_tokens), # DEVICE_PASSWORD_VERIFIER -> tokens ] token = auth.authenticate(force_refresh=True) diff --git a/tests/test_device_srp.py b/tests/test_device_srp.py index f58c660..631c4c5 100644 --- a/tests/test_device_srp.py +++ b/tests/test_device_srp.py @@ -106,16 +106,12 @@ def test_roundtrip_matches_server(self): secret_block = base64.standard_b64encode(os.urandom(64)).decode() timestamp = cognito_timestamp() - signature = srp.process_challenge( - format(big_b, "x"), salt_hex, secret_block, timestamp - ) + signature = srp.process_challenge(format(big_b, "x"), salt_hex, secret_block, timestamp) # 4. Server independently derives S = (A * v^u)^b and the signature. u = int(_hex_hash(_pad_hex(big_a) + _pad_hex(big_b)), 16) s_server = pow((big_a * pow(verifier, u, _BIG_N)) % _BIG_N, b, _BIG_N) - key_server = _hkdf( - bytes.fromhex(_pad_hex(s_server)), bytes.fromhex(_pad_hex(u)) - ) + key_server = _hkdf(bytes.fromhex(_pad_hex(s_server)), bytes.fromhex(_pad_hex(u))) message = ( DGK.encode() + DK.encode() diff --git a/tests/test_models.py b/tests/test_models.py index 365d062..0cc5b79 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -67,32 +67,32 @@ def test_create_vehicle_status(self): assert status.battery_percent == 85 def test_from_dict_full(self): - """Test creating VehicleStatus from full dictionary.""" + """Test creating VehicleStatus from a full last_known_state payload.""" data = { "vehicle_id": "123", "device_key": "device_123", - "is_running": True, - "is_locked": False, - "battery_voltage": 12.6, - "battery_percent": 85, - "odometer": 15000.5, - "fuel_level": 75, - "interior_temperature": 72, - "exterior_temperature": 65, - "location": { + "last_known_state": { + "controller": { + "engine_on": True, + "armed": False, + "main_battery_voltage": 12.6, + "current_temperature": 72, + }, + "mileage": 15000.5, "latitude": 37.7749, "longitude": -122.4194, + "timestamp": "2024-01-01T12:00:00", }, - "last_updated": "2024-01-01T12:00:00", } status = VehicleStatus.from_dict(data) assert status.vehicle_id == "123" assert status.is_running is True - assert status.battery_percent == 85 + assert status.is_locked is False + assert status.battery_voltage == 12.6 assert status.odometer == 15000.5 - assert status.fuel_level == 75 + assert status.interior_temperature == 72 assert status.location is not None assert status.location.latitude == 37.7749