Coverage for src/debputy/plugin/api/spec.py: 87%

1import contextlib

2import dataclasses

3import os

4import tempfile

5import textwrap

6from typing import (

7 Iterable,

8 Optional,

9 Callable,

10 Literal,

11 Union,

12 Iterator,

13 overload,

14 FrozenSet,

15 Sequence,

16 TypeVar,

17 Any,

18 TYPE_CHECKING,

19 TextIO,

20 BinaryIO,

21 Generic,

22 ContextManager,

23 List,

24 Type,

25 Tuple,

26)

28from debian.substvars import Substvars

30from debputy import util

31from debputy.exceptions import TestPathWithNonExistentFSPathError, PureVirtualPathError

32from debputy.interpreter import Interpreter, extract_shebang_interpreter_from_file

33from debputy.manifest_parser.util import parse_symbolic_mode

34from debputy.packages import BinaryPackage

35from debputy.types import S

37if TYPE_CHECKING:

38 from debputy.manifest_parser.base_types import (

39 StaticFileSystemOwner,

40 StaticFileSystemGroup,

41 )

44PluginInitializationEntryPoint = Callable[["DebputyPluginInitializer"], None]

45MetadataAutoDetector = Callable[

46 ["VirtualPath", "BinaryCtrlAccessor", "PackageProcessingContext"], None

47]

48PackageProcessor = Callable[["VirtualPath", None, "PackageProcessingContext"], None]

49DpkgTriggerType = Literal[

50 "activate",

51 "activate-await",

52 "activate-noawait",

53 "interest",

54 "interest-await",

55 "interest-noawait",

56]

57Maintscript = Literal["postinst", "preinst", "prerm", "postrm"]

58PackageTypeSelector = Union[Literal["deb", "udeb"], Iterable[Literal["deb", "udeb"]]]

59ServiceUpgradeRule = Literal[

60 "do-nothing",

61 "reload",

62 "restart",

63 "stop-then-start",

64]

66DSD = TypeVar("DSD")

67ServiceDetector = Callable[

68 ["VirtualPath", "ServiceRegistry[DSD]", "PackageProcessingContext"],

69 None,

70]

71ServiceIntegrator = Callable[

72 [

73 Sequence["ServiceDefinition[DSD]"],

74 "BinaryCtrlAccessor",

75 "PackageProcessingContext",

76 ],

77 None,

78]

80PMT = TypeVar("PMT")

83@dataclasses.dataclass(slots=True, frozen=True)

84class PackagerProvidedFileReferenceDocumentation:

85 description: Optional[str] = None

86 format_documentation_uris: Sequence[str] = tuple()

88 def replace(self, **changes: Any) -> "PackagerProvidedFileReferenceDocumentation":

89 return dataclasses.replace(self, **changes)

92def packager_provided_file_reference_documentation(

93 *,

94 description: Optional[str] = None,

95 format_documentation_uris: Optional[Sequence[str]] = tuple(),

96) -> PackagerProvidedFileReferenceDocumentation:

97 """Provide documentation for a given packager provided file.

99 :param description: Textual description presented to the user.

100 :param format_documentation_uris: A sequence of URIs to documentation that describes

101 the format of the file. Most relevant first.

102 :return:

103 """

104 uris = tuple(format_documentation_uris) if format_documentation_uris else tuple()

105 return PackagerProvidedFileReferenceDocumentation(

106 description=description,

107 format_documentation_uris=uris,

108 )

109

110

111class PathMetadataReference(Generic[PMT]):

112 """An accessor to plugin provided metadata

113

114 This is a *short-lived* reference to a piece of metadata. It should *not* be stored beyond

115 the boundaries of the current plugin execution context as it can be become invalid (as an

116 example, if the path associated with this path is removed, then this reference become invalid)

117 """

118

119 @property

120 def is_present(self) -> bool:

121 """Determine whether the value has been set

122

123 If the current plugin cannot access the value, then this method unconditionally returns

124 `False` regardless of whether the value is there.

125

126 :return: `True` if the value has been set to a not None value (and not been deleted).

127 Otherwise, this property is `False`.

128 """

129 raise NotImplementedError

130

131 @property

132 def can_read(self) -> bool:

133 """Test whether it is possible to read the metadata

134

135 Note: That the metadata being readable does *not* imply that the metadata is present.

136

137 :return: True if it is possible to read the metadata. This is always True for the

138 owning plugin.

139 """

140 raise NotImplementedError

141

142 @property

143 def can_write(self) -> bool:

144 """Test whether it is possible to update the metadata

145

146 :return: True if it is possible to update the metadata.

147 """

148 raise NotImplementedError

149

150 @property

151 def value(self) -> Optional[PMT]:

152 """Fetch the currently stored value if present.

153

154 :return: The value previously stored if any. Returns `None` if the value was never

155 stored, explicitly set to `None` or was deleted.

156 """

157 raise NotImplementedError

158

159 @value.setter

160 def value(self, value: Optional[PMT]) -> None:

161 """Replace any current value with the provided value

162

163 This operation is only possible if the path is writable *and* the caller is from

164 the owning plugin OR the owning plugin made the reference read-write.

165 """

166 raise NotImplementedError

167

168 @value.deleter

169 def value(self) -> None:

170 """Delete any current value.

171

172 This has the same effect as setting the value to `None`. It has the same restrictions

173 as the value setter.

174 """

175 self.value = None

176

177

178@dataclasses.dataclass(slots=True)

179class PathDef:

180 path_name: str

181 mode: Optional[int] = None

182 mtime: Optional[int] = None

183 has_fs_path: Optional[bool] = None

184 fs_path: Optional[str] = None

185 link_target: Optional[str] = None

186 content: Optional[str] = None

187 materialized_content: Optional[str] = None

188

189

190def virtual_path_def(

191 path_name: str,

192 /,

193 mode: Optional[int] = None,

194 mtime: Optional[int] = None,

195 fs_path: Optional[str] = None,

196 link_target: Optional[str] = None,

197 content: Optional[str] = None,

198 materialized_content: Optional[str] = None,

199) -> PathDef:

200 """Define a virtual path for use with examples or, in tests, `build_virtual_file_system`

201

202 :param path_name: The full path. Must start with "./". If it ends with "/", the path will be interpreted

203 as a directory (the `is_dir` attribute will be True). Otherwise, it will be a symlink or file depending

204 on whether a `link_target` is provided.

205 :param mode: The mode to use for this path. Defaults to 0644 for files and 0755 for directories. The mode

206 should be None for symlinks.

207 :param mtime: Define the last modified time for this path. If not provided, debputy will provide a default

208 if the mtime attribute is accessed.

209 :param fs_path: Define a file system path for this path. This causes `has_fs_path` to return True and the

210 `fs_path` attribute will return this value. The test is required to make this path available to the extent

211 required. Note that the virtual file system will *not* examine the provided path in any way nor attempt

212 to resolve defaults from the path.

213 :param link_target: A target for the symlink. Providing a not None value for this parameter will make the

214 path a symlink.

215 :param content: The content of the path (if opened). The path must be a file.

216 :param materialized_content: Same as `content` except `debputy` will put the contents into a physical file

217 as needed. Cannot be used with `content` or `fs_path`.

218 :return: An *opaque* object to be passed to `build_virtual_file_system`. While the exact type is provided

219 to aid with typing, the type name and its behaviour is not part of the API.

220 """

221

222 is_dir = path_name.endswith("/")

223 is_symlink = link_target is not None

224

225 if is_symlink:

226 if mode is not None:

227 raise ValueError(

228 f'Please do not provide mode for symlinks. Triggered by "{path_name}"'

229 )

230 if is_dir:

231 raise ValueError(

232 "Path name looks like a directory, but a symlink target was also provided."

233 f' Please remove the trailing slash OR the symlink_target. Triggered by "{path_name}"'

234 )

235

236 if content and (is_dir or is_symlink): 236 ↛ 237line 236 didn't jump to line 237, because the condition on line 236 was never true

237 raise ValueError(

238 "Content was defined however, the path appears to be a directory a or a symlink"

239 f' Please remove the content, the trailing slash OR the symlink_target. Triggered by "{path_name}"'

240 )

241

242 if materialized_content is not None:

243 if content is not None: 243 ↛ 244line 243 didn't jump to line 244, because the condition on line 243 was never true

244 raise ValueError(

245 "The materialized_content keyword is mutually exclusive with the content keyword."

246 f' Triggered by "{path_name}"'

247 )

248 if fs_path is not None: 248 ↛ 249line 248 didn't jump to line 249, because the condition on line 248 was never true

