Coverage for lobster/common/tool2.py: 29%

1# LOBSTER - Lightweight Open BMW Software Traceability Evidence Report 

2# Copyright (C) 2025 Bayerische Motoren Werke Aktiengesellschaft (BMW AG) 

3# 

4# This program is free software: you can redistribute it and/or modify 

5# it under the terms of the GNU Affero General Public License as 

6# published by the Free Software Foundation, either version 3 of the 

7# License, or (at your option) any later version. 

8# 

9# This program is distributed in the hope that it will be useful, but 

10# WITHOUT ANY WARRANTY; without even the implied warranty of 

11# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU 

12# Affero General Public License for more details. 

13# 

14# You should have received a copy of the GNU Affero General Public 

15# License along with this program. If not, see 

16# <https://www.gnu.org/licenses/>. 

17 

18import os 

19 

20from abc import ABCMeta 

21from typing import Iterable, List, Optional, Type, Union 

22from lobster.common.errors import Message_Handler 

23from lobster.common.items import Requirement, Implementation, Activity 

24from lobster.common.io import lobster_write 

25from lobster.common.meta_data_tool_base import MetaDataToolBase 

26from lobster.common.tool2_config import Config 

27from lobster.common.file_collector import FileCollector 

28 

29 

30def read_commented_file(file: str) -> List[str]: 

31 """Reads lines from a file, ignoring comments and empty lines. 

32 Returns all lines (without comments) as a list of strings. 

33 """ 

34 # lobster-trace: req.Lines_In_Inputs_File 

35 with open(file, "r", encoding="UTF-8") as fd: 

36 return select_non_comment_parts(fd.readlines()) 

37 

38 

39def select_non_comment_parts(text_list: Iterable[str]) -> List[str]: 

40 """Selects non-comment parts from a list of strings. 

41 

42 Returns the input list where each entry is stripped of comments, as well as 

43 leading and trailing whitespace, and empty lines are removed. 

44 """ 

45 return list( 

46 filter( 

47 None, 

48 [line.split("#", 1)[0].strip() for line in text_list], 

49 ), 

50 ) 

51 

52 

53def combine_all_inputs( 

54 config: Config, 

55 dir_or_files: Optional[List[str]], 

56) -> List[str]: 

57 """Combines all input sources into a single list.""" 

58 inputs = [] 

59 if config.inputs: 

60 inputs.extend(config.inputs) 

61 if config.inputs_from_file: 

62 inputs.extend(read_commented_file(config.inputs_from_file)) 

63 if dir_or_files: 

64 inputs.extend(dir_or_files) 

65 return inputs 

66 

67 

68def create_worklist( 

69 config: Config, 

70 dir_or_files: Optional[List[str]], 

71) -> List[str]: 

72 """Generates the exact list of files to work on. Directories are iterated 

73 recursively and their files are collected, if they match the extensions. 

74 Subdirectories are iterated if they do not match the exclude patterns. 

75 """ 

76 inputs = combine_all_inputs(config, dir_or_files) 

77 file_collector = FileCollector(config.extensions, config.exclude_patterns) 

78 for item in inputs: 

79 if os.path.isfile(item): 

80 file_collector.add_file(item, throw_on_mismatch=True) 

81 elif os.path.isdir(item): 

82 file_collector.add_dir_recursively(item) 

83 else: 

84 raise ValueError(f"{item} is not a file or directory") 

85 return file_collector.files 

86 

87 

88class MultiFileInputTool(MetaDataToolBase, metaclass=ABCMeta): 

89 """This class serves as base class for tools that process multiple input files.""" 

90 

91 def __init__( 

92 self, 

93 name: str, 

94 description: str, 

95 extensions: Iterable[str], 

96 official: bool, 

97 ): 

98 super().__init__( 

99 name=name, 

100 description=description, 

101 official=official, 

102 ) 

103 self._extensions = [f".{extension}" for extension in (sorted(extensions))] 

104 self._exclude_pattern = [] 

105 self._mh = Message_Handler() 

106 self._config = None 

107 

108 self._argument_parser.add_argument( 

109 "--out", 

110 default=f'{self.name}.lobster', 

111 help=f"Write output to the given file (default: {self.name}.lobster)", 

112 ) 

113 self._argument_parser.add_argument( 

114 "--config", 

115 default=None, 

116 help="Path to configuration file", 

117 required=True 

118 ) 

119 self._argument_parser.add_argument( 

120 "dir_or_files", 

121 nargs="*", 

122 metavar="DIR|FILE", 

123 help="Input directories or files" 

124 ) 

125 

126 def _write_output( 

127 self, 

128 schema: Union[Type[Requirement], Type[Implementation], Type[Activity]], 

129 out_file: str, 

130 items: List[Union[Activity, Implementation, Requirement]], 

131 ): 

132 with open(out_file, "w", encoding="UTF-8") as fd: 

133 lobster_write(fd, schema, self.name, items) 

134 print(f"{self.name}: wrote {len(items)} items to {out_file}")