Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/dttest/build_st/python_tests/v1/st/test_v1_llm_types.py: 100%

1#!/usr/bin/env python3 

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

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

4# Copyright (c) 2025 Huawei Technologies Co., Ltd. 

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

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

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

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

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

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

11# ----------------------------------------------------------------------------------------------------------- 

12 

13import unittest 

14 

15from llm_datadist_v1 import ( 

16 CacheKeyByIdAndIndex, 

17 MemInfo, 

18 Memtype, 

19 TransferConfig, 

20 TransferWithCacheKeyConfig, 

21) 

22 

23 

24class MemInfoSt(unittest.TestCase): 

25 def setUp(self): 

26 print("Begin ", self._testMethodName) 

27 self._memoinfo = MemInfo(Memtype.MEM_TYPE_DEVICE, 0, 1) 

28 

29 def tearDown(self) -> None: 

30 print("End ", self._testMethodName) 

31 

32 def testMemType(self): 

33 self.assertEqual(self._memoinfo.mem_type, Memtype.MEM_TYPE_DEVICE) 

34 

35 def testAddr(self): 

36 self.assertEqual(self._memoinfo.addr, 0) 

37 

38 def testSize(self): 

39 self.assertEqual(self._memoinfo.size, 1) 

40 

41 

42class TransferWithCacheKeyConfigSt(unittest.TestCase): 

43 def setUp(self): 

44 print("Begin ", self._testMethodName) 

45 cache_key = CacheKeyByIdAndIndex(1, 1) 

46 src_layer_range = range(1, 5) 

47 dst_layer_range = range(1, 5) 

48 self._twckc = TransferWithCacheKeyConfig(cache_key, src_layer_range, dst_layer_range) 

49 

50 def tearDown(self) -> None: 

51 print("End ", self._testMethodName) 

52 

53 def testCacheKey(self): 

54 self.assertEqual(self._twckc.cache_key.cluster_id, 1) 

55 self._twckc.cache_key = CacheKeyByIdAndIndex(2, 1) 

56 self.assertEqual(self._twckc.cache_key.cluster_id, 2) 

57 

58 def testSrcLayerRange(self): 

59 self.assertEqual(self._twckc.src_layer_range.start, 1) 

60 self._twckc.src_layer_range = range(2, 5) 

61 self.assertEqual(self._twckc.src_layer_range.start, 2) 

62 

63 def testDstLayerRange(self): 

64 self.assertEqual(self._twckc.dst_layer_range.start, 1) 

65 self._twckc.dst_layer_range = range(2, 5) 

66 self.assertEqual(self._twckc.dst_layer_range.start, 2) 

67 

68 def testSrcBatchIndex(self): 

69 self.assertEqual(self._twckc.src_batch_index, 0) 

70 self._twckc.src_batch_index = 1 

71 self.assertEqual(self._twckc.src_batch_index, 1) 

72 

73 

74class TransferConfigSt(unittest.TestCase): 

75 def setUp(self): 

76 print("Begin ", self._testMethodName) 

77 dst_addrs = [1, 2, 3, 4] 

78 self._transfer_config = TransferConfig(1, dst_addrs, range(1, 5), 0) 

79 

80 def tearDown(self) -> None: 

81 print("End ", self._testMethodName) 

82 

83 def testDstClusterId(self): 

84 self._transfer_config.dst_cluster_id = 5 

85 self.assertEqual(self._transfer_config.dst_cluster_id, 5) 

86 

87 def testSrcBatchIndex(self): 

88 self._transfer_config.src_batch_index = 10 

89 self.assertEqual(self._transfer_config.src_batch_index, 10)