Coverage for lobster/common/items.py: 94%
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/>.
20from enum import Enum, auto
21from abc import ABCMeta
22from hashlib import sha1
23from typing import Optional
25from lobster.common.location import Location
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"
40 self.namespace = namespace
41 self.tag = tag
42 self.version = version
43 self.hash_val = None
45 def __str__(self):
46 rv = f"{self.namespace} {self.tag}"
47 if self.version: 47 ↛ 48line 47 didn't jump to line 48 because the condition on line 47 was never true
48 rv += f"@{self.version}"
49 return rv
51 def key(self) -> str:
52 return self.namespace + " " + self.tag
54 def to_json(self) -> str:
55 return str(self)
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)
63 @classmethod
64 def from_text(cls, namespace, text):
65 assert isinstance(namespace, str)
66 assert isinstance(text, str)
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)
75 def hash(self):
76 if not self.hash_val: 76 ↛ 80line 76 didn't jump to line 80 because the condition on line 76 was always true
77 hfunc = sha1()
78 hfunc.update(self.key().encode("UTF-8"))
79 self.hash_val = hfunc.hexdigest()
80 return self.hash_val
83class Tracing_Status(Enum):
84 OK = auto()
85 PARTIAL = auto()
86 MISSING = auto()
87 JUSTIFIED = auto()
88 ERROR = auto()
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
98 self.level = None
99 self.tag = tag
100 self.location = location
101 self.name = tag.tag
102 self.text = text
104 self.ref_up = []
105 self.ref_down = []
107 self.unresolved_references_cache = set()
108 self.unresolved_references = []
110 self.messages = []
111 self.just_up = []
112 self.just_down = []
113 self.just_global = []
115 self.tracing_status = None
116 self.has_error = False
118 def set_level(self, level):
119 assert isinstance(level, str)
120 self.level = level
122 def error(self, message: str):
123 assert isinstance(message, str)
124 self.messages.append(message)
125 self.has_error = True
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
132 self.unresolved_references.append(target)
133 self.unresolved_references_cache.add(target.key())
135 def perform_source_checks(self, source_info):
136 assert isinstance(source_info, dict)
138 def determine_status(self, config, stab):
139 assert self.level in config
140 assert self.tag.key() in stab
142 level = config[self.level]
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
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")
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}")
170 # Set status
171 if self.has_error: 171 ↛ 172line 171 didn't jump to line 172 because the condition on line 171 was never true
172 self.tracing_status = Tracing_Status.MISSING
173 elif ok_up and ok_down: 173 ↛ 174line 173 didn't jump to line 174 because the condition on line 173 was never true
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
185 # Overwrite status if there are initial errors
186 if self.tracing_status == Tracing_Status.OK and has_init_errors: 186 ↛ 187line 186 didn't jump to line 187 because the condition on line 186 was never true
187 self.tracing_status = Tracing_Status.PARTIAL
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
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"]]
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
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
247 self.framework = framework
248 self.kind = kind
249 self.name = name
250 self.status = status
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
261 def perform_source_checks(self, source_info):
262 assert isinstance(source_info, dict)
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)
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)
279 return item
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)
297 self.language = language
298 self.kind = kind
299 self.name = name
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
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
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)
323 return item
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
342 self.framework = framework
343 self.kind = kind
344 self.status = status
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:
351 rv["text"] = self.text
352 rv["status"] = self.status
353 return rv
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
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)
369 return item