249 raise ValueError(

250 "The materialized_content keyword is mutually exclusive with the fs_path keyword."

251 f' Triggered by "{path_name}"'

252 )

253 return PathDef(

254 path_name,

255 mode=mode,

256 mtime=mtime,

257 has_fs_path=bool(fs_path) or materialized_content is not None,

258 fs_path=fs_path,

259 link_target=link_target,

260 content=content,

261 materialized_content=materialized_content,

262 )

263

264

265class PackageProcessingContext:

266 """Context for auto-detectors of metadata and package processors (no instantiation)

267

268 This object holds some context related data for the metadata detector or/and package

269 processors. It may receive new attributes in the future.

270 """

271

272 __slots__ = ()

273

274 @property

275 def binary_package(self) -> BinaryPackage:

276 """The binary package stanza from `debian/control`"""

277 raise NotImplementedError

278

279 @property

280 def binary_package_version(self) -> str:

281 """The version of the binary package

282

283 Note this never includes the binNMU version for arch:all packages, but it may for arch:any.

284 """

285 raise NotImplementedError

286

287 @property

288 def related_udeb_package(self) -> Optional[BinaryPackage]:

289 """An udeb related to this binary package (if any)"""

290 raise NotImplementedError

291

292 @property

293 def related_udeb_package_version(self) -> Optional[str]:

294 """The version of the related udeb package (if present)

295

296 Note this never includes the binNMU version for arch:all packages, but it may for arch:any.

297 """

298 raise NotImplementedError

299

300 def accessible_package_roots(self) -> Iterable[Tuple[BinaryPackage, "VirtualPath"]]:

301 raise NotImplementedError

302

303 # """The source package stanza from `debian/control`"""

304 # source_package: SourcePackage

305

306

307class DebputyPluginInitializer:

308 __slots__ = ()

309

310 def packager_provided_file(

311 self,

312 stem: str,

313 installed_path: str,

314 *,

315 default_mode: int = 0o0644,

316 default_priority: Optional[int] = None,

317 allow_name_segment: bool = True,

318 allow_architecture_segment: bool = False,

319 post_formatting_rewrite: Optional[Callable[[str], str]] = None,

320 packageless_is_fallback_for_all_packages: bool = False,

321 reservation_only: bool = False,

322 reference_documentation: Optional[

323 PackagerProvidedFileReferenceDocumentation

324 ] = None,

325 ) -> None:

326 """Register a packager provided file (debian/<pkg>.foo)

327

328 Register a packager provided file that debputy should automatically detect and install for the

329 packager (example `debian/foo.tmpfiles` -> `debian/foo/usr/lib/tmpfiles.d/foo.conf`). A packager

330 provided file typically identified by a package prefix and a "stem" and by convention placed

331 in the `debian/` directory.

332

333 Like debhelper, debputy also supports the `foo.bar.tmpfiles` variant where the file is to be

334 installed into the `foo` package but be named after the `bar` segment rather than the package name.

335 This feature can be controlled via the `allow_name_segment` parameter.

336

337 :param stem: The "stem" of the file. This would be the `tmpfiles` part of `debian/foo.tmpfiles`.

338 Note that this value must be unique across all registered packager provided files.

339 :param installed_path: A format string describing where the file should be installed. Would be

340 `/usr/lib/tmpfiles.d/{name}.conf` from the example above.

341

342 The caller should provide a string with one or more of the placeholders listed below (usually `{name}`

343 should be one of them). The format affect the entire path.

344

345 The following placeholders are supported:

346 * `{name}` - The name in the name segment (defaulting the package name if no name segment is given)

347 * `{priority}` / `{priority:02}` - The priority of the file. Only provided priorities are used (that

348 is, default_priority is not None). The latter variant ensuring that the priority takes at least

349 two characters and the `0` character is left-padded for priorities that takes less than two

350 characters.

351 * `{owning_package}` - The name of the package. Should only be used when `{name}` alone is insufficient.

352 If you do not want the "name" segment in the first place, use `allow_name_segment=False` instead.

353

354 The path is always interpreted as relative to the binary package root.

355

356 :param default_mode: The mode the installed file should have by default. Common options are 0o0644 (the default)

357 or 0o0755 (for files that must be executable).

358 :param allow_architecture_segment: If True, the file may have an optional "architecture" segment at the end

359 (`foo.tmpfiles.amd64`), which marks it architecture specific. When False, debputy will detect the

360 "architecture" segment and report the use as an error. Note the architecture segment is only allowed for

361 arch:any packages. If a file targeting an arch:all package uses an architecture specific file it will

362 always result in an error.

363 :param allow_name_segment: If True, the file may have an optional "name" segment after the package name prefix.

364 (`foo.<name-here>.tmpfiles`). When False, debputy will detect the "name" segment and report the use as an

365 error.

366 :param default_priority: Special-case option for packager files that are installed into directories that have

367 "parse ordering" or "priority". These files will generally be installed as something like `20-foo.conf`

368 where the `20-` denotes their "priority". If the plugin is registering such a file type, then it should

369 provide a default priority.

370

371 The following placeholders are supported:

372 * `{name}` - The name in the name segment (defaulting the package name if no name segment is given)

373 * `{priority}` - The priority of the file. Only provided priorities are used (that is, default_priority

374 is not None)

375 * `{owning_package}` - The name of the package. Should only be used when `{name}` alone is insufficient.

376 If you do not want the "name" segment in the first place, use `allow_name_segment=False` instead.

377 :param post_formatting_rewrite: An optional "name correcting" callback. It receives the formatted name and can

378 do any transformation required. The primary use-case for this is to replace "forbidden" characters. The most

379 common case for debputy itself is to replace "." with "_" for tools that refuse to work with files containing

380 "." (`lambda x: x.replace(".", "_")`). The callback operates on basename of formatted version of the

381 `installed_path` and the callback should return the basename.

382 :param packageless_is_fallback_for_all_packages: If True, the packageless variant (such as, `debian/changelog`)

383 is a fallback for every package.

384 :param reference_documentation: Reference documentation for the packager provided file. Use the

385 packager_provided_file_reference_documentation function to provide the value for this parameter.

386 :param reservation_only: When True, tell debputy that the plugin reserves this packager provided file, but that

387 debputy should not actually install it automatically. This is useful in the cases, where the plugin

388 needs to process the file before installing it. The file will be marked as provided by this plugin. This

389 enables introspection and detects conflicts if other plugins attempts to claim the file.

390 """

391 raise NotImplementedError

392

393 def metadata_or_maintscript_detector(

394 self,

395 auto_detector_id: str,

396 auto_detector: MetadataAutoDetector,

397 *,

398 package_type: PackageTypeSelector = "deb",

399 ) -> None:

400 """Provide a pre-assembly hook that can affect the metadata/maintscript of binary ("deb") packages

401

402 The provided hook will be run once per binary package to be assembled, and it can see all the content

403 ("data.tar") planned to be included in the deb. The hook may do any *read-only* analysis of this content

404 and provide metadata, alter substvars or inject maintscript snippets. However, the hook must *not*

405 change the content ("data.tar") part of the deb.

406

407 The hook will be run unconditionally for all binary packages built. When the hook does not apply to all

408 packages, it must provide its own (internal) logic for detecting whether it is relevant and reduced itself

409 to a no-op if it should not apply to the current package.

410

411 Hooks are run in "some implementation defined order" and should not rely on being run before or after

412 any other hook.

413

414 The hooks are only applied to packages defined in `debian/control`. Notably, the metadata detector will

415 not apply to auto-generated `-dbgsym` packages (as those are not listed explicitly in `debian/control`).

416

417 :param auto_detector_id: A plugin-wide unique ID for this detector. Packagers may use this ID for disabling

418 the detector and accordingly the ID is part of the plugin's API toward the packager.

419 :param auto_detector: The code to be called that will be run at the metadata generation state (once for each

420 binary package).

421 :param package_type: Which kind of packages this metadata detector applies to. The package type is generally

422 defined by `Package-Type` field in the binary package. The default is to only run for regular `deb` packages

423 and ignore `udeb` packages.

424 """

425 raise NotImplementedError

426

427 def manifest_variable(

428 self,

429 variable_name: str,

430 value: str,

431 variable_reference_documentation: Optional[str] = None,

432 ) -> None:

433 """Provide a variable that can be used in the package manifest

434

435 >>> # Enable users to use "{{path:BASH_COMPLETION_DIR}}/foo" in their manifest.

436 >>> api.manifest_variable( # doctest: +SKIP

437 ... "path:BASH_COMPLETION_DIR",

438 ... "/usr/share/bash-completion/completions",

439 ... variable_reference_documentation="Directory to install bash completions into",

440 ... )

441

442 :param variable_name: The variable name.

443 :param value: The value the variable should resolve to.

444 :param variable_reference_documentation: A short snippet of reference documentation that explains

445 the purpose of the variable.

446 """

