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

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 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 

47 def parse_config(self, filename): 

48 """ 

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

50 Parameters 

51 ---------- 

52 filename - configuration file 

53 

54 Returns - Nothing 

55 ------- 

56 

57 """ 

58 

59 # Load config 

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

61 

62 # Load requested files 

63 for level in self.config: 

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

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

66 source) 

67 

68 # Resolve references for items 

69 self.resolve_references_for_items() 

70 

71 # Compute status and items count 

72 self.compute_item_count_and_status() 

73 

74 # Compute coverage for items 

75 self.compute_coverage_for_items() 

76 

77 def resolve_references_for_items(self): 

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

79 while src_item.unresolved_references: 

80 dst_tag = src_item.unresolved_references.pop() 

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

82 src_item.error("unknown tracing target %s" % dst_tag.key()) 

83 continue 

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

85 # TODO: Check if policy allows this link 

86 src_item.ref_up.append(dst_tag) 

87 dst_item.ref_down.append(src_item.tag) 

88 

89 # Check versions match, if specified 

90 if dst_tag.version is not None: 

91 if dst_item.tag.version is None: 

92 src_item.error("tracing destination %s is unversioned" 

93 % dst_tag.key()) 

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

95 src_item.error("tracing destination %s has version %s" 

96 " (expected %s)" % 

97 (dst_tag.key(), 

98 dst_item.tag.version, 

99 dst_tag.version)) 

100 

101 def compute_coverage_for_items(self): 

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

103 if level_obj.items == 0: 

104 level_obj.coverage = 0.0 

105 else: 

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

107 

108 def compute_item_count_and_status(self): 

109 for level in self.config: 

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

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

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

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

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

115 if item.tracing_status in (Tracing_Status.OK, 

116 Tracing_Status.JUSTIFIED): 

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

118 

119 def write_report(self, filename): 

120 

121 levels = [] 

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

123 level = { 

124 "name" : level_config.name, 

125 "kind" : level_config.kind, 

126 "items" : [item.to_json() 

127 for item in self.items.values() 

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

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

130 } 

131 levels.append(level) 

132 

133 report = { 

134 "schema" : "lobster-report", 

135 "version" : 2, 

136 "generator" : "lobster_report", 

137 "levels" : levels, 

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

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

140 "matrix" : [], 

141 } 

142 

143 ensure_output_directory(filename) 

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

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

146 fd.write("\n") 

147 

148 def load_report(self, filename): 

149 

150 loc = File_Reference(filename) 

151 

152 # Read and validate JSON 

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

154 try: 

155 data = json.load(fd) 

156 except json.decoder.JSONDecodeError as err: 

157 self.mh.error(File_Reference(filename, 

158 err.lineno, 

159 err.colno), 

160 err.msg) 

161 

162 # Validate basic structure 

163 self.validate_basic_structure_of_lobster_file(data, loc) 

164 

165 # Validate indicated schema 

166 self.validate_indicated_schema(data, loc) 

167 

168 # Validate and parse custom data 

169 self.parse_custom_data(data) 

170 

171 # Read in data 

172 self.compute_items_and_coverage_for_items(data) 

173 

174 def compute_items_and_coverage_for_items(self, data): 

175 """ 

176 Function calculates items and coverage for the items 

177 Parameters 

178 ---------- 

179 data - contents of lobster json file. 

180 

181 Returns - Nothing 

182 ------- 

183 

184 """ 

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

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

187 for level in data["levels"]: 

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

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

190 coverage = Coverage( 

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

192 ) 

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

194 

195 for item_data in level["items"]: 

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

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

198 item_data, 

199 3) 

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

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

202 item_data, 

203 3) 

204 else: 

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

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

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

208 item_data, 

209 3) 

210 

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

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

213 if item.tracing_status in (Tracing_Status.OK, 

214 Tracing_Status.JUSTIFIED): 

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

216 

217 def parse_custom_data(self, data): 

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

219 

220 def validate_indicated_schema(self, data, loc): 

221 """ 

222 Function validates the schema and version. 

223 Parameters 

224 ---------- 

225 data - contents of lobster json file. 

226 loc - location from where the error was raised. 

227 

228 Returns - Nothing 

229 ------- 

230 

231 """ 

232 supported_schema = { 

233 "lobster-report": set([2]), 

234 } 

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

236 self.mh.error(loc, "unknown schema kind %s" % data["schema"]) 

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

238 self.mh.error(loc, 

239 "version %u for schema %s is not supported" % 

240 (data["version"], data["schema"])) 

241 

242 def validate_basic_structure_of_lobster_file(self, data, loc): 

243 """ 

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

245 keys of the lobster json file are validated here. 

246 Parameters 

247 ---------- 

248 data - contents of lobster json file. 

249 loc - location from where the error was raised. 

250 

251 Returns - Nothing 

252 ------- 

253 

254 """ 

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

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

257 

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

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

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

261 dict: "an object"} 

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

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

264 self.mh.error(loc, "required top-levelkey %s not present" % rkey) 

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

266 self.mh.error(loc, "%s is not %s." % (rkey, type_dict[rvalue]))