Coverage for lobster/common/report.py: 50%

1#!/usr/bin/env python3 

2# 

3# LOBSTER - Lightweight Open BMW Software Traceability Evidence Report 

4# Copyright (C) 2023-2025 Bayerische Motoren Werke Aktiengesellschaft (BMW AG) 

5# 

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

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

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

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

10# 

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

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

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

14# Affero General Public License for more details. 

15# 

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

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

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

19import json 

20from collections import OrderedDict 

21from dataclasses import dataclass 

22 

23from lobster.common.level_definition import LevelDefinition 

24from lobster.common.items import Tracing_Status, Requirement, Implementation, Activity 

25from lobster.common.parser import load as load_config 

26from lobster.common.errors import LOBSTER_Error, Message_Handler 

27from lobster.common.io import lobster_read, ensure_output_directory 

28from lobster.common.location import File_Reference 

29 

30 

31@dataclass 

32class Coverage: 

33 level : str 

34 items : int 

35 ok : int 

36 coverage : None 

37 

38 

39class Report: 

40 def __init__(self): 

41 self.mh = Message_Handler() 

42 self.config = OrderedDict() 

43 self.items = {} 

44 self.coverage = {} 

45 self.custom_data = {} 

46 self.source_root = "" 

47 

48 def parse_config(self, filename): 

49 """ 

50 Function parses the lobster config file to generate a .lobster file. 

51 Parameters 

52 ---------- 

53 filename - configuration file 

54 

55 Returns - Nothing 

56 ------- 

57 

58 """ 

59 

60 try: 

61 self.config = load_config(self.mh, filename) 

62 except LOBSTER_Error as exc: 

63 self.mh.error(exc.location, exc.message) 

64 # Note: mh.error raises the exception internally, so we do not need to 

65 # re-raise it here. 

66 

67 # Load requested files 

68 for level in self.config: 

69 for source in self.config[level].source: 

70 lobster_read(self.mh, source["file"], level, self.items, 

71 source) 

72 

73 # Resolve references for items 

74 self.resolve_references_for_items() 

75 

76 # Compute status and items count 

77 self.compute_item_count_and_status() 

78 

79 # Compute coverage for items 

80 self.compute_coverage_for_items() 

81 

82 def resolve_references_for_items(self): 

83 for src_item in self.items.values(): 

84 while src_item.unresolved_references: 

85 dst_tag = src_item.unresolved_references.pop() 

86 if dst_tag.key() not in self.items: 

87 src_item.error(f"unknown tracing target {dst_tag.key()}") 

88 continue 

89 dst_item = self.items[dst_tag.key()] 

90 # TODO: Check if policy allows this link 

91 src_item.ref_up.append(dst_tag) 

92 dst_item.ref_down.append(src_item.tag) 

93 

94 # Check versions match, if specified 

95 if dst_tag.version is not None: 

96 if dst_item.tag.version is None: 

97 src_item.error( 

98 f"tracing destination {dst_tag.key()} is unversioned" 

99 ) 

100 elif dst_tag.version != dst_item.tag.version: 

101 msg = (f"tracing destination {dst_tag.key()} has version " 

102 f"{dst_item.tag.version} (expected {dst_tag.version})") 

103 src_item.error(msg) 

104 

105 def compute_coverage_for_items(self): 

106 for level_obj in self.coverage.values(): 

107 if level_obj.items == 0: 

108 level_obj.coverage = 0.0 

109 else: 

110 level_obj.coverage = float(level_obj.ok * 100) / float(level_obj.items) 

111 

112 def compute_item_count_and_status(self): 

113 for level in self.config: 

114 coverage = Coverage(level=level, items=0, ok=0, coverage=None) 

115 self.coverage.update({level: coverage}) 

116 for item in self.items.values(): 

117 item.determine_status(self.config, self.items) 

118 self.coverage[item.level].items += 1 

119 if item.tracing_status in (Tracing_Status.OK, 

120 Tracing_Status.JUSTIFIED): 

121 self.coverage[item.level].ok += 1 

122 

123 def write_report(self, filename): 

124 

125 levels = [] 

126 for level_config in self.config.values(): 

127 level = { 

128 "name" : level_config.name, 

129 "kind" : level_config.kind, 

130 "items" : [item.to_json() 

131 for item in self.items.values() 

132 if item.level == level_config.name], 

133 "coverage" : self.coverage[level_config.name].coverage 

134 } 

135 levels.append(level) 

136 

137 report = { 

138 "schema" : "lobster-report", 

139 "version" : 2, 

140 "generator" : "lobster_report", 

141 "levels" : levels, 

142 "policy" : {key: value.to_json() 

143 for key, value in self.config.items()}, 

144 "matrix" : [], 

145 } 

146 

147 ensure_output_directory(filename) 

148 with open(filename, "w", encoding="UTF-8") as fd: 

149 json.dump(report, fd, indent=2) 