447 raise NotImplementedError

448

449

450class MaintscriptAccessor:

451 __slots__ = ()

452

453 def on_configure(

454 self,

455 run_snippet: str,

456 /,

457 indent: Optional[bool] = None,

458 perform_substitution: bool = True,

459 skip_on_rollback: bool = False,

460 ) -> None:

461 """Provide a snippet to be run when the package is about to be "configured"

462

463 This condition is the most common "post install" condition and covers the two

464 common cases:

465 * On initial install, OR

466 * On upgrade

467

468 In dpkg maintscript terms, this method roughly corresponds to postinst containing

469 `if [ "$1" = configure ]; then <snippet>; fi`

470

471 Additionally, the condition will by default also include rollback/abort scenarios such as "above-remove",

472 which is normally what you want but most people forget about.

473

474 :param run_snippet: The actual shell snippet to be run in the given condition. The snippet must be idempotent.

475 The snippet may contain newlines as necessary, which will make the result more readable. Additionally, the

476 snippet may contain '{{FOO}}' substitutions by default.

477 :param skip_on_rollback: By default, this condition will also cover common rollback scenarios. This

478 is normally what you want (or benign in most cases due to the idempotence requirement for maintscripts).

479 However, you can disable the rollback cases, leaving only "On initial install OR On upgrade".

480 :param indent: If True, the provided snippet will be indented to fit the condition provided by debputy.

481 In most cases, this is safe to do and provides more readable scripts. However, it may cause issues

482 with some special shell syntax (such as "Heredocs"). When False, the snippet will *not* be re-indented.

483 You are recommended to do 4 spaces of indentation when indent is False for readability.

484 :param perform_substitution: When True, `{{FOO}}` will be substituted in the snippet. When False, no

485 substitution is provided.

486 """

487 raise NotImplementedError

488

489 def on_initial_install(

490 self,

491 run_snippet: str,

492 /,

493 indent: Optional[bool] = None,

494 perform_substitution: bool = True,

495 ) -> None:

496 """Provide a snippet to be run when the package is about to be "configured" for the first time

497

498 The snippet will only be run on the first time the package is installed (ever or since last purge).

499 Note that "first" does not mean "exactly once" as dpkg does *not* provide such semantics. There are two

500 common cases where this can snippet can be run multiple times for the same system (and why the snippet

501 must still be idempotent):

502

503 1) The package is installed (1), then purged and then installed again (2). This can partly be mitigated

504 by having an `on_purge` script to do clean up.

505

506 2) As the package is installed, the `postinst` script terminates prematurely (Disk full, power loss, etc.).

507 The user resolves the problem and runs `dpkg --configure <pkg>`, which in turn restarts the script

508 from the beginning. This is why scripts must be idempotent in general.

509

510 In dpkg maintscript terms, this method roughly corresponds to postinst containing

511 `if [ "$1" = configure ] && [ -z "$2" ]; then <snippet>; fi`

512

513 :param run_snippet: The actual shell snippet to be run in the given condition. The snippet must be idempotent.

514 The snippet may contain newlines as necessary, which will make the result more readable. Additionally, the

515 snippet may contain '{{FOO}}' substitutions by default.

516 :param indent: If True, the provided snippet will be indented to fit the condition provided by debputy.

517 In most cases, this is safe to do and provides more readable scripts. However, it may cause issues

518 with some special shell syntax (such as "Heredocs"). When False, the snippet will *not* be re-indented.

519 You are recommended to do 4 spaces of indentation when indent is False for readability.

520 :param perform_substitution: When True, `{{FOO}}` will be substituted in the snippet. When False, no

521 substitution is provided.

522 """

523 raise NotImplementedError

524

525 def on_upgrade(

526 self,

527 run_snippet: str,

528 /,

529 indent: Optional[bool] = None,

530 perform_substitution: bool = True,

531 ) -> None:

532 """Provide a snippet to be run when the package is about to be "configured" after an upgrade

533

534 The snippet will only be run on any upgrade (that is, it will be skipped on the initial install).

535

536 In dpkg maintscript terms, this method roughly corresponds to postinst containing

537 `if [ "$1" = configure ] && [ -n "$2" ]; then <snippet>; fi`

538

539 :param run_snippet: The actual shell snippet to be run in the given condition. The snippet must be idempotent.

540 The snippet may contain newlines as necessary, which will make the result more readable. Additionally, the

541 snippet may contain '{{FOO}}' substitutions by default.

542 :param indent: If True, the provided snippet will be indented to fit the condition provided by debputy.

543 In most cases, this is safe to do and provides more readable scripts. However, it may cause issues

544 with some special shell syntax (such as "Heredocs"). When False, the snippet will *not* be re-indented.

545 You are recommended to do 4 spaces of indentation when indent is False for readability.

546 :param perform_substitution: When True, `{{FOO}}` will be substituted in the snippet. When False, no

547 substitution is provided.

548 """

549 raise NotImplementedError

550

551 def on_upgrade_from(

552 self,

553 version: str,

554 run_snippet: str,

555 /,

556 indent: Optional[bool] = None,

557 perform_substitution: bool = True,

558 ) -> None:

559 """Provide a snippet to be run when the package is about to be "configured" after an upgrade from a given version

560

561 The snippet will only be run on any upgrade (that is, it will be skipped on the initial install).

562

563 In dpkg maintscript terms, this method roughly corresponds to postinst containing

564 `if [ "$1" = configure ] && dpkg --compare-versions le-nl "$2" ; then <snippet>; fi`

565

566 :param version: The version to upgrade from

567 :param run_snippet: The actual shell snippet to be run in the given condition. The snippet must be idempotent.

568 The snippet may contain newlines as necessary, which will make the result more readable. Additionally, the

569 snippet may contain '{{FOO}}' substitutions by default.

570 :param indent: If True, the provided snippet will be indented to fit the condition provided by debputy.

571 In most cases, this is safe to do and provides more readable scripts. However, it may cause issues

572 with some special shell syntax (such as "Heredocs"). When False, the snippet will *not* be re-indented.

573 You are recommended to do 4 spaces of indentation when indent is False for readability.

574 :param perform_substitution: When True, `{{FOO}}` will be substituted in the snippet. When False, no

575 substitution is provided.

576 """

577 raise NotImplementedError

578

579 def on_before_removal(

580 self,

581 run_snippet: str,

582 /,

583 indent: Optional[bool] = None,

584 perform_substitution: bool = True,

585 ) -> None:

586 """Provide a snippet to be run when the package is about to be removed

587

588 The snippet will be run before dpkg removes any files.

589

590 In dpkg maintscript terms, this method roughly corresponds to prerm containing

591 `if [ "$1" = remove ] ; then <snippet>; fi`

592

593 :param run_snippet: The actual shell snippet to be run in the given condition. The snippet must be idempotent.

594 The snippet may contain newlines as necessary, which will make the result more readable. Additionally, the

595 snippet may contain '{{FOO}}' substitutions by default.

596 :param indent: If True, the provided snippet will be indented to fit the condition provided by debputy.

597 In most cases, this is safe to do and provides more readable scripts. However, it may cause issues

598 with some special shell syntax (such as "Heredocs"). When False, the snippet will *not* be re-indented.

599 You are recommended to do 4 spaces of indentation when indent is False for readability.

600 :param perform_substitution: When True, `{{FOO}}` will be substituted in the snippet. When False, no

601 substitution is provided.

602 """

603 raise NotImplementedError

604

605 def on_removed(

606 self,

607 run_snippet: str,

608 /,

609 indent: Optional[bool] = None,

610 perform_substitution: bool = True,

611 ) -> None:

612 """Provide a snippet to be run when the package has been removed

613

614 The snippet will be run after dpkg removes the package content from the file system.

615

616 **WARNING**: The snippet *cannot* rely on dependencies and must rely on `Essential: yes` packages.

617

618 In dpkg maintscript terms, this method roughly corresponds to postrm containing

619 `if [ "$1" = remove ] ; then <snippet>; fi`

620

621 :param run_snippet: The actual shell snippet to be run in the given condition. The snippet must be idempotent.

622 The snippet may contain newlines as necessary, which will make the result more readable. Additionally, the

623 snippet may contain '{{FOO}}' substitutions by default.

624 :param indent: If True, the provided snippet will be indented to fit the condition provided by debputy.

625 In most cases, this is safe to do and provides more readable scripts. However, it may cause issues

626 with some special shell syntax (such as "Heredocs"). When False, the snippet will *not* be re-indented.

627 You are recommended to do 4 spaces of indentation when indent is False for readability.

628 :param perform_substitution: When True, `{{FOO}}` will be substituted in the snippet. When False, no

629 substitution is provided.

630 """

