Coverage for lobster/common/items.py: 98%

223 statements  

« prev     ^ index     » next       coverage.py v7.17.3, created at 2026-11-13 15:05 +0000

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

19 

20from enum import Enum, auto 

21from abc import ABCMeta 

22from hashlib import sha1 

23from typing import Optional 

24 

25from lobster.common.location import Location 

26 

27 

28class Tracing_Tag: 

29 def __init__( 

30 self, 

31 namespace: str, 

32 tag: str, 

33 version: Optional[str] = None, 

34 ): 

35 assert isinstance(namespace, str) and " " not in namespace 

36 assert isinstance(tag, str) 

37 assert version is None or isinstance(version, (str, int)) 

38 assert not isinstance(version, str) or version != "None" 

39 

40 self.namespace = namespace 

41 self.tag = tag 

42 self.version = version 

43 self.hash_val = None 

44 

45 def __str__(self): 

46 rv = f"{self.namespace} {self.tag}" 

47 if self.version: 

48 rv += f"@{self.version}" 

49 return rv 

50 

51 def key(self) -> str: 

52 return self.namespace + " " + self.tag 

53 

54 def to_json(self) -> str: 

55 return str(self) 

56 

57 @classmethod 

58 def from_json(cls, json): 

59 assert isinstance(json, str) 

60 namespace, rest = json.split(" ", 1) 

61 return Tracing_Tag.from_text(namespace, rest) 

62 

63 @classmethod 

64 def from_text(cls, namespace, text): 

65 assert isinstance(namespace, str) 

66 assert isinstance(text, str) 

67 

68 if "@" in text: 

69 tag, version = text.split("@", 1) 

70 else: 

71 tag = text 

72 version = None 

73 return Tracing_Tag(namespace, tag, version) 

74 

75 def hash(self): 

76 if not self.hash_val: 

77 hfunc = sha1() 

78 hfunc.update(self.key().encode("UTF-8")) 

79 self.hash_val = hfunc.hexdigest() 

80 return self.hash_val 

81 

82 

83class Tracing_Status(Enum): 

84 OK = auto() 

85 PARTIAL = auto() 

86 MISSING = auto() 

87 JUSTIFIED = auto() 

88 ERROR = auto() 

89 

90 

91class Item(metaclass=ABCMeta): 

92 def __init__(self, tag: Tracing_Tag, location: Location, 

93 text: Optional[str] = None): 

94 assert isinstance(tag, Tracing_Tag) 

95 assert isinstance(location, Location) 

96 assert isinstance(text, str) or text is None 

97 

98 self.level = None 

99 self.tag = tag 

100 self.location = location 

101 self.name = tag.tag 

102 self.text = text 

103 

104 self.ref_up = [] 

105 self.ref_down = [] 

106 

107 self.unresolved_references_cache = set() 

108 self.unresolved_references = [] 

109 

110 self.messages = [] 

111 self.just_up = [] 

112 self.just_down = [] 

113 self.just_global = [] 

114 

115 self.tracing_status = None 

116 self.has_error = False 

117 

118 def set_level(self, level): 

119 assert isinstance(level, str) 

120 self.level = level 

121 

122 def error(self, message: str): 

123 assert isinstance(message, str) 

124 self.messages.append(message) 

125 self.has_error = True 

126 

127 def add_tracing_target(self, target: Tracing_Tag): 

128 assert isinstance(target, Tracing_Tag) 

129 if target.key() in self.unresolved_references_cache: 129 ↛ 130line 129 didn't jump to line 130 because the condition on line 129 was never true

130 return 

131 

132 self.unresolved_references.append(target) 

133 self.unresolved_references_cache.add(target.key()) 

134 

135 def perform_source_checks(self, source_info): 

136 assert isinstance(source_info, dict) 

137 

138 def determine_status(self, config, stab): 

139 assert self.level in config 

140 assert self.tag.key() in stab 

141 

142 level = config[self.level] 

143 

144 has_up_ref = len(self.ref_up) > 0 

