Coverage for src/debputy/plugin/api/test_api/test_impl.py: 80%
300 statements
« prev ^ index » next coverage.py v7.8.2, created at 2026-07-22 10:58 +0000
1import contextlib
2import dataclasses
3import inspect
4import os.path
5from collections.abc import Mapping, Sequence, Iterator, KeysView, Callable
6from importlib.resources.abc import Traversable
7from io import BytesIO
8from pathlib import Path
9from typing import (
10 cast,
11)
13from debian.deb822 import Deb822
14from debian.debian_support import DpkgArchTable
15from debian.substvars import Substvars
17from debputy.architecture_support import DpkgArchitectureBuildProcessValuesTable
18from debputy.filesystem_scan import OSFSROOverlay, InMemoryVirtualRootDir
19from debputy.packages import BinaryPackage, SourcePackage
20from debputy.plugin.api import (
21 PluginInitializationEntryPoint,
22 VirtualPath,
23 PackageProcessingContext,
24 DpkgTriggerType,
25 Maintscript,
26)
27from debputy.plugin.api.example_processing import process_discard_rule_example
28from debputy.plugin.api.feature_set import PluginProvidedFeatureSet
29from debputy.plugin.api.impl import (
30 plugin_metadata_for_debputys_own_plugin,
31 DebputyPluginInitializerProvider,
32 parse_json_plugin_desc,
33 MaintscriptAccessorProviderBase,
34 BinaryCtrlAccessorProviderBase,
35 PLUGIN_TEST_SUFFIX,
36 find_json_plugin,
37 ServiceDefinitionImpl,
38)
39from debputy.plugin.api.impl_types import (
40 PackagerProvidedFileClassSpec,
41 DebputyPluginMetadata,
42 PluginProvidedTrigger,
43 ServiceManagerDetails,
44)
45from debputy.plugin.api.spec import (
46 MaintscriptAccessor,
47 FlushableSubstvars,
48 ServiceRegistry,
49 DSD,
50 ServiceUpgradeRule,
51)
52from debputy.plugin.api.test_api.test_spec import (
53 InitializedPluginUnderTest,
54 RegisteredPackagerProvidedFile,
55 RegisteredTrigger,
56 RegisteredMaintscript,
57 DEBPUTY_TEST_AGAINST_INSTALLED_PLUGINS,
58 ADRExampleIssue,
59 DetectedService,
60 RegisteredMetadata,
61)
62from debputy.plugins.debputy.debputy_plugin import initialize_debputy_features
63from debputy.substitution import SubstitutionImpl, VariableContext, Substitution
64from debputy.util import package_cross_check_precheck
65from debputy.version import DEBPUTY_PLUGIN_ROOT_DIR
67RegisteredPackagerProvidedFile.register(PackagerProvidedFileClassSpec)
70type ManifestConfigurationImplementation[T] = Callable[
71 [SourcePackage | BinaryPackage, type[T]], T
72]
75@dataclasses.dataclass(frozen=True, slots=True)
76class PackageProcessingContextTestProvider(PackageProcessingContext):
77 source_package: SourcePackage
78 binary_package: BinaryPackage
79 binary_package_version: str
80 related_udeb_package: BinaryPackage | None
81 related_udeb_package_version: str | None
82 accessible_package_roots: Callable[[], Sequence[tuple[BinaryPackage, VirtualPath]]]
83 manifest_configuration: ManifestConfigurationImplementation
85 # TODO: implement (when needed)
86 # dpkg_arch_query_table
87 # deb_options_and_profiles (pull from binary ?)
88 # source_condition_context
89 # condition_context
92def _initialize_plugin_under_test(
93 plugin_metadata: DebputyPluginMetadata,
94 load_debputy_plugin: bool = True,
95) -> "InitializedPluginUnderTest":
96 feature_set = PluginProvidedFeatureSet()
97 substitution = SubstitutionImpl(
98 unresolvable_substitutions=frozenset(["SOURCE_DATE_EPOCH", "PACKAGE"]),
99 variable_context=VariableContext(
100 OSFSROOverlay.create_root_dir("debian", "debian"),
101 ),
102 plugin_feature_set=feature_set,
103 )
105 if load_debputy_plugin:
106 debputy_plugin_metadata = plugin_metadata_for_debputys_own_plugin(
107 initialize_debputy_features
108 )
109 # Load debputy's own plugin first, so conflicts with debputy's plugin are detected early
110 debputy_provider = DebputyPluginInitializerProvider(
111 debputy_plugin_metadata,
112 feature_set,
113 substitution,
114 )
115 debputy_provider.load_plugin()
117 plugin_under_test_provider = DebputyPluginInitializerProvider(
118 plugin_metadata,
119 feature_set,
120 substitution,
121 )
122 plugin_under_test_provider.load_plugin()
124 return InitializedPluginUnderTestImpl(
125 plugin_metadata.plugin_name,
126 feature_set,
127 substitution,
128 )
131def _auto_load_plugin_from_filename(
132 py_test_filename: str,
133) -> "InitializedPluginUnderTest":
134 dirname, basename = os.path.split(py_test_filename)
135 plugin_name = PLUGIN_TEST_SUFFIX.sub("", basename).replace("_", "-")
137 test_location = os.environ.get("DEBPUTY_TEST_PLUGIN_LOCATION", "uninstalled")
138 if test_location == "uninstalled":
139 json_basename = f"{plugin_name}.json"
140 json_desc_file = os.path.join(dirname, json_basename)
141 if "/" not in json_desc_file: 141 ↛ 142line 141 didn't jump to line 142 because the condition on line 141 was never true
142 json_desc_file = f"./{json_desc_file}"
144 if os.path.isfile(json_desc_file): 144 ↛ 147line 144 didn't jump to line 147 because the condition on line 144 was always true
145 return _initialize_plugin_from_desc(json_desc_file)
147 json_desc_file_in = f"{json_desc_file}.in"
148 if os.path.isfile(json_desc_file_in):
149 return _initialize_plugin_from_desc(json_desc_file)
150 raise FileNotFoundError(
151 f"Cannot determine the plugin JSON metadata descriptor: Expected it to be"
152 f" {json_desc_file} or {json_desc_file_in}"
153 )
155 if test_location == "installed": 155 ↛ 159line 155 didn't jump to line 159 because the condition on line 155 was always true
156 plugin_metadata = find_json_plugin([str(DEBPUTY_PLUGIN_ROOT_DIR)], plugin_name)
157 return _initialize_plugin_under_test(plugin_metadata, load_debputy_plugin=True)
159 raise ValueError(
160 'Invalid or unsupported "DEBPUTY_TEST_PLUGIN_LOCATION" environment variable. It must be either'
161 ' unset OR one of "installed", "uninstalled".'
162 )
165def initialize_plugin_under_test(
166 *,
167 plugin_desc_file: str | None = None,
168) -> "InitializedPluginUnderTest":
169 """Load and initialize a plugin for testing it
171 This method will load the plugin via plugin description, which is the method that `debputy` does at
172 run-time (in contrast to `initialize_plugin_under_test_preloaded`, which bypasses this concrete part
173 of the flow).
175 :param plugin_desc_file: The plugin description file (`.json`) that describes how to load the plugin.
176 If omitted, `debputy` will attempt to attempt the plugin description file based on the test itself.
177 This works for "single-file" plugins, where the description file and the test are right next to
178 each other.
179 :return: The loaded plugin for testing
180 """
181 if plugin_desc_file is None:
182 caller_file = inspect.stack()[1].filename
183 return _auto_load_plugin_from_filename(caller_file)
184 if DEBPUTY_TEST_AGAINST_INSTALLED_PLUGINS: 184 ↛ 185line 184 didn't jump to line 185 because the condition on line 184 was never true
185 raise RuntimeError(
186 "Running the test against an installed plugin does not work when"
187 " plugin_desc_file is provided. Please skip this test. You can "
188 " import DEBPUTY_TEST_AGAINST_INSTALLED_PLUGINS and use that as"
189 " conditional for this purpose."
190 )
191 return _initialize_plugin_from_desc(plugin_desc_file)
194def _initialize_plugin_from_desc(
195 desc_file: str,
196) -> "InitializedPluginUnderTest":
197 if not desc_file.endswith((".json", ".json.in")): 197 ↛ 198line 197 didn't jump to line 198 because the condition on line 197 was never true
198 raise ValueError("The plugin file must end with .json or .json.in")
200 plugin_metadata = parse_json_plugin_desc(desc_file)
202 return _initialize_plugin_under_test(plugin_metadata, load_debputy_plugin=True)
205def initialize_plugin_under_test_from_inline_json(
206 plugin_name: str,
207 json_content: str,
208) -> "InitializedPluginUnderTest":
209 with BytesIO(json_content.encode("utf-8")) as fd:
210 plugin_metadata = parse_json_plugin_desc(plugin_name, fd=fd)
212 return _initialize_plugin_under_test(plugin_metadata, load_debputy_plugin=True)
215def initialize_plugin_under_test_preloaded(
216 api_compat_version: int,
217 plugin_initializer: PluginInitializationEntryPoint,
218 /,
219 plugin_name: str = "plugin-under-test",
220 load_debputy_plugin: bool = True,
221 plugin_doc_path_resolver: Callable[[], Traversable | Path | None] = lambda: None,
222) -> "InitializedPluginUnderTest":
223 """Internal API: Initialize a plugin for testing without loading it from a file
225 This method by-passes the standard loading mechanism, meaning you will not test that your plugin
226 description file is correct. Notably, any feature provided via the JSON description file will
227 **NOT** be visible for the test.
229 This API is mostly useful for testing parts of debputy itself.
231 :param api_compat_version: The API version the plugin was written for. Use the same version as the
232 version from the entry point (The `v1` part of `debputy.plugins.v1.initialize` translate into `1`).
233 :param plugin_initializer: The entry point of the plugin
234 :param plugin_name: Normally, debputy would derive this from the entry point. In the test, it will
235 use a test name and version. However, you can explicitly set if you want the real name/version.
236 :param load_debputy_plugin: Whether to load debputy's own plugin first. Doing so provides a more
237 realistic test and enables the test to detect conflicts with debputy's own plugins (de facto making
238 the plugin unloadable in practice if such a conflict is present). This option is mostly provided
239 to enable debputy to use this method for self testing.
240 :param plugin_doc_path_resolver: How to resolve the documentation (if relevant for the test). The
241 default is to not load the documentation.
242 :return: The loaded plugin for testing
243 """
245 if DEBPUTY_TEST_AGAINST_INSTALLED_PLUGINS: 245 ↛ 246line 245 didn't jump to line 246 because the condition on line 245 was never true
246 raise RuntimeError(
247 "Running the test against an installed plugin does not work when"
248 " the plugin is preload. Please skip this test. You can "
249 " import DEBPUTY_TEST_AGAINST_INSTALLED_PLUGINS and use that as"
250 " conditional for this purpose."
251 )
253 plugin_metadata = DebputyPluginMetadata(
254 plugin_name=plugin_name,
255 api_compat_version=api_compat_version,
256 plugin_initializer=plugin_initializer,
257 plugin_loader=None,
258 plugin_path="<loaded-via-test>",
259 plugin_doc_path_resolver=plugin_doc_path_resolver,
260 )
262 return _initialize_plugin_under_test(
263 plugin_metadata,
264 load_debputy_plugin=load_debputy_plugin,
265 )
268class _MockArchTable:
269 @staticmethod
270 def matches_architecture(_a: str, _b: str) -> bool:
271 return True
274FAKE_DPKG_QUERY_TABLE = cast(DpkgArchTable, _MockArchTable())
275del _MockArchTable
278def package_metadata_context(
279 *,
280 host_arch: str = "amd64",
281 package_fields: dict[str, str] | None = None,
282 related_udeb_package_fields: dict[str, str] | None = None,
283 binary_package_version: str = "1.0-1",
284 related_udeb_package_version: str | None = None,
285 should_be_acted_on: bool = True,
286 related_udeb_fs_root: VirtualPath | None = None,
287 accessible_package_roots: Sequence[tuple[Mapping[str, str], VirtualPath]] = tuple(),
288 source_package_fields: dict[str, str] | None = None,
289 manifest_configuration: ManifestConfigurationImplementation = lambda x, y: None,
290) -> PackageProcessingContext:
291 process_table = DpkgArchitectureBuildProcessValuesTable(fake_host=host_arch)
292 f = {
293 "Package": "foo",
294 "Architecture": "any",
295 }
296 if package_fields is not None:
297 f.update(package_fields)
299 bin_package = BinaryPackage(
300 Deb822(f),
301 process_table,
302 FAKE_DPKG_QUERY_TABLE,
303 is_main_package=True,
304 should_be_acted_on=should_be_acted_on,
305 )
306 udeb_package = None
307 s = {
308 "Source": bin_package.name,
309 }
310 if source_package_fields is not None: 310 ↛ 311line 310 didn't jump to line 311 because the condition on line 310 was never true
311 s.update(source_package_fields)
312 source_package = SourcePackage(Deb822(s))
313 if related_udeb_package_fields is not None: 313 ↛ 314line 313 didn't jump to line 314 because the condition on line 313 was never true
314 uf = dict(related_udeb_package_fields)
315 uf.setdefault("Package", f'{f["Package"]}-udeb')
316 uf.setdefault("Architecture", f["Architecture"])
317 uf.setdefault("Package-Type", "udeb")
318 udeb_package = BinaryPackage(
319 Deb822(uf),
320 process_table,
321 FAKE_DPKG_QUERY_TABLE,
322 is_main_package=False,
323 should_be_acted_on=True,
324 )
325 if related_udeb_package_version is None:
326 related_udeb_package_version = binary_package_version
327 if accessible_package_roots:
328 apr = []
329 for fields, apr_fs_root in accessible_package_roots:
330 apr_fields = Deb822(dict(fields))
331 if "Package" not in apr_fields: 331 ↛ 332line 331 didn't jump to line 332 because the condition on line 331 was never true
332 raise ValueError(
333 "Missing mandatory Package field in member of accessible_package_roots"
334 )
335 if "Architecture" not in apr_fields: 335 ↛ 336line 335 didn't jump to line 336 because the condition on line 335 was never true
336 raise ValueError(
337 "Missing mandatory Architecture field in member of accessible_package_roots"
338 )
339 apr_package = BinaryPackage(
340 apr_fields,
341 process_table,
342 FAKE_DPKG_QUERY_TABLE,
343 is_main_package=False,
344 should_be_acted_on=True,
345 )
346 r = package_cross_check_precheck(bin_package, apr_package)
347 if not r[0]: 347 ↛ 348line 347 didn't jump to line 348 because the condition on line 347 was never true
348 raise ValueError(
349 f"{apr_package.name} would not be accessible for {bin_package.name}"
350 )
351 apr.append((apr_package, apr_fs_root))
353 if related_udeb_fs_root is not None: 353 ↛ 354line 353 didn't jump to line 354 because the condition on line 353 was never true
354 if udeb_package is None:
355 raise ValueError(
356 "related_udeb_package_fields must be given when related_udeb_fs_root is given"
357 )
358 r = package_cross_check_precheck(bin_package, udeb_package)
359 if not r[0]:
360 raise ValueError(
361 f"{udeb_package.name} would not be accessible for {bin_package.name}, so providing"
362 " related_udeb_fs_root is irrelevant"
363 )
364 apr.append((udeb_package, related_udeb_fs_root))
365 final_apr = tuple(apr)
366 else:
367 final_apr = tuple()
369 return PackageProcessingContextTestProvider(
370 source_package=source_package,
371 binary_package=bin_package,
372 related_udeb_package=udeb_package,
373 binary_package_version=binary_package_version,
374 related_udeb_package_version=related_udeb_package_version,
375 accessible_package_roots=lambda: final_apr,
376 manifest_configuration=manifest_configuration,
377 )
380def manifest_variable_resolution_context(
381 *,
382 debian_dir: VirtualPath | None = None,
383) -> VariableContext:
384 if debian_dir is None:
385 debian_dir = InMemoryVirtualRootDir()
387 return VariableContext(debian_dir)
390class MaintscriptAccessorTestProvider(MaintscriptAccessorProviderBase):
391 __slots__ = ("_plugin_metadata", "_plugin_source_id", "_maintscript_container")
393 def __init__(
394 self,
395 plugin_metadata: DebputyPluginMetadata,
396 plugin_source_id: str,
397 maintscript_container: dict[str, list[RegisteredMaintscript]],
398 ):
399 self._plugin_metadata = plugin_metadata
400 self._plugin_source_id = plugin_source_id
401 self._maintscript_container = maintscript_container
403 @classmethod
404 def _apply_condition_to_script(
405 cls, condition: str, run_snippet: str, /, indent: bool | None = None
406 ) -> str:
407 return run_snippet
409 def _append_script(
410 self,
411 caller_name: str,
412 maintscript: Maintscript,
413 full_script: str,
414 /,
415 perform_substitution: bool = True,
416 ) -> None:
417 if self._plugin_source_id not in self._maintscript_container:
418 self._maintscript_container[self._plugin_source_id] = []
419 self._maintscript_container[self._plugin_source_id].append(
420 RegisteredMaintscript(
421 maintscript,
422 caller_name,
423 full_script,
424 perform_substitution,
425 )
426 )
429class RegisteredMetadataImpl(RegisteredMetadata):
430 __slots__ = (
431 "_substvars",
432 "_triggers",
433 "_maintscripts",
434 )
436 def __init__(
437 self,
438 substvars: Substvars,
439 triggers: list[RegisteredTrigger],
440 maintscripts: list[RegisteredMaintscript],
441 ) -> None:
442 self._substvars = substvars
443 self._triggers = triggers
444 self._maintscripts = maintscripts
446 @property
447 def substvars(self) -> Substvars:
448 return self._substvars
450 @property
451 def triggers(self) -> list[RegisteredTrigger]:
452 return self._triggers
454 def maintscripts(
455 self,
456 *,
457 maintscript: Maintscript | None = None,
458 ) -> list[RegisteredMaintscript]:
459 if maintscript is None:
460 return self._maintscripts
461 return [m for m in self._maintscripts if m.maintscript == maintscript]
464class BinaryCtrlAccessorTestProvider(BinaryCtrlAccessorProviderBase):
465 __slots__ = ("_maintscript_container",)
467 def __init__(
468 self,
469 plugin_metadata: DebputyPluginMetadata,
470 plugin_source_id: str,
471 context: PackageProcessingContext,
472 ) -> None:
473 super().__init__(
474 plugin_metadata,
475 plugin_source_id,
476 context,
477 {},
478 FlushableSubstvars(),
479 (None, None),
480 )
481 self._maintscript_container: dict[str, list[RegisteredMaintscript]] = {}
483 def _create_maintscript_accessor(self) -> MaintscriptAccessor:
484 return MaintscriptAccessorTestProvider(
485 self._plugin_metadata,
486 self._plugin_source_id,
487 self._maintscript_container,
488 )
490 def registered_metadata(self) -> RegisteredMetadata:
491 return RegisteredMetadataImpl(
492 self._substvars,
493 [
494 RegisteredTrigger.from_plugin_provided_trigger(t)
495 for t in self._triggers.values()
496 if t.provider_source_id == self._plugin_source_id
497 ],
498 self._maintscript_container.get(self._plugin_source_id, []),
499 )
502class ServiceRegistryTestImpl(ServiceRegistry[DSD]):
503 __slots__ = ("_service_manager_details", "_service_definitions")
505 def __init__(
506 self,
507 service_manager_details: ServiceManagerDetails,
508 detected_services: list[DetectedService[DSD]],
509 ) -> None:
510 self._service_manager_details = service_manager_details
511 self._service_definitions = detected_services
513 def register_service(
514 self,
515 path: VirtualPath,
516 name: str | list[str],
517 *,
518 type_of_service: str = "service", # "timer", etc.
519 service_scope: str = "system",
520 enable_by_default: bool = True,
521 start_by_default: bool = True,
522 default_upgrade_rule: ServiceUpgradeRule = "restart",
523 service_context: DSD | None = None,
524 ) -> None:
525 names = name if isinstance(name, list) else [name]
526 if len(names) < 1: 526 ↛ 527line 526 didn't jump to line 527 because the condition on line 526 was never true
527 raise ValueError(
528 f"The service must have at least one name - {path.absolute} did not have any"
529 )
530 self._service_definitions.append(
531 DetectedService(
532 path,
533 names,
534 type_of_service,
535 service_scope,
536 enable_by_default,
537 start_by_default,
538 default_upgrade_rule,
539 service_context,
540 )
541 )
544@contextlib.contextmanager
545def _read_only_fs_root(fs_root: VirtualPath) -> Iterator[VirtualPath]:
546 if fs_root.is_read_write: 546 ↛ 552line 546 didn't jump to line 552 because the condition on line 546 was always true
547 assert isinstance(fs_root, InMemoryVirtualRootDir)
548 fs_root.is_read_write = False
549 yield fs_root
550 fs_root.is_read_write = True
551 else:
552 yield fs_root
555class InitializedPluginUnderTestImpl(InitializedPluginUnderTest):
556 def __init__(
557 self,
558 plugin_name: str,
559 feature_set: PluginProvidedFeatureSet,
560 substitution: SubstitutionImpl,
561 ) -> None:
562 self._feature_set = feature_set
563 self._plugin_name = plugin_name
564 self._packager_provided_files: None | (
565 dict[str, RegisteredPackagerProvidedFile]
566 ) = None
567 self._triggers: dict[tuple[DpkgTriggerType, str], PluginProvidedTrigger] = {}
568 self._maintscript_container: dict[str, list[RegisteredMaintscript]] = {}
569 self._substitution = substitution
570 assert plugin_name in self._feature_set.plugin_data
572 @property
573 def _plugin_metadata(self) -> DebputyPluginMetadata:
574 return self._feature_set.plugin_data[self._plugin_name]
576 def packager_provided_files_by_stem(
577 self,
578 ) -> Mapping[str, RegisteredPackagerProvidedFile]:
579 ppf = self._packager_provided_files
580 if ppf is None:
581 result: dict[str, RegisteredPackagerProvidedFile] = {}
582 for spec in self._feature_set.packager_provided_files.values():
583 if spec.debputy_plugin_metadata.plugin_name != self._plugin_name:
584 continue
585 # Registered as a virtual subclass, so this should always be True
586 assert isinstance(spec, RegisteredPackagerProvidedFile)
587 result[spec.stem] = spec
588 self._packager_provided_files = result
589 ppf = result
590 return ppf
592 def run_metadata_detector(
593 self,
594 metadata_detector_id: str,
595 fs_root: VirtualPath,
596 context: PackageProcessingContext | None = None,
597 ) -> RegisteredMetadata:
598 if not fs_root.is_root_dir(): 598 ↛ 599line 598 didn't jump to line 599 because the condition on line 598 was never true
599 raise ValueError("Provided path must be the file system root.")
600 detectors = self._feature_set.metadata_maintscript_detectors[self._plugin_name]
601 matching_detectors = [
602 d for d in detectors if d.detector_id == metadata_detector_id
603 ]
604 if len(matching_detectors) != 1: 604 ↛ 605line 604 didn't jump to line 605 because the condition on line 604 was never true
605 assert not matching_detectors
606 raise ValueError(
607 f"The plugin {self._plugin_name} did not provide a metadata detector with ID"
608 f' "{metadata_detector_id}"'
609 )
610 if context is None:
611 context = package_metadata_context()
612 detector = matching_detectors[0]
613 if not detector.applies_to(context.binary_package):
614 raise ValueError(
615 f'The detector "{metadata_detector_id}" from {self._plugin_name} does not apply to the'
616 " given package. Consider using `package_metadata_context()` to emulate a binary package"
617 " with the correct specification. As an example: "
618 '`package_metadata_context(package_fields={"Package-Type": "udeb"})` would emulate a udeb'
619 " package."
620 )
622 ctrl = BinaryCtrlAccessorTestProvider(
623 self._plugin_metadata,
624 metadata_detector_id,
625 context,
626 )
627 with _read_only_fs_root(fs_root) as ro_root:
628 detector.run_detector(
629 ro_root,
630 ctrl,
631 context,
632 )
633 return ctrl.registered_metadata()
635 def run_package_processor(
636 self,
637 package_processor_id: str,
638 fs_root: VirtualPath,
639 context: PackageProcessingContext | None = None,
640 ) -> None:
641 if not fs_root.is_root_dir(): 641 ↛ 642line 641 didn't jump to line 642 because the condition on line 641 was never true
642 raise ValueError("Provided path must be the file system root.")
643 pp_key = (self._plugin_name, package_processor_id)
644 package_processor = self._feature_set.all_package_processors.get(pp_key)
645 if package_processor is None: 645 ↛ 646line 645 didn't jump to line 646 because the condition on line 645 was never true
646 raise ValueError(
647 f"The plugin {self._plugin_name} did not provide a package processor with ID"
648 f' "{package_processor_id}"'
649 )
650 if context is None: 650 ↛ 652line 650 didn't jump to line 652 because the condition on line 650 was always true
651 context = package_metadata_context()
652 if not fs_root.is_read_write: 652 ↛ 653line 652 didn't jump to line 653 because the condition on line 652 was never true
653 raise ValueError(
654 "The provided fs_root is read-only and it must be read-write for package processor"
655 )
656 if not package_processor.applies_to(context.binary_package): 656 ↛ 657line 656 didn't jump to line 657 because the condition on line 656 was never true
657 raise ValueError(
658 f'The package processor "{package_processor_id}" from {self._plugin_name} does not apply'
659 " to the given package. Consider using `package_metadata_context()` to emulate a binary"
660 " package with the correct specification. As an example: "
661 '`package_metadata_context(package_fields={"Package-Type": "udeb"})` would emulate a udeb'
662 " package."
663 )
664 package_processor.run_package_processor(
665 fs_root,
666 None,
667 context,
668 )
670 @property
671 def declared_manifest_variables(self) -> frozenset[str]:
672 return frozenset(
673 {
674 k
675 for k, v in self._feature_set.manifest_variables.items()
676 if v.plugin_metadata.plugin_name == self._plugin_name
677 }
678 )
680 def automatic_discard_rules_examples_with_issues(self) -> Sequence[ADRExampleIssue]:
681 issues = []
682 for adr in self._feature_set.auto_discard_rules.values():
683 if adr.plugin_metadata.plugin_name != self._plugin_name: 683 ↛ 684line 683 didn't jump to line 684 because the condition on line 683 was never true
684 continue
685 for idx, example in enumerate(adr.examples):
686 result = process_discard_rule_example(
687 adr,
688 example,
689 )
690 if result.inconsistent_paths:
691 issues.append(
692 ADRExampleIssue(
693 adr.name,
694 idx,
695 [
696 x.absolute + ("/" if x.is_dir else "")
697 for x in result.inconsistent_paths
698 ],
699 )
700 )
701 return issues
703 def run_service_detection_and_integrations(
704 self,
705 service_manager: str,
706 fs_root: VirtualPath,
707 context: PackageProcessingContext | None = None,
708 *,
709 service_context_type_hint: type[DSD] | None = None,
710 ) -> tuple[list[DetectedService[DSD]], RegisteredMetadata]:
711 if not fs_root.is_root_dir(): 711 ↛ 712line 711 didn't jump to line 712 because the condition on line 711 was never true
712 raise ValueError("Provided path must be the file system root.")
713 try:
714 service_manager_details = self._feature_set.service_managers[
715 service_manager
716 ]
717 if service_manager_details.plugin_metadata.plugin_name != self._plugin_name: 717 ↛ 718line 717 didn't jump to line 718 because the condition on line 717 was never true
718 raise KeyError(service_manager)
719 except KeyError:
720 raise ValueError(
721 f"The plugin {self._plugin_name} does not provide a"
722 f" service manager called {service_manager}"
723 ) from None
725 if context is None: 725 ↛ 727line 725 didn't jump to line 727 because the condition on line 725 was always true
726 context = package_metadata_context()
727 detected_services: list[DetectedService[DSD]] = []
728 registry = ServiceRegistryTestImpl(service_manager_details, detected_services)
729 service_manager_details.service_detector(
730 fs_root,
731 registry,
732 context,
733 )
734 ctrl = BinaryCtrlAccessorTestProvider(
735 self._plugin_metadata,
736 service_manager_details.service_manager,
737 context,
738 )
739 if detected_services:
740 service_definitions = [
741 ServiceDefinitionImpl(
742 ds.names[0],
743 ds.names,
744 ds.path,
745 ds.type_of_service,
746 ds.service_scope,
747 ds.enable_by_default,
748 ds.start_by_default,
749 ds.default_upgrade_rule,
750 self._plugin_name,
751 True,
752 ds.service_context,
753 )
754 for ds in detected_services
755 ]
756 service_manager_details.service_integrator(
757 service_definitions,
758 ctrl,
759 context,
760 )
761 return detected_services, ctrl.registered_metadata()
763 def manifest_variables(
764 self,
765 *,
766 resolution_context: VariableContext | None = None,
767 mocked_variables: Mapping[str, str] | None = None,
768 ) -> Mapping[str, str]:
769 valid_manifest_variables = frozenset(
770 {
771 n
772 for n, v in self._feature_set.manifest_variables.items()
773 if v.plugin_metadata.plugin_name == self._plugin_name
774 }
775 )
776 if resolution_context is None:
777 resolution_context = manifest_variable_resolution_context()
778 substitution = self._substitution.copy_for_subst_test(
779 self._feature_set,
780 resolution_context,
781 extra_substitutions=mocked_variables,
782 )
783 return SubstitutionTable(
784 valid_manifest_variables,
785 substitution,
786 )
789class SubstitutionTable(Mapping[str, str]):
790 def __init__(
791 self, valid_manifest_variables: frozenset[str], substitution: Substitution
792 ) -> None:
793 self._valid_manifest_variables = valid_manifest_variables
794 self._resolved: set[str] = set()
795 self._substitution = substitution
797 def __contains__(self, item: object) -> bool:
798 return item in self._valid_manifest_variables
800 def __getitem__(self, key: str) -> str:
801 if key not in self._valid_manifest_variables: 801 ↛ 802line 801 didn't jump to line 802 because the condition on line 801 was never true
802 raise KeyError(key)
803 v = self._substitution.substitute(
804 "{{" + key + "}}", f"test of manifest variable `{key}`"
805 )
806 self._resolved.add(key)
807 return v
809 def __len__(self) -> int:
810 return len(self._valid_manifest_variables)
812 def __iter__(self) -> Iterator[str]:
813 return iter(self._valid_manifest_variables)
815 def keys(self) -> KeysView[str]:
816 return cast("KeysView[str]", self._valid_manifest_variables)