150 fd.write("\n") 

151 

152 def load_report(self, filename): 

153 

154 loc = File_Reference(filename) 

155 

156 # Read and validate JSON 

157 with open(filename, encoding="UTF-8") as fd: 

158 try: 

159 data = json.load(fd) 

160 except json.decoder.JSONDecodeError as err: 

161 self.mh.error(File_Reference(filename, 

162 err.lineno, 

163 err.colno), 

164 err.msg) 

165 

166 # Validate basic structure 

167 self.validate_basic_structure_of_lobster_file(data, loc) 

168 

169 # Validate indicated schema 

170 self.validate_indicated_schema(data, loc) 

171 

172 # Validate and parse custom data 

173 self.parse_custom_data(data) 

174 

175 # Read in data 

176 self.compute_items_and_coverage_for_items(data) 

177 

178 def compute_items_and_coverage_for_items(self, data): 

179 """ 

180 Function calculates items and coverage for the items 

181 Parameters 

182 ---------- 

183 data - contents of lobster json file. 

184 

185 Returns - Nothing 

186 ------- 

187 

188 """ 

189 self.config = {key: LevelDefinition.from_json(value) 

190 for key, value in data["policy"].items()} 

191 for level in data["levels"]: 

192 if level["name"] not in self.config: 192 ↛ 193line 192 didn't jump to line 193 because the condition on line 192 was never true

193 raise KeyError(f"level '{level['name']}' not found in config") 

194 coverage = Coverage( 

195 level=level["name"], items=0, ok=0, coverage=level["coverage"] 

196 ) 

197 self.coverage.update({level["name"]: coverage}) 

198 

199 for item_data in level["items"]: 

200 if level["kind"] == "requirements": 200 ↛ 201line 200 didn't jump to line 201 because the condition on line 200 was never true

201 item = Requirement.from_json(level["name"], 

202 item_data, 

203 3) 

204 elif level["kind"] == "implementation": 204 ↛ 209line 204 didn't jump to line 209 because the condition on line 204 was always true

205 item = Implementation.from_json(level["name"], 

206 item_data, 

207 3) 

208 else: 

209 if level["kind"] != "activity": 

210 raise ValueError(f"unknown level kind '{level['kind']}'") 

211 item = Activity.from_json(level["name"], 

212 item_data, 

213 3) 

214 

215 self.items[item.tag.key()] = item 

216 self.coverage[item.level].items += 1 

217 if item.tracing_status in (Tracing_Status.OK, 

218 Tracing_Status.JUSTIFIED): 

219 self.coverage[item.level].ok += 1 

220 

221 def parse_custom_data(self, data): 

222 self.custom_data = data.get('custom_data', None) 

223 

224 def validate_indicated_schema(self, data, loc): 

225 """ 

226 Function validates the schema and version. 

227 Parameters 

228 ---------- 

229 data - contents of lobster json file. 

230 loc - location from where the error was raised. 

231 

232 Returns - Nothing 

233 ------- 

234 

235 """ 

236 supported_schema = { 

237 "lobster-report": {2}, 

238 } 

239 if data["schema"] not in supported_schema: 239 ↛ 240line 239 didn't jump to line 240 because the condition on line 239 was never true

240 self.mh.error(loc, f"unknown schema kind {data['schema']}") 

241 if data["version"] not in supported_schema[data["schema"]]: 241 ↛ 242line 241 didn't jump to line 242 because the condition on line 241 was never true

242 self.mh.error(loc, 

243 f"version {data['version']} for schema " 

244 f"{data['schema']} is not supported") 

245 

246 def validate_basic_structure_of_lobster_file(self, data, loc): 

247 """ 

248 Function validates the basic structure of lobster file. All the first level 

249 keys of the lobster json file are validated here. 

250 Parameters 

251 ---------- 

252 data - contents of lobster json file. 

253 loc - location from where the error was raised. 

254 

255 Returns - Nothing 

256 ------- 

257 

258 """ 

259 if not isinstance(data, dict): 259 ↛ 260line 259 didn't jump to line 260 because the condition on line 259 was never true

260 self.mh.error(loc, "parsed json is not an object") 

261 

262 rkey_dict = {"schema": str, "version": int, "generator": str, "levels": list, 

263 "policy": dict, "matrix": list} 

264 type_dict = {int: "an integer", str: "a string", list: "an array", 

265 dict: "an object"} 

266 for rkey, rvalue in rkey_dict.items(): 

267 if rkey not in data: 267 ↛ 268line 267 didn't jump to line 268 because the condition on line 267 was never true

268 self.mh.error(loc, f"required top-level key {rkey} not present") 

269 if not isinstance(data[rkey], rvalue): 269 ↛ 270line 269 didn't jump to line 270 because the condition on line 269 was never true

270 self.mh.error(loc, f"{rkey} is not {type_dict[rvalue]}.")