Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/dttest/api/python/ge/ge/error.py: 96%

1#!/usr/bin/env python3 

2# -*- coding: utf-8 -*- 

3# ------------------------------------------------------------------- 

4# ----------------------------------------------------------------------------------------------------------- 

5# Copyright (c) 2026 Huawei Technologies Co., Ltd. 

6# This program is free software, you can redistribute it and/or modify it under the terms and conditions of 

7# CANN Open Software License Agreement Version 2.0 (the "License"). 

8# Please refer to the License for details. You may not use this file except in compliance with the License. 

9# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, 

10# INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. 

11# See LICENSE in the root of the software repository for the full text of the License. 

12# ----------------------------------------------------------------------------------------------------------- 

13 

14"""GE error helpers.""" 

15 

16from ge._capi.pygeapi_wrapper import geapi_lib 

17 

18 

19class GeError(RuntimeError): 

20 """Runtime error raised for GE C API failures.""" 

21 

22 def __init__( 

23 self, 

24 message: str, 

25 error_message: str = None, 

26 api_name: str = None, 

27 context: dict = None, 

28 ) -> None: 

29 super().__init__(message) 

30 self.error_message = error_message 

31 self.api_name = api_name 

32 self.context = context or {} 

33 

34 

35def _decode_c_string(value) -> str: 

36 if not value: 

37 return "" 

38 return value.decode("utf-8", errors="replace") 

39 

40 

41def get_ge_error_msg() -> str: 

42 return _decode_c_string(geapi_lib.GeApiWrapper_GEGetErrorMsg()) 

43 

44 

45def raise_ge_error(api_name: str, status: int = None, **context) -> None: 

46 del status 

47 error_message = get_ge_error_msg() 

48 context_text = ", ".join(f"{key}={value}" for key, value in context.items()) 

49 

50 message_parts = [f"{api_name} failed"] 

51 if context_text: 

52 message_parts.append(context_text) 

53 if error_message: 

54 message_parts.append(error_message) 

55 

56 raise GeError( 

57 "; ".join(message_parts), 

58 error_message=error_message, 

59 api_name=api_name, 

60 context=context, 

61 )