Coverage for src/debputy/manifest_parser/util.py: 89%
235 statements
« prev ^ index » next coverage.py v7.8.2, created at 2026-07-22 10:58 +0000
1import dataclasses
2import typing
3from typing import (
4 Optional,
5 get_origin,
6 get_args,
7 Any,
8 TypeVar,
9 TYPE_CHECKING,
10 Literal,
11)
12from collections.abc import Iterator, Mapping, Iterable, Container
14from debputy.yaml.compat import CommentedBase
16from debputy.manifest_parser.exceptions import ManifestParseException
18if TYPE_CHECKING:
19 from debputy.manifest_parser.parser_data import ParserContextData
20 from debputy.manifest_parser.parse_hints import DebputyParseHint
21 from debputy.plugin.api.spec import DebputyIntegrationMode
24MP = TypeVar("MP", bound="DebputyParseHint")
25StrOrInt = str | int
26AttributePathAliasMapping = Mapping[
27 StrOrInt, tuple[StrOrInt, Optional["AttributePathAliasMapping"]]
28]
29LineReportKind = Literal["key", "value", "container"]
32class AttributePath:
33 __slots__ = ("parent", "container", "name", "alias_mapping", "path_hint")
35 # Summary: if parent is defined, name/key is too.
37 @typing.overload
38 def __init__( 38 ↛ exitline 38 didn't return from function '__init__' because
39 self,
40 parent: Optional["AttributePath"],
41 key: str | int,
42 *,
43 container: Any | None = None,
44 alias_mapping: AttributePathAliasMapping | None = None,
45 ) -> None: ...
47 @typing.overload
48 def __init__( 48 ↛ exitline 48 didn't return from function '__init__' because
49 self,
50 parent: None,
51 key: None,
52 *,
53 container: Any | None = None,
54 alias_mapping: AttributePathAliasMapping | None = None,
55 ) -> None: ...
57 def __init__(
58 self,
59 parent,
60 key,
61 *,
62 container=None,
63 alias_mapping=None,
64 ) -> None:
65 self.parent = parent
66 self.container = container
67 self.name = key
68 self.path_hint: str | None = None
69 self.alias_mapping = alias_mapping
71 @classmethod
72 def root_path(cls, container: Any | None) -> "AttributePath":
73 return AttributePath(None, None, container=container)
75 @classmethod
76 def builtin_path(cls) -> "AttributePath":
77 return AttributePath(None, "$builtin$")
79 @classmethod
80 def test_path(cls) -> "AttributePath":
81 return AttributePath(None, "$test$")
83 @classmethod
84 def highlighted_range(cls) -> "AttributePath":
85 return AttributePath(None, "highlighted range")
87 def copy_with_path_hint(self, path_hint: str) -> "AttributePath":
88 p = self.__class__(self.parent, self.name, alias_mapping=self.alias_mapping)
89 p.path_hint = path_hint
90 return p
92 def path_segments(self) -> Iterable[str | int]:
93 for name, _path_hint in self._iter_path():
94 yield name
96 def _resolve_path(self, report_kind: LineReportKind) -> str:
97 parent = self.parent
98 key = self.name
99 if report_kind == "container":
100 named_parent = parent is not None and parent.parent is not None
101 key = parent.name if named_parent else None
102 parent = parent.parent if named_parent else None
103 container = parent.container if parent is not None else None
105 if isinstance(container, CommentedBase):
106 lc = container.lc
107 try:
108 if isinstance(key, str):
109 if report_kind == "key":
110 lc_data = lc.key(key)
111 else:
112 lc_data = lc.value(key)
113 else:
114 lc_data = lc.item(key)
115 except (AttributeError, RuntimeError, LookupError, TypeError):
116 lc_data = None
117 else:
118 lc_data = None
120 parts: list[str] = []
121 path_hint: str | None = None
123 for k, s_path_hint in self._iter_path():
124 if s_path_hint is not None:
125 path_hint = s_path_hint
126 if isinstance(k, int):
127 parts.append(f"[{k}]")
128 else:
129 if parts:
130 parts.append(".")
131 parts.append(k)
133 if lc_data is not None:
134 line_pos, col = lc_data
135 # Translate 0-based (index) to 1-based (line number)
136 line_pos += 1
137 parts.append(f" [Line {line_pos} column {col}]")
139 elif path_hint: 139 ↛ 140line 139 didn't jump to line 140 because the condition on line 139 was never true
140 parts.append(f" <Search for: {path_hint}>")
141 if not parts: 141 ↛ 142line 141 didn't jump to line 142 because the condition on line 141 was never true
142 return "document root"
143 return "".join(parts)
145 @property
146 def path_container_lc(self) -> str:
147 return self._resolve_path("container")
149 @property
150 def path_key_lc(self) -> str:
151 return self._resolve_path("key")
153 @property
154 def path(self) -> str:
155 return self._resolve_path("value")
157 def __str__(self) -> str:
158 return self.path
160 def __getitem__(self, item: str | int) -> "AttributePath":
161 alias_mapping = None
162 if self.alias_mapping:
163 match = self.alias_mapping.get(item)
164 if match:
165 item, alias_mapping = match
166 if item == "":
167 # Support `sources[0]` mapping to `source` by `sources -> source` and `0 -> ""`.
168 return AttributePath(
169 self.parent,
170 self.name,
171 alias_mapping=alias_mapping,
172 container=self.container,
173 )
174 container = self.container
175 if container is not None:
176 try:
177 child_container = self.container[item]
178 except (AttributeError, RuntimeError, LookupError, TypeError):
179 child_container = None
180 else:
181 child_container = None
182 return AttributePath(
183 self,
184 item,
185 alias_mapping=alias_mapping,
186 container=child_container,
187 )
189 def _iter_path(self) -> Iterator[tuple[str | int, str | None]]:
190 "Parents, from the (excluded) root to (included) self."
191 if self.name is not None:
192 parent = self.parent
193 if parent is not None:
194 yield from parent._iter_path()
195 yield self.name, self.path_hint
198def check_integration_mode(
199 path: AttributePath,
200 parser_context: Optional["ParserContextData"] = None,
201 expected_debputy_integration_mode: (
202 Container["DebputyIntegrationMode"] | None
203 ) = None,
204) -> None:
205 if expected_debputy_integration_mode is None:
206 return
207 if parser_context is None:
208 raise AssertionError(
209 f"Cannot use integration mode restriction when parsing {path.path} since it is not parsed in the manifest context"
210 )
211 if parser_context.debputy_integration_mode not in expected_debputy_integration_mode: 211 ↛ 212line 211 didn't jump to line 212 because the condition on line 211 was never true
212 raise ManifestParseException(
213 f"The attribute {path.path} cannot be used as it is not allowed for"
214 f" the current debputy integration mode ({parser_context.debputy_integration_mode})."
215 f" Please remove the manifest definition or change the integration mode"
216 )
219@dataclasses.dataclass(slots=True, frozen=True)
220class _SymbolicModeSegment:
221 base_mode: int
222 base_mask: int
223 cap_x_mode: int
224 cap_x_mask: int
226 def apply(self, current_mode: int, is_dir: bool) -> int:
227 if current_mode & 0o111 or is_dir:
228 chosen_mode = self.cap_x_mode
229 mode_mask = self.cap_x_mask
230 else:
231 chosen_mode = self.base_mode
232 mode_mask = self.base_mask
233 # set ("="): mode mask clears relevant segment and current_mode are the desired bits
234 # add ("+"): mode mask keeps everything and current_mode are the desired bits
235 # remove ("-"): mode mask clears relevant bits and current_mode are 0
236 return (current_mode & mode_mask) | chosen_mode
239def _symbolic_mode_bit_inverse(v: int) -> int:
240 # The & part is necessary because otherwise python narrows the inversion to the minimum number of bits
241 # required, which is not what we want.
242 return ~v & 0o7777
245def parse_symbolic_mode(
246 symbolic_mode: str,
247 attribute_path: AttributePath | None,
248) -> Iterator[_SymbolicModeSegment]:
249 sticky_bit = 0o01000
250 setuid_bit = 0o04000
251 setgid_bit = 0o02000
252 mode_group_flag = 0o7
253 subject_mask_and_shift = {
254 "u": (mode_group_flag << 6, 6),
255 "g": (mode_group_flag << 3, 3),
256 "o": (mode_group_flag << 0, 0),
257 }
258 bits = {
259 "r": (0o4, 0o4),
260 "w": (0o2, 0o2),
261 "x": (0o1, 0o1),
262 "X": (0o0, 0o1),
263 "s": (0o0, 0o0), # Special-cased below (it depends on the subject)
264 "t": (0o0, 0o0), # Special-cased below
265 }
266 modifiers = {
267 "+",
268 "-",
269 "=",
270 }
271 in_path = f" in {attribute_path.path}" if attribute_path is not None else ""
272 for orig_part in symbolic_mode.split(","):
273 base_mode = 0
274 cap_x_mode = 0
275 part = orig_part
276 subjects = set()
277 while part and part[0] in ("u", "g", "o", "a"):
278 subject = part[0]
279 if subject == "a":
280 subjects = {"u", "g", "o"}
281 else:
282 subjects.add(subject)
283 part = part[1:]
284 if not subjects:
285 subjects = {"u", "g", "o"}
287 if part and part[0] in modifiers: 287 ↛ 289line 287 didn't jump to line 289 because the condition on line 287 was always true
288 modifier = part[0]
289 elif not part:
290 raise ValueError(
291 f'Invalid symbolic mode{in_path}: expected [+-=] to be present (from "{orig_part}")'
292 )
293 else:
294 raise ValueError(
295 f'Invalid symbolic mode{in_path}: Expected "{part[0]}" to be one of [+-=]'
296 f' (from "{orig_part}")'
297 )
298 part = part[1:]
299 s_bit_seen = False
300 t_bit_seen = False
301 while part and part[0] in bits:
302 if part == "s":
303 s_bit_seen = True
304 elif part == "t": 304 ↛ 305line 304 didn't jump to line 305 because the condition on line 304 was never true
305 t_bit_seen = True
306 elif part in ("u", "g", "o"): 306 ↛ 307line 306 didn't jump to line 307 because the condition on line 306 was never true
307 raise NotImplementedError(
308 f"Cannot parse symbolic mode{in_path}: Sorry, we do not support referencing an"
309 " existing subject's permissions (a=u) in symbolic modes."
310 )
311 else:
312 matched_bits = bits.get(part[0])
313 if matched_bits is None: 313 ↛ 314line 313 didn't jump to line 314 because the condition on line 313 was never true
314 valid_bits = "".join(bits)
315 raise ValueError(
316 f'Invalid symbolic mode{in_path}: Expected "{part[0]}" to be one of the letters'
317 f' in "{valid_bits}" (from "{orig_part}")'
318 )
319 base_mode_bits, cap_x_mode_bits = bits[part[0]]
320 base_mode |= base_mode_bits
321 cap_x_mode |= cap_x_mode_bits
322 part = part[1:]
324 if part: 324 ↛ 325line 324 didn't jump to line 325 because the condition on line 324 was never true
325 raise ValueError(
326 f'Invalid symbolic mode{in_path}: Could not parse "{part[0]}" from "{orig_part}"'
327 )
329 final_base_mode = 0
330 final_cap_x_mode = 0
331 segment_mask = 0
332 for subject in subjects:
333 mask, shift = subject_mask_and_shift[subject]
334 segment_mask |= mask
335 final_base_mode |= base_mode << shift
336 final_cap_x_mode |= cap_x_mode << shift
337 if modifier == "=":
338 segment_mask |= setuid_bit if "u" in subjects else 0
339 segment_mask |= setgid_bit if "g" in subjects else 0
340 segment_mask |= sticky_bit if "o" in subjects else 0
341 if s_bit_seen:
342 if "u" in subjects: 342 ↛ 345line 342 didn't jump to line 345 because the condition on line 342 was always true
343 final_base_mode |= setuid_bit
344 final_cap_x_mode |= setuid_bit
345 if "g" in subjects:
346 final_base_mode |= setgid_bit
347 final_cap_x_mode |= setgid_bit
348 if t_bit_seen: 348 ↛ 349line 348 didn't jump to line 349 because the condition on line 348 was never true
349 final_base_mode |= sticky_bit
350 final_cap_x_mode |= sticky_bit
351 if modifier == "+":
352 final_base_mask = ~0
353 final_cap_x_mask = ~0
354 elif modifier == "-":
355 final_base_mask = _symbolic_mode_bit_inverse(final_base_mode)
356 final_cap_x_mask = _symbolic_mode_bit_inverse(final_cap_x_mode)
357 final_base_mode = 0
358 final_cap_x_mode = 0
359 elif modifier == "=":
360 # FIXME: Handle "unmentioned directory's setgid/setuid bits"
361 inverted_mask = _symbolic_mode_bit_inverse(segment_mask)
362 final_base_mask = inverted_mask
363 final_cap_x_mask = inverted_mask
364 else:
365 raise AssertionError(
366 f"Unknown modifier in symbolic mode: {modifier} - should not have happened"
367 )
368 yield _SymbolicModeSegment(
369 base_mode=final_base_mode,
370 base_mask=final_base_mask,
371 cap_x_mode=final_cap_x_mode,
372 cap_x_mask=final_cap_x_mask,
373 )
376def unpack_type(
377 orig_type: Any,
378 parsing_typed_dict_attribute: bool,
379) -> tuple[Any, Any | None, tuple[Any, ...]]:
380 raw_type = orig_type
381 if type(orig_type) == typing.TypeAliasType:
382 orig_type = orig_type.__value__
383 origin = get_origin(orig_type)
384 args = get_args(orig_type)
386 if not parsing_typed_dict_attribute and repr(origin) in ( 386 ↛ 390line 386 didn't jump to line 390 because the condition on line 386 was never true
387 "typing.NotRequired",
388 "typing.Required",
389 ):
390 raise ValueError(
391 f"The Required/NotRequired attributes cannot be used outside typed dicts,"
392 f" the type that triggered the error: {orig_type}"
393 )
395 while repr(origin) in ("typing.NotRequired", "typing.Required"):
396 if len(args) != 1: 396 ↛ 397line 396 didn't jump to line 397 because the condition on line 396 was never true
397 raise ValueError(
398 f"The type {raw_type} should have exactly one type parameter"
399 )
400 raw_type = args[0]
401 origin = get_origin(raw_type)
402 args = get_args(raw_type)
404 assert not isinstance(raw_type, tuple)
406 return raw_type, origin, args
409def find_annotation(
410 annotations: tuple[Any, ...],
411 anno_class: type[MP],
412) -> MP | None:
413 m = None
414 for anno in annotations:
415 if isinstance(anno, anno_class):
416 if m is not None: 416 ↛ 417line 416 didn't jump to line 417 because the condition on line 416 was never true
417 raise ValueError(
418 f"The annotation {anno_class.__name__} was used more than once"
419 )
420 m = anno
421 return m