Coverage for rdak/errors.py: 100%

1# SPDX-License-Identifier: GPL-2.0-or-later 

2# © 2026, Ansgar 🙀 <ansgar@debian.org> 

3 

4"""Error handling for the rdak client. 

5 

6`CliError` carries a message and an exit code; `cli.main` reports it on 

7stderr and returns the code. gRPC failures are translated into `CliError` 

8with a friendly, status-specific message. 

9""" 

10 

11from collections.abc import Iterator 

12from contextlib import contextmanager 

13 

14import grpc 

15 

16 

17class CliError(Exception): 

18 """An error to report to the user, with a process exit code.""" 

19 

20 def __init__(self, message: str, *, exit_code: int = 1) -> None: 

21 super().__init__(message) 

22 self.exit_code = exit_code 

23 

24 

25_STATUS_MESSAGES = { 

26 grpc.StatusCode.UNAUTHENTICATED: "authentication failed", 

27 grpc.StatusCode.PERMISSION_DENIED: "permission denied", 

28 grpc.StatusCode.NOT_FOUND: "not found", 

29 grpc.StatusCode.FAILED_PRECONDITION: "request rejected by the server", 

30 grpc.StatusCode.INVALID_ARGUMENT: "invalid argument", 

31 grpc.StatusCode.UNAVAILABLE: "cannot reach the server", 

32} 

33 

34 

35def rpc_error_to_cli(err: grpc.RpcError) -> CliError: 

36 """Translate a gRPC error into a user-facing `CliError`.""" 

37 code = err.code() 

38 base = _STATUS_MESSAGES.get(code) 

39 if base is None: 

40 base = code.name.lower().replace("_", " ") if code is not None else "RPC failed" 

41 details = err.details() 

42 if details and details not in base: 

43 return CliError(f"{base}: {details}") 

44 return CliError(base) 

45 

46 

47@contextmanager 

48def rpc_errors() -> Iterator[None]: 

49 """Convert `grpc.RpcError` raised inside the block into `CliError`.""" 

50 try: 

51 yield 

52 except grpc.RpcError as e: 

53 raise rpc_error_to_cli(e) from e