631 raise NotImplementedError

632

633 def on_purge(

634 self,

635 run_snippet: str,

636 /,

637 indent: Optional[bool] = None,

638 perform_substitution: bool = True,

639 ) -> None:

640 """Provide a snippet to be run when the package is being purged.

641

642 The snippet will when the package is purged from the system.

643

644 **WARNING**: The snippet *cannot* rely on dependencies and must rely on `Essential: yes` packages.

645

646 In dpkg maintscript terms, this method roughly corresponds to postrm containing

647 `if [ "$1" = purge ] ; then <snippet>; fi`

648

649 :param run_snippet: The actual shell snippet to be run in the given condition. The snippet must be idempotent.

650 The snippet may contain newlines as necessary, which will make the result more readable. Additionally, the

651 snippet may contain '{{FOO}}' substitutions by default.

652 :param indent: If True, the provided snippet will be indented to fit the condition provided by debputy.

653 In most cases, this is safe to do and provides more readable scripts. However, it may cause issues

654 with some special shell syntax (such as "Heredocs"). When False, the snippet will *not* be re-indented.

655 You are recommended to do 4 spaces of indentation when indent is False for readability.

656 :param perform_substitution: When True, `{{FOO}}` will be substituted in the snippet. When False, no

657 substitution is provided.

658 """

659 raise NotImplementedError

660

661 def unconditionally_in_script(

662 self,

663 maintscript: Maintscript,

664 run_snippet: str,

665 /,

666 perform_substitution: bool = True,

667 ) -> None:

668 """Provide a snippet to be run in a given script

669

670 Run a given snippet unconditionally from a given script. The snippet must contain its own conditional

671 for when it should be run.

672

673 :param maintscript: The maintscript to insert the snippet into.

674 :param run_snippet: The actual shell snippet to be run. The snippet will be run unconditionally and should

675 contain its own conditions as necessary. The snippet must be idempotent. The snippet may contain newlines

676 as necessary, which will make the result more readable. Additionally, the snippet may contain '{{FOO}}'

677 substitutions by default.

678 :param perform_substitution: When True, `{{FOO}}` will be substituted in the snippet. When False, no

679 substitution is provided.

680 """

681 raise NotImplementedError

682

683 def escape_shell_words(self, *args: str) -> str:

684 """Provide sh-shell escape of strings

685

686 `assert escape_shell("foo", "fu bar", "baz") == 'foo "fu bar" baz'`

687

688 This is useful for ensuring file names and other "input" are considered one parameter even when they

689 contain spaces or shell meta-characters.

690

691 :param args: The string(s) to be escaped.

692 :return: Each argument escaped such that each argument becomes a single "word" and then all these words are

693 joined by a single space.

694 """

695 return util.escape_shell(*args)

696

697

698class BinaryCtrlAccessor:

699 __slots__ = ()

700

701 def dpkg_trigger(self, trigger_type: DpkgTriggerType, trigger_target: str) -> None:

702 """Register a declarative dpkg level trigger

703

704 The provided trigger will be added to the package's metadata (the triggers file of the control.tar).

705

706 If the trigger has already been added previously, a second call with the same trigger data will be ignored.

707 """

708 raise NotImplementedError

709

710 @property

711 def maintscript(self) -> MaintscriptAccessor:

712 """Attribute for manipulating maintscripts"""

713 raise NotImplementedError

714

715 @property

716 def substvars(self) -> "FlushableSubstvars":

717 """Attribute for manipulating dpkg substvars (deb-substvars)"""

718 raise NotImplementedError

719

720

721class VirtualPath:

722 __slots__ = ()

723

724 @property

725 def name(self) -> str:

726 """Basename of the path a.k.a. last segment of the path

727

728 In a path "usr/share/doc/pkg/changelog.gz" the basename is "changelog.gz".

729

730 For a directory, the basename *never* ends with a `/`.

731 """

732 raise NotImplementedError

733

734 @property

735 def iterdir(self) -> Iterable["VirtualPath"]:

736 """Returns an iterable that iterates over all children of this path

737

738 For directories, this returns an iterable of all children. For non-directories,

739 the iterable is always empty.

740 """

741 raise NotImplementedError

742

743 def lookup(self, path: str) -> Optional["VirtualPath"]:

744 """Perform a path lookup relative to this path

745

746 As an example `doc_dir = fs_root.lookup('./usr/share/doc')`

747

748 If the provided path starts with `/`, then the lookup is performed relative to the

749 file system root. That is, you can assume the following to always be True:

750

751 `fs_root.lookup("usr") == any_path_beneath_fs_root.lookup('/usr')`

752

753 Note: This method requires the path to be attached (see `is_detached`) regardless of

754 whether the lookup is relative or absolute.

755

756 If the path traverse a symlink, the symlink will be resolved.

757

758 :param path: The path to look. Can contain "." and ".." segments. If starting with `/`,

759 look up is performed relative to the file system root, otherwise the lookup is relative

760 to this path.

761 :return: The path object for the desired path if it can be found. Otherwise, None.

762 """

763 raise NotImplementedError

764

765 def all_paths(self) -> Iterable["VirtualPath"]:

766 """Iterate over this path and all of its descendants (if any)

767

768 If used on the root path, then every path in the package is returned.

769

770 The iterable is ordered, so using the order in output will be produce

771 bit-for-bit reproducible output. Additionally, a directory will always

772 be seen before its descendants. Otherwise, the order is implementation

773 defined.

774

775 The iteration is lazy and as a side effect do account for some obvious

776 mutation. Like if the current path is removed, then none of its children

777 will be returned (provided mutation happens before the lazy iteration

778 was required to resolve it). Likewise, mutation of the directory will

779 also work (again, provided mutation happens before the lazy iteration order).

780

781 :return: An ordered iterable of this path followed by its descendants.

782 """

783 raise NotImplementedError

784

785 @property

786 def is_detached(self) -> bool:

787 """Returns True if this path is detached

788

789 Paths that are detached from the file system will not be present in the package and

790 most operations are unsafe on them. This usually only happens if the path or one of

791 its parent directories are unlinked (rm'ed) from the file system tree.

792

793 All paths are attached by default and will only become detached as a result of

794 an action to mutate the virtual file system. Note that the file system may not

795 always be manipulated.

796

797 :return: True if the entry is detached. Detached entries should be discarded, so they

798 can be garbage collected.

799 """

800 raise NotImplementedError

801

802 # The __getitem__ behaves like __getitem__ from Dict but __iter__ would ideally work like a Sequence.

803 # However, that does not feel compatible, so lets force people to use .children instead for the Sequence

804 # behaviour to avoid surprises for now.

805 # (Maybe it is a non-issue, but it is easier to add the API later than to remove it once we have committed

806 # to using it)

807 __iter__ = None

808

809 def __getitem__(self, key: object) -> "VirtualPath":

810 """Lookup a (direct) child by name

811

812 Ignoring the possible `KeyError`, then the following are the same:

813 `fs_root["usr"] == fs_root.lookup('usr')`

814

815 Note that unlike `.lookup` this can only locate direct children.

816 """

817 raise NotImplementedError

818

819 def __delitem__(self, key) -> None:

820 """Remove a child from this node if it exists

821

822 If that child is a directory, then the entire tree is removed (like `rm -fr`).

823 """

824 raise NotImplementedError

825

826 def get(self, key: str) -> "Optional[VirtualPath]":

827 """Lookup a (direct) child by name

828

829 The following are the same:

830 `fs_root.get("usr") == fs_root.lookup('usr')`

831

832 Note that unlike `.lookup` this can only locate direct children.

833 """

834 try:

835 return self[key]

836 except KeyError:

837 return None

838

839 def __contains__(self, item: object) -> bool:

840 """Determine if this path includes a given child (either by object or string)

841

842 Examples:

843

844 if 'foo' in dir: ...

845 """

846 if isinstance(item, VirtualPath):

847 return item.parent_dir is self

848 if not isinstance(item, str):

849 return False

850 m = self.get(item)

851 return m is not None

852

853 @property

854 def path(self) -> str:

855 """Returns the "full" path for this file system entry

856

857 This is the path that debputy uses to refer to this file system entry. It is always

858 normalized. Use the `absolute` attribute for how the path looks

859 when the package is installed. Alternatively, there is also `fs_path`, which is the

860 path to the underlying file system object (assuming there is one). That is the one

861 you need if you want to read the file.

862

863 This is attribute is mostly useful for debugging or for looking up the path relative

864 to the "root" of the virtual file system that debputy maintains.

865

866 If the path is detached (see `is_detached`), then this method returns the path as it

867 was known prior to being detached.

868 """

869 raise NotImplementedError

870

871 @property

872 def absolute(self) -> str:

873 """Returns the absolute version of this path

874

875 This is how to refer to this path when the package is installed.

876

877 If the path is detached (see `is_detached`), then this method returns the last known location

878 of installation (prior to being detached).

879

880 :return: The absolute path of this file as it would be on the installed system.

881 """

882 p = self.path.lstrip(".")

883 if not p.startswith("/"):

884 return f"/{p}"

885 return p

886

887 @property

888 def parent_dir(self) -> Optional["VirtualPath"]:

889 """The parent directory of this path

890

891 Note this operation requires the path is "attached" (see `is_detached`). All paths are attached

892 by default but unlinking paths will cause them to become detached.

893

894 :return: The parent path or None for the root.

895 """

896 raise NotImplementedError

897

898 def stat(self) -> os.stat_result:

899 """Attempt to do stat of the underlying path (if it exists)

900

901 *Avoid* using `stat()` whenever possible where a more specialized attribute exist. The

902 `stat()` call returns the data from the file system and often, `debputy` does *not* track

903 its state in the file system. As an example, if you want to know the file system mode of

904 a path, please use the `mode` attribute instead.

905

906 This never follow symlinks (it behaves like `os.lstat`). It will raise an error

907 if the path is not backed by a file system object (that is, `has_fs_path` is False).

908

909 :return: The stat result or an error.

910 """

911 raise NotImplementedError()

912

913 @property

914 def size(self) -> int:

915 """Resolve the file size (`st_size`)

916

917 This may be using `stat()` and therefore `fs_path`.

918

919 :return: The size of the file in bytes

920 """

921 return self.stat().st_size

922

923 @property

924 def mode(self) -> int:

925 """Determine the mode bits of this path object

926

927 Note that:

928 * like with `stat` above, this never follows symlinks.

929 * the mode returned by this method is not always a 1:1 with the mode in the

930 physical file system. As an optimization, `debputy` skips unnecessary writes

931 to the underlying file system in many cases.

932

933

934 :return: The mode bits for the path.

935 """

936 raise NotImplementedError

937

938 @mode.setter

939 def mode(self, new_mode: int) -> None:

940 """Set the octal file mode of this path

941

942 Note that:

943 * this operation will fail if `path.is_read_write` returns False.

944 * this operation is generally *not* synced to the physical file system (as

945 an optimization).

946

947 :param new_mode: The new octal mode for this path. Note that `debputy` insists

948 that all paths have the `user read bit` and, for directories also, the

949 `user execute bit`. The absence of these minimal mode bits causes hard to

950 debug errors.

951 """

952 raise NotImplementedError

953

954 @property

955 def is_executable(self) -> bool:

956 """Determine whether a path is considered executable

957

958 Generally, this means that at least one executable bit is set. This will

959 basically always be true for directories as directories need the execute

960 parameter to be traversable.

961

962 :return: True if the path is considered executable with its current mode

963 """

964 return bool(self.mode & 0o0111)

965

966 def chmod(self, new_mode: Union[int, str]) -> None:

967 """Set the file mode of this path

968

969 This is similar to setting the `mode` attribute. However, this method accepts

970 a string argument, which will be parsed as a symbolic mode (example: `u+rX,go=rX`).

971

972 Note that:

973 * this operation will fail if `path.is_read_write` returns False.

974 * this operation is generally *not* synced to the physical file system (as

975 an optimization).

976

977 :param new_mode: The new mode for this path.

978 Note that `debputy` insists that all paths have the `user read bit` and, for

979 directories also, the `user execute bit`. The absence of these minimal mode

980 bits causes hard to debug errors.

981 """

982 if isinstance(new_mode, str):

983 segments = parse_symbolic_mode(new_mode, None)

984 final_mode = self.mode

985 is_dir = self.is_dir

986 for segment in segments:

987 final_mode = segment.apply(final_mode, is_dir)

988 self.mode = final_mode

989 else:

990 self.mode = new_mode

991

992 def chown(

993 self,

994 owner: Optional["StaticFileSystemOwner"],

995 group: Optional["StaticFileSystemGroup"],

996 ) -> None:

997 """Change the owner/group of this path

998

999 :param owner: The desired owner definition for this path. If None, then no change of owner is performed.

1000 :param group: The desired group definition for this path. If None, then no change of group is performed.

1001 """

1002 raise NotImplementedError

1003

1004 @property

1005 def mtime(self) -> float:

1006 """Determine the mtime of this path object

1007

1008 Note that:

1009 * like with `stat` above, this never follows symlinks.

1010 * the mtime returned has *not* been clamped against ´SOURCE_DATE_EPOCH`. Timestamp

1011 normalization is handled later by `debputy`.

1012 * the mtime returned by this method is not always a 1:1 with the mtime in the

1013 physical file system. As an optimization, `debputy` skips unnecessary writes

1014 to the underlying file system in many cases.

1015

1016 :return: The mtime for the path.

1017 """

1018 raise NotImplementedError

1019

1020 @mtime.setter

1021 def mtime(self, new_mtime: float) -> None:

1022 """Set the mtime of this path

1023

1024 Note that:

1025 * this operation will fail if `path.is_read_write` returns False.

1026 * this operation is generally *not* synced to the physical file system (as

1027 an optimization).

1028

1029 :param new_mtime: The new mtime of this path. Note that the caller does not need to

1030 account for `SOURCE_DATE_EPOCH`. Timestamp normalization is handled later.

1031 """

1032 raise NotImplementedError

1033

1034 def readlink(self) -> str:

1035 """Determine the link target of this path assuming it is a symlink

1036

1037 For paths where `is_symlink` is True, this already returns a link target even when

1038 `has_fs_path` is False.

1039

1040 :return: The link target of the path or an error is this is not a symlink

1041 """

1042 raise NotImplementedError()

1043

1044 @overload

1045 def open( 1045 ↛ exitline 1045 didn't jump to the function exit

1046 self,

1047 *,

1048 byte_io: Literal[False] = False,

1049 buffering: Optional[int] = ...,

1050 ) -> TextIO: ...

1051

1052 @overload

1053 def open( 1053 ↛ exitline 1053 didn't jump to the function exit

1054 self,

1055 *,

1056 byte_io: Literal[True],

1057 buffering: Optional[int] = ...,

1058 ) -> BinaryIO: ...

1059

1060 @overload

1061 def open( 1061 ↛ exitline 1061 didn't jump to the function exit

1062 self,

1063 *,

1064 byte_io: bool,

1065 buffering: Optional[int] = ...,

1066 ) -> Union[TextIO, BinaryIO]: ...

1067

1068 def open(

1069 self,

1070 *,

1071 byte_io: bool = False,

1072 buffering: int = -1,

1073 ) -> Union[TextIO, BinaryIO]:

1074 """Open the file for reading. Usually used with a context manager

1075

1076 By default, the file is opened in text mode (utf-8). Binary mode can be requested

1077 via the `byte_io` parameter. This operation is only valid for files (`is_file` returns

1078 `True`). Usage on symlinks and directories will raise exceptions.

1079

1080 This method *often* requires the `fs_path` to be present. However, tests as a notable

1081 case can inject content without having the `fs_path` point to a real file. (To be clear,

1082 such tests are generally expected to ensure `has_fs_path` returns `True`).

1083

1084

1085 :param byte_io: If True, open the file in binary mode (like `rb` for `open`)

1086 :param buffering: Same as open(..., buffering=...) where supported. Notably during

1087 testing, the content may be purely in memory and use a BytesIO/StringIO

1088 (which does not accept that parameter, but then is buffered in a different way)

1089 :return: The file handle.

1090 """

1091

1092 if not self.is_file: 1092 ↛ 1093line 1092 didn't jump to line 1093, because the condition on line 1092 was never true

1093 raise TypeError(f"Cannot open {self.path} for reading: It is not a file")

1094

1095 if byte_io:

1096 return open(self.fs_path, "rb", buffering=buffering)

1097 return open(self.fs_path, "rt", encoding="utf-8", buffering=buffering)

1098

1099 @property

1100 def fs_path(self) -> str:

