Coverage for lobster/common/yaml_policy_reader.py: 88%

1#!/usr/bin/env python3 

2# 

3# LOBSTER - Lightweight Open BMW Software Traceability Evidence Report 

4# Copyright (C) 2026 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 

20# Reads a YAML tracing policy file into a RawPolicy, validating it against 

21# lobster/common/tracing_policy.yamale first (same precedent as 

22# lobster/tools/trlc/lobster_trlc_config.py). 

23 

24from pathlib import Path 

25from typing import Optional 

26 

27import yaml 

28from yamale import YamaleError 

29 

30from lobster.common.exceptions import LOBSTER_Exception 

31from lobster.common.location import File_Reference 

32from lobster.common.raw_policy import ( 

33 RawLevel, RawPolicy, RawRequiresCandidate, RawSource, RawTraceTo, 

34) 

35from lobster.common.yaml_validation import validate_yaml_file 

36 

37SCHEMA_FILE = Path(__file__).parent / "tracing_policy.yamale" 

38 

39 

40# pylint: disable-next=too-many-ancestors 

41class UniqueKeyLoader(yaml.SafeLoader): 

42 """Safe loader that rejects duplicate mapping keys instead of keeping the last.""" 

43 

44 def construct_mapping(self, node, deep=False): 

45 seen = set() 

46 for key_node, _ in node.value: 

47 key = self.construct_object(key_node, deep=True) 

48 if key in seen: 48 ↛ 49line 48 didn't jump to line 49 because the condition on line 48 was never true

49 raise yaml.constructor.ConstructorError( 

50 None, None, f"duplicate key {key!r}", key_node.start_mark) 

51 seen.add(key) 

52 return super().construct_mapping(node, deep) 

53 

54 

55def _line_of_path(filename: str, path: str) -> Optional[int]: 

56 """Return the 1-based line of a yamale error path like 'levels.0.kind'. 

57 

58 Falls back to the deepest node that exists, e.g. for missing keys. 

59 """ 

60 with open(filename, "r", encoding="UTF-8") as f: 

61 node = yaml.compose(f, Loader=yaml.SafeLoader) 

62 if node is None: 62 ↛ 63line 62 didn't jump to line 63 because the condition on line 62 was never true

63 return None 

64 for part in path.split("."): 

65 child = None 

66 if isinstance(node, yaml.MappingNode): 

67 for key_node, value_node in node.value: 67 ↛ 75line 67 didn't jump to line 75 because the loop on line 67 didn't complete

68 if key_node.value == part: 

69 child = value_node 

70 break 

71 elif isinstance(node, yaml.SequenceNode) and part.isdigit(): 71 ↛ 75line 71 didn't jump to line 75 because the condition on line 71 was always true

72 index = int(part) 

73 if index < len(node.value): 73 ↛ 75line 73 didn't jump to line 75 because the condition on line 73 was always true

74 child = node.value[index] 

75 if child is None: 75 ↛ 76line 75 didn't jump to line 76 because the condition on line 75 was never true

76 break 

77 node = child 

78 return node.start_mark.line + 1 

79 

80 

81def _describe_error(filename: str, err: Exception) -> str: 

82 if isinstance(err, yaml.YAMLError): 82 ↛ 83line 82 didn't jump to line 83 because the condition on line 82 was never true

83 mark = getattr(err, "problem_mark", None) 

84 where = f"{filename}:{mark.line + 1}" if mark else filename 

85 return f"{where}: invalid tracing policy: {err}" 

86 

87 messages = [] 

88 for result in err.results: 

89 for error in result.errors: 

90 path, _, text = str(error).partition(": ") 

91 line = _line_of_path(filename, path) if text else None 

92 where = f"{filename}:{line}" if line else filename 

93 messages.append(f"{where}: invalid tracing policy: {error}") 

94 return "\n".join(messages) or f"{filename}: invalid tracing policy: {err}" 

95 

96 

97def read_yaml_policy(filename: str) -> RawPolicy: 

98 if not Path(filename).is_file(): 

99 raise FileNotFoundError(f"Config file not found: {filename}") 

100 

101 try: 

102 with open(filename, "r", encoding="UTF-8") as f: 

103 data = yaml.load(f, Loader=UniqueKeyLoader) 

104 validate_yaml_file(schema_file=SCHEMA_FILE, filename=filename) 

105 except (YamaleError, yaml.YAMLError) as err: 

106 raise LOBSTER_Exception(_describe_error(filename, err)) from err 

107 

108 loc = File_Reference(filename) 

109 levels = [] 

110 for level_data in data.get("levels", []): 

111 level = RawLevel(name=level_data["name"], 

112 name_loc=loc, 

113 kind=level_data["kind"]) 

114 for source_file in level_data.get("source") or []: 

115 level.source.append(RawSource(file=source_file, loc=loc)) 

116 for target in level_data.get("trace-to") or []: 

117 level.trace_to.append(RawTraceTo(target=target, loc=loc)) 

118 for or_group in level_data.get("requires") or []: 

119 level.requires.append( 

120 [RawRequiresCandidate(name=name, loc=loc) 

121 for name in or_group]) 

122 levels.append(level) 

123 

124 return RawPolicy(levels=levels)