145 has_just_up = len(self.just_up) > 0 or len(self.just_global) > 0 

146 has_just_down = len(self.just_down) > 0 or len(self.just_global) > 0 

147 has_init_errors = len(self.messages) > 0 

148 

149 # Check up references 

150 ok_up = True 

151 if level.needs_tracing_up and (not has_up_ref) and (not has_just_up): 

152 ok_up = False 

153 self.messages.append("missing up reference") 

154 

155 # Check set of down references 

156 ok_down = True 

157 if level.needs_tracing_down: 

158 has_trace = {name : False 

159 for name in config 

160 if self.level in config[name].traces} 

161 for ref in self.ref_down: 

162 has_trace[stab[ref.key()].level] = True 

163 for chain in level.breakdown_requirements: 

164 if not any(has_trace[src] for src in chain) and \ 

165 not has_just_down: 

166 ok_down = False 

167 chain_str = " or ".join(sorted(chain)) 

168 self.messages.append(f"missing reference to {chain_str}") 

169 

170 # Set status 

171 if self.has_error: 

172 self.tracing_status = Tracing_Status.MISSING 

173 elif ok_up and ok_down: 

174 if has_just_up or has_just_down: 

175 self.tracing_status = Tracing_Status.JUSTIFIED 

176 else: 

177 self.tracing_status = Tracing_Status.OK 

178 elif (ok_up or ok_down) and \ 

179 level.needs_tracing_up and \ 

180 level.needs_tracing_down: 

181 self.tracing_status = Tracing_Status.PARTIAL 

182 else: 

183 self.tracing_status = Tracing_Status.MISSING 

184 

185 # Overwrite status if there are initial errors 

186 if self.tracing_status == Tracing_Status.OK and has_init_errors: 

187 self.tracing_status = Tracing_Status.PARTIAL 

188 

189 def additional_data_from_json(self, level, data, schema_version): 

190 assert isinstance(level, str) 

191 assert isinstance(data, dict) 

192 assert schema_version >= 3 

193 

194 self.set_level(level) 

195 for ref in data.get("refs", []): 

196 self.add_tracing_target(Tracing_Tag.from_json(ref)) 

197 self.ref_up = [Tracing_Tag.from_json(ref) 

198 for ref in data.get("ref_up", [])] 

199 self.ref_down = [Tracing_Tag.from_json(ref) 

200 for ref in data.get("ref_down", [])] 

201 self.messages = data.get("messages", []) 

202 self.just_up = data.get("just_up", []) 

203 self.just_down = data.get("just_down", []) 

204 self.just_global = data.get("just_global", []) 

205 if "tracing_status" in data: 

206 self.tracing_status = Tracing_Status[data["tracing_status"]] 

207 

208 def to_json(self): 

209 rv = { 

210 "tag" : self.tag.to_json(), 

211 "location" : self.location.to_json(), 

212 "name" : self.name, 

213 "messages" : self.messages, 

214 "just_up" : self.just_up, 

215 "just_down" : self.just_down, 

216 "just_global" : self.just_global, 

217 } 

218 if self.unresolved_references: 

219 rv["refs"] = [tag.to_json() 

220 for tag in self.unresolved_references] 

221 if self.ref_up or self.ref_down: 

222 rv["ref_up"] = [tag.to_json() for tag in self.ref_up] 

223 rv["ref_down"] = [tag.to_json() for tag in self.ref_down] 

224 if self.tracing_status: 

225 rv["tracing_status"] = self.tracing_status.name 

226 return rv 

227 

228 

229class Requirement(Item): 

230 def __init__( 

231 self, 

232 tag: Tracing_Tag, 

233 location: Location, 

234 framework: str, 

235 kind: str, 

236 name: str, 

237 text: Optional[str] = None, 

238 status: Optional[str] = None, 

239 ): 

240 super().__init__(tag, location, text=text) 

241 assert isinstance(framework, str) 

242 assert isinstance(kind, str) 

243 assert isinstance(name, str) 

244 assert isinstance(text, str) or text is None 