1101 """Request the underling fs_path of this path

1102

1103 Only available when `has_fs_path` is True. Generally this should only be used for files to read

1104 the contents of the file and do some action based on the parsed result.

1105

1106 The path should only be used for read-only purposes as debputy may assume that it is safe to have

1107 multiple paths pointing to the same file system path.

1108

1109 Note that:

1110 * This is often *not* available for directories and symlinks.

1111 * The debputy in-memory file system overrules the physical file system. Attempting to "fix" things

1112 by using `os.chmod` or `os.unlink`'ing files, etc. will generally not do as you expect. Best case,

1113 your actions are ignored and worst case it will cause the build to fail as it violates debputy's

1114 internal invariants.

1115

1116 :return: The path to the underlying file system object on the build system or an error if no such

1117 file exist (see `has_fs_path`).

1118 """

1119 raise NotImplementedError()

1120

1121 @property

1122 def is_dir(self) -> bool:

1123 """Determine if this path is a directory

1124

1125 Never follows symlinks.

1126

1127 :return: True if this path is a directory. False otherwise.

1128 """

1129 raise NotImplementedError()

1130

1131 @property

1132 def is_file(self) -> bool:

1133 """Determine if this path is a directory

1134

1135 Never follows symlinks.

1136

1137 :return: True if this path is a regular file. False otherwise.

1138 """

1139 raise NotImplementedError()

1140

1141 @property

1142 def is_symlink(self) -> bool:

1143 """Determine if this path is a symlink

1144

1145 :return: True if this path is a symlink. False otherwise.

1146 """

1147 raise NotImplementedError()

1148

1149 @property

1150 def has_fs_path(self) -> bool:

1151 """Determine whether this path is backed by a file system path

1152

1153 :return: True if this path is backed by a file system object on the build system.

1154 """

1155 raise NotImplementedError()

1156

1157 @property

1158 def is_read_write(self) -> bool:

1159 """When true, the file system entry may be mutated

1160

1161 Read-write rules are:

1162

1163 +--------------------------+-------------------+------------------------+

1164 | File system | From / Inside | Read-Only / Read-Write |

1165 +--------------------------+-------------------+------------------------+

1166 | Source directory | Any context | Read-Only |

1167 | Binary staging directory | Package Processor | Read-Write |

1168 | Binary staging directory | Metadata Detector | Read-Only |

1169 +--------------------------+-------------------+------------------------+

1170

1171 These rules apply to the virtual file system (`debputy` cannot enforce

1172 these rules in the underlying file system). The `debputy` code relies

1173 on these rules for its logic in multiple places to catch bugs and for

1174 optimizations.

1175

1176 As an example, the reason why the file system is read-only when Metadata

1177 Detectors are run is based the contents of the file system has already

1178 been committed. New files will not be included, removals of existing

1179 files will trigger a hard error when the package is assembled, etc.

1180 To avoid people spending hours debugging why their code does not work

1181 as intended, `debputy` instead throws a hard error if you try to mutate

1182 the file system when it is read-only mode to "fail fast".

1183

1184 :return: Whether file system mutations are permitted.

1185 """

1186 return False

1187

1188 def mkdir(self, name: str) -> "VirtualPath":

1189 """Create a new subdirectory of the current path

1190

1191 :param name: Basename of the new directory. The directory must not contain a path

1192 with this basename.

1193 :return: The new subdirectory

1194 """

1195 raise NotImplementedError

1196

1197 def mkdirs(self, path: str) -> "VirtualPath":

1198 """Ensure a given path exists and is a directory.

1199

1200 :param path: Path to the directory to create. Any parent directories will be

1201 created as needed. If the path already exists and is a directory, then it

1202 is returned. If any part of the path exists and that is not a directory,

1203 then the `mkdirs` call will raise an error.

1204 :return: The directory denoted by the given path

1205 """

1206 raise NotImplementedError

1207

1208 def add_file(

1209 self,

1210 name: str,

1211 *,

1212 unlink_if_exists: bool = True,

1213 use_fs_path_mode: bool = False,

1214 mode: int = 0o0644,

1215 mtime: Optional[float] = None,

1216 ) -> ContextManager["VirtualPath"]:

1217 """Add a new regular file as a child of this path

1218

1219 This method will insert a new file into the virtual file system as a child

1220 of the current path (which must be a directory). The caller must use the

1221 return value as a context manager (see example). During the life-cycle of

1222 the managed context, the caller can fill out the contents of the file

1223 from the new path's `fs_path` attribute. The `fs_path` will exist as an

1224 empty file when the context manager is entered.

1225

1226 Once the context manager exits, mutation of the `fs_path` is no longer permitted.

1227

1228 >>> import subprocess

1229 >>> path = ... # doctest: +SKIP

1230 >>> with path.add_file("foo") as new_file, open(new_file.fs_path, "w") as fd: # doctest: +SKIP

1231 ... fd.writelines(["Some", "Content", "Here"])

1232

1233 The caller can replace the provided `fs_path` entirely provided at the end result

1234 (when the context manager exits) is a regular file with no hard links.

1235

1236 Note that this operation will fail if `path.is_read_write` returns False.

1237

1238 :param name: Basename of the new file

1239 :param unlink_if_exists: If the name was already in use, then either an exception is thrown

1240 (when `unlink_if_exists` is False) or the path will be removed via ´unlink(recursive=False)`

1241 (when `unlink_if_exists` is True)

1242 :param use_fs_path_mode: When True, the file created will have this mode in the physical file

1243 system. When the context manager exists, `debputy` will refresh its mode to match the mode

1244 in the physical file system. This is primarily useful if the caller uses a subprocess to

1245 mutate the path and the file mode is relevant for this tool (either as input or output).

1246 When the parameter is false, the new file is guaranteed to be readable and writable for

1247 the current user. However, no other guarantees are given (not even that it matches the

1248 `mode` parameter and any changes to the mode in the physical file system will be ignored.

1249 :param mode: This is the initial file mode. Note the `use_fs_path_mode` parameter for how

1250 this interacts with the physical file system.

1251 :param mtime: If the caller has a more accurate mtime than the mtime of the generated file,

1252 then it can be provided here. Note that all mtimes will later be clamped based on

1253 `SOURCE_DATE_EPOCH`. This parameter is only for when the conceptual mtime of this path

1254 should be earlier than `SOURCE_DATE_EPOCH`.

1255 :return: A Context manager that upon entering provides a `VirtualPath` instance for the

1256 new file. The instance remains valid after the context manager exits (assuming it exits

1257 successfully), but the file denoted by `fs_path` must not be changed after the context

1258 manager exits

1259 """

1260 raise NotImplementedError

1261

1262 def replace_fs_path_content(

1263 self,

1264 *,

1265 use_fs_path_mode: bool = False,

1266 ) -> ContextManager[str]:

1267 """Replace the contents of this file via inline manipulation

1268

1269 Used as a context manager to provide the fs path for manipulation.

1270

1271 Example:

1272 >>> import subprocess

1273 >>> path = ... # doctest: +SKIP

1274 >>> with path.replace_fs_path_content() as fs_path: # doctest: +SKIP

1275 ... subprocess.check_call(['strip', fs_path]) # doctest: +SKIP

1276

1277 The provided file system path should be manipulated inline. The debputy framework may

1278 copy it first as necessary and therefore the provided fs_path may be different from

1279 `path.fs_path` prior to entering the context manager.

1280

1281 Note that this operation will fail if `path.is_read_write` returns False.

1282

1283 If the mutation causes the returned `fs_path` to be a non-file or a hard-linked file

1284 when the context manager exits, `debputy` will raise an error at that point. To preserve

1285 the internal invariants of `debputy`, the path will be unlinked as `debputy` cannot

1286 reliably restore the path.

1287

1288 :param use_fs_path_mode: If True, any changes to the mode on the physical FS path will be

1289 recorded as the desired mode of the file when the contextmanager ends. The provided FS path

1290 with start with the current mode when `use_fs_path_mode` is True. Otherwise, `debputy` will

1291 ignore the mode of the file system entry and reuse its own current mode

1292 definition.

1293 :return: A Context manager that upon entering provides the path to a muable (copy) of

1294 this path's `fs_path` attribute. The file on the underlying path may be mutated however

1295 the caller wishes until the context manager exits.

1296 """

1297 raise NotImplementedError

1298

1299 def add_symlink(self, link_name: str, link_target: str) -> "VirtualPath":

1300 """Add a new regular file as a child of this path

1301

1302 This will create a new symlink inside the current path. If the path already exists,

1303 the existing path will be unlinked via `unlink(recursive=False)`.

1304

1305 Note that this operation will fail if `path.is_read_write` returns False.

1306

1307 :param link_name: The basename of the link file entry.

1308 :param link_target: The target of the link. Link target normalization will

1309 be handled by `debputy`, so the caller can use relative or absolute paths.

1310 (At the time of writing, symlink target normalization happens late)

1311 :return: The newly created symlink.

1312 """

