Coverage for src/debputy/plugins/debputy/package_processors.py: 54%
175 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-10-04 09:01 +0000
1import contextlib
2import functools
3import gzip
4import os
5import re
6import subprocess
7import typing
8from contextlib import ExitStack
9from typing import IO, Any
10from collections.abc import Iterator, Callable
12from debputy.filesystem_scan import InMemoryVirtualPathBase
13from debputy.plugin.api import VirtualPath
14from debputy.util import (
15 _error,
16 xargs,
17 escape_shell,
18 _info,
19 assume_not_none,
20 _debug_log,
21)
24@contextlib.contextmanager
25def _open_maybe_gzip(path: VirtualPath) -> Iterator[IO[bytes] | gzip.GzipFile]:
26 if path.name.endswith(".gz"):
27 with gzip.GzipFile(path.fs_path, "rb") as fd:
28 yield fd
29 else:
30 with path.open(byte_io=True) as fd:
31 yield fd
34_SO_LINK_RE = re.compile(rb"[.]so\s+(.*)\s*")
35_LA_DEP_LIB_RE = re.compile(rb"'.+'")
38def _detect_so_link(path: VirtualPath) -> str | None:
39 so_link_re = _SO_LINK_RE
40 with _open_maybe_gzip(path) as fd:
41 for line in fd:
42 m = so_link_re.search(line)
43 if m:
44 return m.group(1).decode("utf-8")
45 return None
48def _replace_with_symlink(path: VirtualPath, so_link_target: str) -> None:
49 adjusted_target = so_link_target
50 parent_dir = path.parent_dir
51 assert parent_dir is not None # For the type checking
52 if parent_dir.name == os.path.dirname(adjusted_target):
53 # Avoid man8/../man8/foo links
54 adjusted_target = os.path.basename(adjusted_target)
55 elif "/" in so_link_target:
56 # symlinks and so links have a different base directory when the link has a "/".
57 # Adjust with an extra "../" to align the result
58 adjusted_target = "../" + adjusted_target
60 path.unlink()
61 parent_dir.add_symlink(path.name, adjusted_target)
64@functools.lru_cache(1)
65def _has_man_recode() -> bool:
66 # Ideally, we would just use shutil.which or something like that.
67 # Unfortunately, in debhelper, we experienced problems with which
68 # returning "yes" for a man tool that actually could not be run
69 # on salsa CI.
70 #
71 # Therefore, we adopt the logic of dh_installman to run the tool
72 # with --help to confirm it is not broken, because no one could
73 # figure out what happened in the salsa CI and my life is still
74 # too short to figure it out.
75 try:
76 subprocess.check_call(
77 ["man-recode", "--help"],
78 stdin=subprocess.DEVNULL,
79 stdout=subprocess.DEVNULL,
80 stderr=subprocess.DEVNULL,
81 restore_signals=True,
82 )
83 except subprocess.CalledProcessError:
84 return False
85 return True
88def process_manpages(fs_root: VirtualPath, _unused1: Any, _unused2: Any) -> None:
89 man_dir = fs_root.lookup("./usr/share/man")
90 if not man_dir:
91 return
93 re_encode = []
94 for path in (p for p in man_dir.all_paths() if p.is_file and p.has_fs_path):
95 size = path.size
96 if size == 0:
97 continue
98 so_link_target = None
99 if size <= 1024:
100 # debhelper has a 1024 byte guard on the basis that ".so file tend to be small".
101 # That guard worked well for debhelper, so lets keep it for now on that basis alone.
102 so_link_target = _detect_so_link(path)
103 if so_link_target:
104 _replace_with_symlink(path, so_link_target)
105 else:
106 re_encode.append(path)
108 if not re_encode or not _has_man_recode():
109 return
111 with ExitStack() as manager:
112 manpages = [
113 manager.enter_context(p.replace_fs_path_content()) for p in re_encode
114 ]
115 static_cmd = ["man-recode", "--to-code", "UTF-8", "--suffix", ".encoded"]
116 for cmd in xargs(static_cmd, manpages):
117 _info(f"Ensuring manpages have utf-8 encoding via: {escape_shell(*cmd)}")
118 try:
119 subprocess.check_call(
120 cmd,
121 stdin=subprocess.DEVNULL,
122 restore_signals=True,
123 )
124 except subprocess.CalledProcessError:
125 _error(
126 "The man-recode process failed. Please review the output of `man-recode` to understand"
127 " what went wrong."
128 )
129 for manpage in manpages:
130 dest_name = manpage
131 if dest_name.endswith(".gz"):
132 encoded_name = dest_name[:-3] + ".encoded"
133 with open(dest_name, "wb") as out:
134 _debug_log(
135 f"Recompressing {dest_name} via gzip -9nc {escape_shell(encoded_name)}"
136 )
137 try:
138 subprocess.check_call(
139 [
140 "gzip",
141 "-9nc",
142 encoded_name,
143 ],
144 stdin=subprocess.DEVNULL,
145 stdout=out,
146 )
147 except subprocess.CalledProcessError:
148 _error(
149 f"The command {escape_shell('gzip', '-nc', f'{encoded_name}')} > {dest_name} failed!"
150 )
151 else:
152 os.rename(f"{dest_name}.encoded", manpage)
155def _filter_compress_paths() -> Callable[[VirtualPath], Iterator[VirtualPath]]:
156 ignore_dir_basenames = {
157 "_sources",
158 }
159 ignore_basenames = {
160 ".htaccess",
161 "index.sgml",
162 "objects.inv",
163 "search_index.json",
164 "copyright",
165 }
166 ignore_extensions = {
167 ".htm",
168 ".html",
169 ".xhtml",
170 ".gif",
171 ".png",
172 ".jpg",
173 ".jpeg",
174 ".gz",
175 ".taz",
176 ".tgz",
177 ".z",
178 ".bz2",
179 ".epub",
180 ".jar",
181 ".zip",
182 ".odg",
183 ".odp",
184 ".odt",
185 ".css",
186 ".xz",
187 ".lz",
188 ".lzma",
189 ".haddock",
190 ".hs",
191 ".pdf",
192 ".woff",
193 ".woff2",
194 ".svg",
195 ".svgz",
196 ".js",
197 ".devhelp2",
198 ".map", # Technically, dh_compress has this one case-sensitive
199 ".rda",
200 ".rdata",
201 ".rds",
202 }
203 ignore_special_cases = ("-gz", "-z", "_z")
205 def _filtered_walk(path: VirtualPath) -> Iterator[VirtualPath]:
206 for path, children in typing.cast(InMemoryVirtualPathBase, path).walk():
207 if path.name in ignore_dir_basenames: 207 ↛ 208line 207 didn't jump to line 208 because the condition on line 207 was never true
208 children.clear()
209 continue
210 if path.is_dir and path.name == "examples": 210 ↛ 212line 210 didn't jump to line 212 because the condition on line 210 was never true
211 # Ignore anything beneath /usr/share/doc/*/examples
212 parent = path.parent_dir
213 grand_parent = parent.parent_dir if parent else None
214 if grand_parent and grand_parent.absolute == "/usr/share/doc":
215 children.clear()
216 continue
217 name = path.name
218 if (
219 path.is_symlink
220 or not path.is_file
221 or name in ignore_basenames
222 or not path.has_fs_path
223 ):
224 continue
226 name_lc = name.lower()
227 _, ext = os.path.splitext(name_lc)
229 if ext in ignore_extensions or name_lc.endswith(ignore_special_cases): 229 ↛ 230line 229 didn't jump to line 230 because the condition on line 229 was never true
230 continue
231 yield path
233 return _filtered_walk
236def _find_compressable_paths(fs_root: VirtualPath) -> Iterator[VirtualPath]:
237 path_filter = _filter_compress_paths()
239 for p, compress_size_threshold in (
240 ("./usr/share/info", 0),
241 ("./usr/share/man", 0),
242 ("./usr/share/doc", 4096),
243 ):
244 path = fs_root.lookup(p)
245 if path is None:
246 continue
247 paths = path_filter(path)
248 if compress_size_threshold: 248 ↛ 251line 248 didn't jump to line 251 because the condition on line 248 was never true
249 # The special-case for changelog and NEWS is from dh_compress. Generally these files
250 # have always been compressed regardless of their size.
251 paths = (
252 p
253 for p in paths
254 if p.size > compress_size_threshold
255 # Case-insensitivity is a known delta from `dh_compress` at the time of writing.
256 or p.name.lower().startswith(("changelog", "news"))
257 )
258 yield from paths
259 x11_path = fs_root.lookup("./usr/share/fonts/X11")
260 if x11_path: 260 ↛ 261line 260 didn't jump to line 261 because the condition on line 260 was never true
261 yield from (
262 p for p in x11_path.all_paths() if p.is_file and p.name.endswith(".pcf")
263 )
266def apply_compression(fs_root: VirtualPath, _unused1: Any, _unused2: Any) -> None:
267 # TODO: Support hardlinks
268 compressed_files: dict[str, str] = {}
269 for path in _find_compressable_paths(fs_root):
270 parent_dir = assume_not_none(path.parent_dir)
271 with (
272 parent_dir.add_file(f"{path.name}.gz", mtime=path.mtime) as new_file,
273 open(new_file.fs_path, "wb") as fd,
274 ):
275 try:
276 subprocess.check_call(["gzip", "-9nc", path.fs_path], stdout=fd)
277 except subprocess.CalledProcessError:
278 full_command = f"gzip -9nc {escape_shell(path.fs_path)} > {escape_shell(new_file.fs_path)}"
279 _error(
280 f"The compression of {path.path} failed. Please review the error message from gzip to"
281 f" understand what went wrong. Full command was: {full_command}"
282 )
283 compressed_files[path.path] = new_file.path
284 del parent_dir[path.name]
286 all_remaining_symlinks = {p.path: p for p in fs_root.all_paths() if p.is_symlink}
287 changed = True
288 while changed:
289 changed = False
290 remaining: list[VirtualPath] = list(all_remaining_symlinks.values())
291 for symlink in remaining:
292 target = symlink.readlink()
293 dir_target, basename_target = os.path.split(target)
294 new_basename_target = f"{basename_target}.gz"
295 symlink_parent_dir = assume_not_none(symlink.parent_dir)
296 dir_path = (
297 symlink_parent_dir.lookup(dir_target)
298 if dir_target
299 else symlink_parent_dir
300 )
301 if ( 301 ↛ 306line 301 didn't jump to line 306 because the condition on line 301 was never true
302 not dir_path
303 or basename_target in dir_path
304 or new_basename_target not in dir_path
305 ):
306 continue
307 del all_remaining_symlinks[symlink.path]
308 changed = True
310 new_link_name = (
311 f"{symlink.name}.gz"
312 if not symlink.name.endswith(".gz")
313 else symlink.name
314 )
315 symlink_parent_dir.add_symlink(
316 new_link_name, os.path.join(dir_target, new_basename_target)
317 )
318 symlink.unlink()
321def _la_files(fs_root: VirtualPath) -> Iterator[VirtualPath]:
322 lib_dir = fs_root.lookup("/usr/lib")
323 if not lib_dir:
324 return
325 # Original code only iterators directly in /usr/lib. To be a faithful conversion, we do the same
326 # here.
327 # Eagerly resolve the list as the replacement can trigger a runtime error otherwise
328 paths = list(lib_dir.iterdir())
329 yield from (p for p in paths if p.is_file and p.name.endswith(".la"))
332# Conceptually, the same feature that dh_gnome provides.
333# The clean_la_files function based on the dh_gnome version written by Luca Falavigna in 2010,
334# who in turn references a Makefile version of the feature.
335# https://salsa.debian.org/gnome-team/gnome-pkg-tools/-/commit/2868e1e41ea45443b0fb340bf4c71c4de87d4a5b
336def clean_la_files(
337 fs_root: VirtualPath,
338 _unused1: Any,
339 _unused2: Any,
340) -> None:
341 for path in _la_files(fs_root):
342 buffer = []
343 with path.open(byte_io=True) as fd:
344 replace_file = False
345 for line in fd:
346 if line.startswith(b"dependency_libs"):
347 replacement = _LA_DEP_LIB_RE.sub(b"''", line)
348 if replacement != line:
349 replace_file = True
350 line = replacement
351 buffer.append(line)
353 if not replace_file:
354 continue
355 _info(f"Clearing the dependency_libs line in {path.path}")
356 with path.replace_fs_path_content() as fs_path, open(fs_path, "wb") as wfd:
357 wfd.writelines(buffer)