245 assert isinstance(status, str) or status is None 

246 

247 self.framework = framework 

248 self.kind = kind 

249 self.name = name 

250 self.status = status 

251 

252 def to_json(self): 

253 rv = super().to_json() 

254 rv["framework"] = self.framework 

255 rv["kind"] = self.kind 

256 if self.text is not None: 

257 rv["text"] = self.text 

258 rv["status"] = self.status 

259 return rv 

260 

261 def perform_source_checks(self, source_info): 

262 assert isinstance(source_info, dict) 

263 

264 @classmethod 

265 def from_json(cls, level, data, schema_version): 

266 assert isinstance(level, str) 

267 assert isinstance(data, dict) 

268 assert schema_version in (3, 4) 

269 

270 item = Requirement(tag = Tracing_Tag.from_json(data["tag"]), 

271 location = Location.from_json(data["location"]), 

272 framework = data["framework"], 

273 kind = data["kind"], 

274 name = data["name"], 

275 text = data.get("text", None), 

276 status = data.get("status", None)) 

277 item.additional_data_from_json(level, data, schema_version) 

278 

279 return item 

280 

281 

282class Implementation(Item): 

283 def __init__( 

284 self, 

285 tag: Tracing_Tag, 

286 location: Location, 

287 language: str, 

288 kind: str, 

289 name: str, 

290 text: Optional[str] = None, 

291 ): 

292 super().__init__(tag, location, text=text) 

293 assert isinstance(language, str) 

294 assert isinstance(kind, str) 

295 assert isinstance(name, str) 

296 

297 self.language = language 

298 self.kind = kind 

299 self.name = name 

300 

301 def to_json(self): 

302 rv = super().to_json() 

303 rv["language"] = self.language 

304 rv["kind"] = self.kind 

305 if self.text is not None: 305 ↛ 306line 305 didn't jump to line 306 because the condition on line 305 was never true

306 rv["text"] = self.text 

307 return rv 

308 

309 @classmethod 

310 def from_json(cls, level, data, schema_version): 

311 assert isinstance(level, str) 

312 assert isinstance(data, dict) 

313 assert schema_version == 3 

314 

315 item = Implementation(tag = Tracing_Tag.from_json(data["tag"]), 

316 location = Location.from_json(data["location"]), 

317 language = data["language"], 

318 kind = data["kind"], 

319 name = data["name"], 

320 text = data.get("text", None)) 

321 item.additional_data_from_json(level, data, schema_version) 

322 

323 return item 

324 

325 

326class Activity(Item): 

327 def __init__( 

328 self, 

329 tag: Tracing_Tag, 

330 location: Location, 

331 framework: str, 

332 kind: str, 

333 text: Optional[str] = None, 

334 status: Optional[str] = None, 

335 ): 

336 super().__init__(tag, location, text=text) 

337 assert isinstance(framework, str) 

338 assert isinstance(kind, str) 

339 assert isinstance(text, str) or text is None 

340 assert isinstance(status, str) or status is None 

341 

342 self.framework = framework 

343 self.kind = kind 

344 self.status = status 

345 

346 def to_json(self): 

347 rv = super().to_json() 

348 rv["framework"] = self.framework 

349 rv["kind"] = self.kind 

350 if self.text is not None: 350 ↛ 351line 350 didn't jump to line 351 because the condition on line 350 was never true

351 rv["text"] = self.text 

352 rv["status"] = self.status 

353 return rv 

354 

355 @classmethod 

356 def from_json(cls, level, data, schema_version): 

357 assert isinstance(level, str) 

358 assert isinstance(data, dict) 

359 assert schema_version == 3 

360 

361 item = Activity(tag = Tracing_Tag.from_json(data["tag"]), 

362 location = Location.from_json(data["location"]), 

363 framework = data["framework"], 

364 kind = data["kind"], 

365 text = data.get("text", None), 

366 status = data.get("status", None)) 

367 item.additional_data_from_json(level, data, schema_version) 

368 

369 return item