1313 raise NotImplementedError

1314

1315 def unlink(self, *, recursive: bool = False) -> None:

1316 """Unlink a file or a directory

1317

1318 This operation will remove the path from the file system (causing `is_detached` to return True).

1319

1320 When the path is a:

1321

1322 * symlink, then the symlink itself is removed. The target (if present) is not affected.

1323 * *non-empty* directory, then the `recursive` parameter decides the outcome. An empty

1324 directory will be removed regardless of the value of `recursive`.

1325

1326 Note that:

1327 * the root directory cannot be deleted.

1328 * this operation will fail if `path.is_read_write` returns False.

1329

1330 :param recursive: If True, then non-empty directories will be unlinked as well removing everything inside them

1331 as well. When False, an error is raised if the path is a non-empty directory

1332 """

1333 raise NotImplementedError

1334

1335 def interpreter(self) -> Optional[Interpreter]:

1336 """Determine the interpreter of the file (`#!`-line details)

1337

1338 Note: this method is only applicable for files (`is_file` is True).

1339

1340 :return: The detected interpreter if present or None if no interpreter can be detected.

1341 """

1342 if not self.is_file:

1343 raise TypeError("Only files can have interpreters")

1344 try:

1345 with self.open(byte_io=True, buffering=4096) as fd:

1346 return extract_shebang_interpreter_from_file(fd)

1347 except (PureVirtualPathError, TestPathWithNonExistentFSPathError):

1348 return None

1349

1350 def metadata(

1351 self,

1352 metadata_type: Type[PMT],

1353 ) -> PathMetadataReference[PMT]:

1354 """Fetch the path metadata reference to access the underlying metadata

1355

1356 Calling this method returns a reference to an arbitrary piece of metadata associated

1357 with this path. Plugins can store any arbitrary data associated with a given path.

1358 Keep in mind that the metadata is stored in memory, so keep the size in moderation.

1359

1360 To store / update the metadata, the path must be in read-write mode. However,

1361 already stored metadata remains accessible even if the path becomes read-only.

1362

1363 Note this method is not applicable if the path is detached

1364

1365 :param metadata_type: Type of the metadata being stored.

1366 :return: A reference to the metadata.

1367 """

1368 raise NotImplementedError

1369

1370

1371class FlushableSubstvars(Substvars):

1372 __slots__ = ()

1373

1374 @contextlib.contextmanager

1375 def flush(self) -> Iterator[str]:

1376 """Temporarily write the substvars to a file and then re-read it again

1377

1378 >>> s = FlushableSubstvars()

1379 >>> 'Test:Var' in s

1380 False

1381 >>> with s.flush() as name, open(name, 'wt', encoding='utf-8') as fobj:

1382 ... _ = fobj.write('Test:Var=bar\\n') # "_ = " is to ignore the return value of write

1383 >>> 'Test:Var' in s

1384 True

1385

1386 Used as a context manager to define when the file is flushed and can be

1387 accessed via the file system. If the context terminates successfully, the

1388 file is read and its content replaces the current substvars.

1389

1390 This is mostly useful if the plugin needs to interface with a third-party

1391 tool that requires a file as interprocess communication (IPC) for sharing

1392 the substvars.

1393

1394 The file may be truncated or completed replaced (change inode) as long as

1395 the provided path points to a regular file when the context manager

1396 terminates successfully.

1397

1398 Note that any manipulation of the substvars via the `Substvars` API while

1399 the file is flushed will silently be discarded if the context manager completes

1400 successfully.

1401 """

1402 with tempfile.NamedTemporaryFile(mode="w+t", encoding="utf-8") as tmp:

1403 self.write_substvars(tmp)

1404 tmp.flush() # Temping to use close, but then we have to manually delete the file.

1405 yield tmp.name

1406 # Re-open; seek did not work when I last tried (if I did it work, feel free to

1407 # convert back to seek - as long as it works!)

1408 with open(tmp.name, "rt", encoding="utf-8") as fd:

1409 self.read_substvars(fd)

1410

1411 def save(self) -> None:

1412 # Promote the debputy extension over `save()` for the plugins.

1413 if self._substvars_path is None:

1414 raise TypeError(

1415 "Please use `flush()` extension to temporarily write the substvars to the file system"

1416 )

1417 super().save()

1418

1419

1420class ServiceRegistry(Generic[DSD]):

1421 __slots__ = ()

1422

1423 def register_service(

1424 self,

1425 path: VirtualPath,

1426 name: Union[str, List[str]],

1427 *,

1428 type_of_service: str = "service", # "timer", etc.

1429 service_scope: str = "system",

1430 enable_by_default: bool = True,

1431 start_by_default: bool = True,

1432 default_upgrade_rule: ServiceUpgradeRule = "restart",

1433 service_context: Optional[DSD] = None,

1434 ) -> None:

1435 """Register a service detected in the package

1436

1437 All the details will either be provided as-is or used as default when the plugin provided

1438 integration code is called.

1439

1440 Two services from different service managers are considered related when:

1441

1442 1) They are of the same type (`type_of_service`) and has the same scope (`service_scope`), AND

1443 2) Their plugin provided names has an overlap

1444

1445 Related services can be covered by the same service definition in the manifest.

1446

1447 :param path: The path defining this service.

1448 :param name: The name of the service. Multiple ones can be provided if the service has aliases.

1449 Note that when providing multiple names, `debputy` will use the first name in the list as the

1450 default name if it has to choose. Any alternative name provided can be used by the packager

1451 to identify this service.

1452 :param type_of_service: The type of service. By default, this is "service", but plugins can

1453 provide other types (such as "timer" for the systemd timer unit).

1454 :param service_scope: The scope for this service. By default, this is "system" meaning the

1455 service is a system-wide service. Service managers can define their own scopes such as

1456 "user" (which is used by systemd for "per-user" services).

1457 :param enable_by_default: Whether the service should be enabled by default, assuming the

1458 packager does not explicitly override this setting.

1459 :param start_by_default: Whether the service should be started by default on install, assuming

1460 the packager does not explicitly override this setting.

1461 :param default_upgrade_rule: The default value for how the service should be processed during

1462 upgrades. Options are:

1463 * `do-nothing`: The plugin should not interact with the running service (if any)

1464 (maintenance of the enabled start, start on install, etc. are still applicable)

1465 * `reload`: The plugin should attempt to reload the running service (if any).

1466 Note: In combination with `auto_start_in_install == False`, be careful to not

1467 start the service if not is not already running.

1468 * `restart`: The plugin should attempt to restart the running service (if any).

1469 Note: In combination with `auto_start_in_install == False`, be careful to not

1470 start the service if not is not already running.

1471 * `stop-then-start`: The plugin should stop the service during `prerm upgrade`

1472 and start it against in the `postinst` script.

1473

1474 :param service_context: Any custom data that the detector want to pass along to the

1475 integrator for this service.

1476 """

1477 raise NotImplementedError

1478

1479

1480@dataclasses.dataclass(slots=True, frozen=True)

1481class ParserAttributeDocumentation:

1482 attributes: FrozenSet[str]

1483 description: Optional[str]

1484

1485

1486def undocumented_attr(attr: str) -> ParserAttributeDocumentation:

1487 """Describe an attribute as undocumented

1488

1489 If you for some reason do not want to document a particular attribute, you can mark it as

1490 undocumented. This is required if you are only documenting a subset of the attributes,

1491 because `debputy` assumes any omission to be a mistake.

1492 """

1493 return ParserAttributeDocumentation(

1494 frozenset({attr}),

1495 None,

1496 )

1497

1498

1499@dataclasses.dataclass(slots=True, frozen=True)

1500class ParserDocumentation:

1501 title: Optional[str] = None

1502 description: Optional[str] = None

1503 attribute_doc: Optional[Sequence[ParserAttributeDocumentation]] = None

1504 alt_parser_description: Optional[str] = None

1505 documentation_reference_url: Optional[str] = None

1506

1507 def replace(self, **changes: Any) -> "ParserDocumentation":

1508 return dataclasses.replace(self, **changes)

1509

1510

1511@dataclasses.dataclass(slots=True, frozen=True)

1512class TypeMappingExample(Generic[S]):

1513 source_input: S

1514

1515

1516@dataclasses.dataclass(slots=True, frozen=True)

1517class TypeMappingDocumentation(Generic[S]):

1518 description: Optional[str] = None

