1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
|
import dataclasses
import os
from abc import ABCMeta
from typing import (
Iterable,
Mapping,
Callable,
Optional,
Union,
List,
Tuple,
Set,
Sequence,
Generic,
Type,
Self,
FrozenSet,
)
from debian.substvars import Substvars
from debputy import filesystem_scan
from debputy.plugin.api import (
VirtualPath,
PackageProcessingContext,
DpkgTriggerType,
Maintscript,
)
from debputy.plugin.api.impl_types import PluginProvidedTrigger
from debputy.plugin.api.spec import DSD, ServiceUpgradeRule, PathDef
from debputy.substitution import VariableContext
DEBPUTY_TEST_AGAINST_INSTALLED_PLUGINS = (
os.environ.get("DEBPUTY_TEST_PLUGIN_LOCATION", "uninstalled") == "installed"
)
@dataclasses.dataclass(slots=True, frozen=True)
class ADRExampleIssue:
name: str
example_index: int
inconsistent_paths: Sequence[str]
def build_virtual_file_system(
paths: Iterable[Union[str, PathDef]],
read_write_fs: bool = True,
) -> VirtualPath:
"""Create a pure-virtual file system for use with metadata detectors
This method will generate a virtual file system a list of path names or virtual path definitions. It will
also insert any implicit path required to make the file system connected. As an example:
>>> fs_root = build_virtual_file_system(['./usr/share/doc/package/copyright'])
>>> # The file we explicitly requested is obviously there
>>> fs_root.lookup('./usr/share/doc/package/copyright') is not None
True
>>> # but so is every directory up to that point
>>> all(fs_root.lookup(d).is_dir
... for d in ['./usr', './usr/share', './usr/share/doc', './usr/share/doc/package']
... )
True
Any string provided will be pased to `virtual_path` using all defaults for other parameters, making `str`
arguments a nice easy shorthand if you just want a path to exist, but do not really care about it otherwise
(or `virtual_path_def` defaults happens to work for you).
Here is a very small example of how to create some basic file system objects to get you started:
>>> from debputy.plugin.api import virtual_path_def
>>> path_defs = [
... './usr/share/doc/', # Create a directory
... virtual_path_def("./bin/zcat", link_target="/bin/gzip"), # Create a symlink
... virtual_path_def("./bin/gzip", mode=0o755), # Create a file (with a custom mode)
... ]
>>> fs_root = build_virtual_file_system(path_defs)
>>> fs_root.lookup('./usr/share/doc').is_dir
True
>>> fs_root.lookup('./bin/zcat').is_symlink
True
>>> fs_root.lookup('./bin/zcat').readlink() == '/bin/gzip'
True
>>> fs_root.lookup('./bin/gzip').is_file
True
>>> fs_root.lookup('./bin/gzip').mode == 0o755
True
:param paths: An iterable any mix of path names (str) and virtual_path_def definitions
(results from `virtual_path_def`).
:param read_write_fs: Whether the file system is read-write (True) or read-only (False).
Note that this is the default permission; the plugin test API may temporarily turn a
read-write to read-only temporarily (when running a metadata detector, etc.).
:return: The root of the generated file system
"""
return filesystem_scan.build_virtual_fs(paths, read_write_fs=read_write_fs)
@dataclasses.dataclass(slots=True, frozen=True)
class RegisteredTrigger:
dpkg_trigger_type: DpkgTriggerType
dpkg_trigger_target: str
def serialized_format(self) -> str:
"""The semantic contents of the DEBIAN/triggers file"""
return f"{self.dpkg_trigger_type} {self.dpkg_trigger_target}"
@classmethod
def from_plugin_provided_trigger(
cls,
plugin_provided_trigger: PluginProvidedTrigger,
) -> "Self":
return cls(
plugin_provided_trigger.dpkg_trigger_type,
plugin_provided_trigger.dpkg_trigger_target,
)
@dataclasses.dataclass(slots=True, frozen=True)
class RegisteredMaintscript:
"""Details about a maintscript registered by a plugin"""
"""Which maintscript is applies to (e.g., "postinst")"""
maintscript: Maintscript
"""Which method was used to trigger the script (e.g., "on_configure")"""
registration_method: str
"""The snippet provided by the plugin as it was provided
That is, no indentation/conditions/substitutions have been applied to this text
"""
plugin_provided_script: str
"""Whether substitutions would have been applied in a production run"""
requested_substitution: bool
@dataclasses.dataclass(slots=True, frozen=True)
class DetectedService(Generic[DSD]):
path: VirtualPath
names: Sequence[str]
type_of_service: str
service_scope: str
enable_by_default: bool
start_by_default: bool
default_upgrade_rule: ServiceUpgradeRule
service_context: Optional[DSD]
class RegisteredPackagerProvidedFile(metaclass=ABCMeta):
"""Record of a registered packager provided file - No instantiation
New "mandatory" attributes may be added in minor versions, which means instantiation will break tests.
Plugin providers should therefore not create instances of this dataclass. It is visible only to aid
test writing by providing type-safety / auto-completion.
"""
"""The name stem used for generating the file"""
stem: str
"""The recorded directory these file should be installed into"""
installed_path: str
"""The mode that debputy will give these files when installed (unless overridden)"""
default_mode: int
"""The default priority assigned to files unless overriden (if priories are assigned at all)"""
default_priority: Optional[int]
"""The filename format to be used"""
filename_format: Optional[str]
"""The formatting correcting callback"""
post_formatting_rewrite: Optional[Callable[[str], str]]
def compute_dest(
self,
assigned_name: str,
*,
assigned_priority: Optional[int] = None,
owning_package: Optional[str] = None,
path: Optional[VirtualPath] = None,
) -> Tuple[str, str]:
"""Determine the basename of this packager provided file
This method is useful for verifying that the `installed_path` and `post_formatting_rewrite` works
as intended. As example, some programs do not support "." in their configuration files, so you might
have a post_formatting_rewrite à la `lambda x: x.replace(".", "_")`. Then you can test it by
calling `assert rppf.compute_dest("python3.11")[1] == "python3_11"` to verify that if a package like
`python3.11` were to use this packager provided file, it would still generate a supported file name.
For the `assigned_name` parameter, then this is normally derived from the filename. Examples for
how to derive it:
* `debian/my-pkg.stem` => `my-pkg`
* `debian/my-pkg.my-custom-name.stem` => `my-custom-name`
Note that all parts (`my-pkg`, `my-custom-name` and `stem`) can contain periods (".") despite
also being a delimiter. Additionally, `my-custom-name` is not restricted to being a valid package
name, so it can have any file-system valid character in it.
For the 0.01% case, where the plugin is using *both* `{name}` *and* `{owning_package}` in the
installed_path, then you can separately *also* set the `owning_package` attribute. However, by
default the `assigned_named` is used for both when `owning_package` is not provided.
:param assigned_name: The name assigned. Usually this is the name of the package containing the file.
:param assigned_priority: Optionally a priority override for the file (if priority is supported). Must be
omitted/None if priorities are not supported.
:param owning_package: Optionally the name of the owning package. It is only needed for those exceedingly
rare cases where the `installed_path` contains both `{owning_package}` (usually in addition to `{name}`).
:param path: Special-case param, only needed for when testing a special `debputy` PPF..
:return: A tuple of the directory name and the basename (in that order) that combined makes up that path
that debputy would use.
"""
raise NotImplementedError
class RegisteredMetadata:
__slots__ = ()
@property
def substvars(self) -> Substvars:
"""Returns the Substvars
:return: The substvars in their current state.
"""
raise NotImplementedError
@property
def triggers(self) -> List[RegisteredTrigger]:
raise NotImplementedError
def maintscripts(
self,
*,
maintscript: Optional[Maintscript] = None,
) -> List[RegisteredMaintscript]:
"""Extract the maintscript provided by the given metadata detector
:param maintscript: If provided, only snippet registered for the given maintscript is returned. Can be
used to say "Give me all the 'postinst' snippets by this metadata detector", which can simplify
verification in some cases.
:return: A list of all matching maintscript registered by the metadata detector. If the detector has
not been run, then the list will be empty. If the metadata detector has been run multiple times,
then this is the aggregation of all the runs.
"""
raise NotImplementedError
class InitializedPluginUnderTest:
def packager_provided_files(self) -> Iterable[RegisteredPackagerProvidedFile]:
"""An iterable of all packager provided files registered by the plugin under test
If you want a particular order, please sort the result.
"""
return self.packager_provided_files_by_stem().values()
def packager_provided_files_by_stem(
self,
) -> Mapping[str, RegisteredPackagerProvidedFile]:
"""All packager provided files registered by the plugin under test grouped by name stem"""
raise NotImplementedError
def run_metadata_detector(
self,
metadata_detector_id: str,
fs_root: VirtualPath,
context: Optional[PackageProcessingContext] = None,
) -> RegisteredMetadata:
"""Run a metadata detector (by its ID) against a given file system
:param metadata_detector_id: The ID of the metadata detector to run
:param fs_root: The file system the metadata detector should see (must be the root of the file system)
:param context: The context the metadata detector should see. If not provided, one will be mock will be
provided to the extent possible.
:return: The metadata registered by the metadata detector
"""
raise NotImplementedError
def run_package_processor(
self,
package_processor_id: str,
fs_root: VirtualPath,
context: Optional[PackageProcessingContext] = None,
) -> None:
"""Run a package processor (by its ID) against a given file system
Note: Dependency processors are *not* run first.
:param package_processor_id: The ID of the package processor to run
:param fs_root: The file system the package processor should see (must be the root of the file system)
:param context: The context the package processor should see. If not provided, one will be mock will be
provided to the extent possible.
"""
raise NotImplementedError
@property
def declared_manifest_variables(self) -> Union[Set[str], FrozenSet[str]]:
"""Extract the manifest variables declared by the plugin
:return: All manifest variables declared by the plugin
"""
raise NotImplementedError
def automatic_discard_rules_examples_with_issues(self) -> Sequence[ADRExampleIssue]:
"""Validate examples of the automatic discard rules
For any failed example, use `debputy plugin show automatic-discard-rules <name>` to see
the failed example in full.
:return: If any examples have issues, this will return a non-empty sequence with an
entry with each issue.
"""
raise NotImplementedError
def run_service_detection_and_integrations(
self,
service_manager: str,
fs_root: VirtualPath,
context: Optional[PackageProcessingContext] = None,
*,
service_context_type_hint: Optional[Type[DSD]] = None,
) -> Tuple[List[DetectedService[DSD]], RegisteredMetadata]:
"""Run the service manager's detection logic and return the results
This method can be used to validate the service detection and integration logic of a plugin
for a given service manager.
First the service detector is run and if it finds any services, the integrator code is then
run on those services with their default values.
:param service_manager: The name of the service manager as provided during the initialization
:param fs_root: The file system the system detector should see (must be the root of
the file system)
:param context: The context the service detector should see. If not provided, one will be mock
will be provided to the extent possible.
:param service_context_type_hint: Unused; but can be used as a type hint for `mypy` (etc.)
to align the return type.
:return: A tuple of the list of all detected services in the provided file system and the
metadata generated by the integrator (if any services were detected).
"""
raise NotImplementedError
def manifest_variables(
self,
*,
resolution_context: Optional[VariableContext] = None,
mocked_variables: Optional[Mapping[str, str]] = None,
) -> Mapping[str, str]:
"""Provide a table of the manifest variables registered by the plugin
Each key is a manifest variable and the value of said key is the value of the manifest
variable. Lazy loaded variables are resolved when accessed for the first time and may
raise exceptions if the preconditions are not correct.
Note this method can be called multiple times with different parameters to provide
different contexts. Lazy loaded variables are resolved at most once per context.
:param resolution_context: An optional context for lazy loaded manifest variables.
Create an instance of it via `manifest_variable_resolution_context`.
:param mocked_variables: An optional mapping that provides values for certain manifest
variables. This can be used if you want a certain variable to have a certain value
for the test to be stable (or because the manifest variable you are mocking is from
another plugin, and you do not want to deal with the implementation details of how
it is set). Any variable that depends on the mocked variable will use the mocked
variable in the given context.
:return: A table of the manifest variables provided by the plugin. Note this table
only contains manifest variables registered by the plugin. Attempting to resolve
other variables (directly), such as mocked variables or from other plugins, will
trigger a `KeyError`.
"""
raise NotImplementedError
|