1519 examples: Sequence[TypeMappingExample[S]] = tuple()

1520

1521

1522def type_mapping_example(source_input: S) -> TypeMappingExample[S]:

1523 return TypeMappingExample(source_input)

1524

1525

1526def type_mapping_reference_documentation(

1527 *,

1528 description: Optional[str] = None,

1529 examples: Union[TypeMappingExample[S], Iterable[TypeMappingExample[S]]] = tuple(),

1530) -> TypeMappingDocumentation[S]:

1531 e = (

1532 tuple([examples])

1533 if isinstance(examples, TypeMappingExample)

1534 else tuple(examples)

1535 )

1536 return TypeMappingDocumentation(

1537 description=description,

1538 examples=e,

1539 )

1540

1541

1542def documented_attr(

1543 attr: Union[str, Iterable[str]],

1544 description: str,

1545) -> ParserAttributeDocumentation:

1546 """Describe an attribute or a group of attributes

1547

1548 :param attr: A single attribute or a sequence of attributes. The attribute must be the

1549 attribute name as used in the source format version of the TypedDict.

1550

1551 If multiple attributes are provided, they will be documented together. This is often

1552 useful if these attributes are strongly related (such as different names for the same

1553 target attribute).

1554 :param description: The description the user should see for this attribute / these

1555 attributes. This parameter can be a Python format string with variables listed in

1556 the description of `reference_documentation`.

1557 :return: An opaque representation of the documentation,

1558 """

1559 attributes = [attr] if isinstance(attr, str) else attr

1560 return ParserAttributeDocumentation(

1561 frozenset(attributes),

1562 description,

1563 )

1564

1565

1566def reference_documentation(

1567 title: str = "Auto-generated reference documentation for {RULE_NAME}",

1568 description: Optional[str] = textwrap.dedent(

1569 """\

1570 This is an automatically generated reference documentation for {RULE_NAME}. It is generated

1571 from input provided by {PLUGIN_NAME} via the debputy API.

1572

1573 (If you are the provider of the {PLUGIN_NAME} plugin, you can replace this text with

1574 your own documentation by providing the `inline_reference_documentation` when registering

1575 the manifest rule.)

1576 """

1577 ),

1578 attributes: Optional[Sequence[ParserAttributeDocumentation]] = None,

1579 non_mapping_description: Optional[str] = None,

1580 reference_documentation_url: Optional[str] = None,

1581) -> ParserDocumentation:

1582 """Provide inline reference documentation for the manifest snippet

1583

1584 For parameters that mention that they are a Python format, the following format variables

1585 are available:

1586

1587 * RULE_NAME: Name of the rule. If manifest snippet has aliases, this will be the name of

1588 the alias provided by the user.

1589 * MANIFEST_FORMAT_DOC: Path OR URL to the "MANIFEST-FORMAT" reference documentation from

1590 `debputy`. By using the MANIFEST_FORMAT_DOC variable, you ensure that you point to the

1591 file that matches the version of `debputy` itself.

1592 * PLUGIN_NAME: Name of the plugin providing this rule.

1593

1594 :param title: The text you want the user to see as for your rule. A placeholder is provided by default.

1595 This parameter can be a Python format string with the above listed variables.

1596 :param description: The text you want the user to see as a description for the rule. An auto-generated

1597 placeholder is provided by default saying that no human written documentation was provided.

1598 This parameter can be a Python format string with the above listed variables.

1599 :param attributes: A sequence of attribute-related documentation. Each element of the sequence should

1600 be the result of `documented_attr` or `undocumented_attr`. The sequence must cover all source

1601 attributes exactly once.

1602 :param non_mapping_description: The text you want the user to see as the description for your rule when

1603 `debputy` describes its non-mapping format. Must not be provided for rules that do not have an

1604 (optional) non-mapping format as source format. This parameter can be a Python format string with

1605 the above listed variables.

1606 :param reference_documentation_url: A URL to the reference documentation.

1607 :return: An opaque representation of the documentation,

1608 """

1609 return ParserDocumentation(

1610 title,

1611 description,

1612 attributes,

1613 non_mapping_description,

1614 reference_documentation_url,

1615 )

1616

1617

1618class ServiceDefinition(Generic[DSD]):

1619 __slots__ = ()

1620

1621 @property

1622 def name(self) -> str:

1623 """Name of the service registered by the plugin

1624

1625 This is always a plugin provided name for this service (that is, `x.name in x.names`

1626 will always be `True`). Where possible, this will be the same as the one that the

1627 packager provided when they provided any configuration related to this service.

1628 When not possible, this will be the first name provided by the plugin (`x.names[0]`).

1629

1630 If all the aliases are equal, then using this attribute will provide traceability

1631 between the manifest and the generated maintscript snippets. When the exact name

1632 used is important, the plugin should ignore this attribute and pick the name that

1633 is needed.

1634 """

1635 raise NotImplementedError

1636

1637 @property

1638 def names(self) -> Sequence[str]:

1639 """All *plugin provided* names and aliases of the service

1640

1641 This is the name/sequence of names that the plugin provided when it registered

1642 the service earlier.

1643 """

1644 raise NotImplementedError

1645

1646 @property

1647 def path(self) -> VirtualPath:

1648 """The registered path for this service

1649

1650 :return: The path that was associated with this service when it was registered

1651 earlier.

1652 """

1653 raise NotImplementedError

1654

1655 @property

1656 def type_of_service(self) -> str:

1657 """Type of the service such as "service" (daemon), "timer", etc.

1658

1659 :return: The type of service scope. It is the same value as the one as the plugin provided

1660 when registering the service (if not explicitly provided, it defaults to "service").

1661 """

1662 raise NotImplementedError

1663

1664 @property

1665 def service_scope(self) -> str:

1666 """Service scope such as "system" or "user"

1667

1668 :return: The service scope. It is the same value as the one as the plugin provided

1669 when registering the service (if not explicitly provided, it defaults to "system")

1670 """

1671 raise NotImplementedError

1672

1673 @property

1674 def auto_enable_on_install(self) -> bool:

1675 """Whether the service should be auto-enabled on install

1676

1677 :return: True if the service should be enabled automatically, false if not.

1678 """

1679 raise NotImplementedError

1680

1681 @property

1682 def auto_start_on_install(self) -> bool:

1683 """Whether the service should be auto-started on install

1684

1685 :return: True if the service should be started automatically, false if not.

1686 """

1687 raise NotImplementedError

1688

1689 @property

1690 def on_upgrade(self) -> ServiceUpgradeRule:

1691 """How to handle the service during an upgrade

1692

1693 Options are:

1694 * `do-nothing`: The plugin should not interact with the running service (if any)

1695 (maintenance of the enabled start, start on install, etc. are still applicable)

1696 * `reload`: The plugin should attempt to reload the running service (if any).

1697 Note: In combination with `auto_start_in_install == False`, be careful to not

1698 start the service if not is not already running.

1699 * `restart`: The plugin should attempt to restart the running service (if any).

1700 Note: In combination with `auto_start_in_install == False`, be careful to not

1701 start the service if not is not already running.

1702 * `stop-then-start`: The plugin should stop the service during `prerm upgrade`

1703 and start it against in the `postinst` script.

1704

1705 Note: In all cases, the plugin should still consider what to do in

1706 `prerm remove`, which is the last point in time where the plugin can rely on the

1707 service definitions in the file systems to stop the services when the package is

1708 being uninstalled.

1709

1710 :return: The service restart rule

1711 """

1712 raise NotImplementedError

1713

1714 @property

1715 def definition_source(self) -> str:

1716 """Describes where this definition came from

1717

1718 If the definition is provided by the packager, then this will reference the part

1719 of the manifest that made this definition. Otherwise, this will be a reference

1720 to the plugin providing this definition.

1721

1722 :return: The source of this definition

1723 """

1724 raise NotImplementedError

1725

1726 @property

1727 def is_plugin_provided_definition(self) -> bool:

1728 """Whether the definition source points to the plugin or a package provided definition

1729

1730 :return: True if definition is 100% from the plugin. False if the definition is partially

1731 or fully from another source (usually, the packager via the manifest).

1732 """

1733 raise NotImplementedError

1734

1735 @property

1736 def service_context(self) -> Optional[DSD]:

1737 """Custom service context (if any) provided by the detector code of the plugin

1738

1739 :return: If the detection code provided a custom data when registering the

1740 service, this attribute will reference that data. If nothing was provided,

1741 then this attribute will be None.

1742 """

1743 raise NotImplementedError

- Coverage for src/debputy/plugin/api/spec.py: - 87% -

- 282 statements - - - - -