Coverage for src/debputy/plugins/debputy/package_processors.py: 54%
175 statements
« prev ^ index » next coverage.py v7.8.2, created at 2026-07-22 10:58 +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 ".woff",
192 ".woff2",
193 ".svg",
194 ".svgz",
195 ".js",
196 ".devhelp2",
197 ".map", # Technically, dh_compress has this one case-sensitive
198 ".rda",
199 ".rdata",
200 ".rds",
201 }
202 ignore_special_cases = ("-gz", "-z", "_z")
204 def _filtered_walk(path: VirtualPath) -> Iterator[VirtualPath]:
205 for path, children in typing.cast(InMemoryVirtualPathBase, path).walk():
206 if path.name in ignore_dir_basenames: 206 ↛ 207line 206 didn't jump to line 207 because the condition on line 206 was never true
207 children.clear()
208 continue
209 if path.is_dir and path.name == "examples": 209 ↛ 211line 209 didn't jump to line 211 because the condition on line 209 was never true
210 # Ignore anything beneath /usr/share/doc/*/examples
211 parent = path.parent_dir
212 grand_parent = parent.parent_dir if parent else None
213 if grand_parent and grand_parent.absolute == "/usr/share/doc":
214 children.clear()
215 continue
216 name = path.name
217 if (
218 path.is_symlink
219 or not path.is_file
220 or name in ignore_basenames
221 or not path.has_fs_path
222 ):
223 continue
225 name_lc = name.lower()
226 _, ext = os.path.splitext(name_lc)
228 if ext in ignore_extensions or name_lc.endswith(ignore_special_cases): 228 ↛ 229line 228 didn't jump to line 229 because the condition on line 228 was never true
229 continue
230 yield path
232 return _filtered_walk
235def _find_compressable_paths(fs_root: VirtualPath) -> Iterator[VirtualPath]:
236 path_filter = _filter_compress_paths()
238 for p, compress_size_threshold in (
239 ("./usr/share/info", 0),
240 ("./usr/share/man", 0),
241 ("./usr/share/doc", 4096),
242 ):
243 path = fs_root.lookup(p)
244 if path is None:
245 continue
246 paths = path_filter(path)
247 if compress_size_threshold: 247 ↛ 250line 247 didn't jump to line 250 because the condition on line 247 was never true
248 # The special-case for changelog and NEWS is from dh_compress. Generally these files
249 # have always been compressed regardless of their size.
250 paths = (
251 p
252 for p in paths
253 if p.size > compress_size_threshold
254 # Case-insensitivity is a known delta from `dh_compress` at the time of writing.
255 or p.name.lower().startswith(("changelog", "news"))
256 )
257 yield from paths
258 x11_path = fs_root.lookup("./usr/share/fonts/X11")
259 if x11_path: 259 ↛ 260line 259 didn't jump to line 260 because the condition on line 259 was never true
260 yield from (
261 p for p in x11_path.all_paths() if p.is_file and p.name.endswith(".pcf")
262 )
265def apply_compression(fs_root: VirtualPath, _unused1: Any, _unused2: Any) -> None:
266 # TODO: Support hardlinks
267 compressed_files: dict[str, str] = {}
268 for path in _find_compressable_paths(fs_root):
269 parent_dir = assume_not_none(path.parent_dir)
270 with (
271 parent_dir.add_file(f"{path.name}.gz", mtime=path.mtime) as new_file,
272 open(new_file.fs_path, "wb") as fd,
273 ):
274 try:
275 subprocess.check_call(["gzip", "-9nc", path.fs_path], stdout=fd)
276 except subprocess.CalledProcessError:
277 full_command = f"gzip -9nc {escape_shell(path.fs_path)} > {escape_shell(new_file.fs_path)}"
278 _error(
279 f"The compression of {path.path} failed. Please review the error message from gzip to"
280 f" understand what went wrong. Full command was: {full_command}"
281 )
282 compressed_files[path.path] = new_file.path
283 del parent_dir[path.name]
285 all_remaining_symlinks = {p.path: p for p in fs_root.all_paths() if p.is_symlink}
286 changed = True
287 while changed:
288 changed = False
289 remaining: list[VirtualPath] = list(all_remaining_symlinks.values())
290 for symlink in remaining:
291 target = symlink.readlink()
292 dir_target, basename_target = os.path.split(target)
293 new_basename_target = f"{basename_target}.gz"
294 symlink_parent_dir = assume_not_none(symlink.parent_dir)
295 dir_path = (
296 symlink_parent_dir.lookup(dir_target)
297 if dir_target
298 else symlink_parent_dir
299 )
300 if ( 300 ↛ 305line 300 didn't jump to line 305 because the condition on line 300 was never true
301 not dir_path
302 or basename_target in dir_path
303 or new_basename_target not in dir_path
304 ):
305 continue
306 del all_remaining_symlinks[symlink.path]
307 changed = True
309 new_link_name = (
310 f"{symlink.name}.gz"
311 if not symlink.name.endswith(".gz")
312 else symlink.name
313 )
314 symlink_parent_dir.add_symlink(
315 new_link_name, os.path.join(dir_target, new_basename_target)
316 )
317 symlink.unlink()
320def _la_files(fs_root: VirtualPath) -> Iterator[VirtualPath]:
321 lib_dir = fs_root.lookup("/usr/lib")
322 if not lib_dir:
323 return
324 # Original code only iterators directly in /usr/lib. To be a faithful conversion, we do the same
325 # here.
326 # Eagerly resolve the list as the replacement can trigger a runtime error otherwise
327 paths = list(lib_dir.iterdir())
328 yield from (p for p in paths if p.is_file and p.name.endswith(".la"))
331# Conceptually, the same feature that dh_gnome provides.
332# The clean_la_files function based on the dh_gnome version written by Luca Falavigna in 2010,
333# who in turn references a Makefile version of the feature.
334# https://salsa.debian.org/gnome-team/gnome-pkg-tools/-/commit/2868e1e41ea45443b0fb340bf4c71c4de87d4a5b
335def clean_la_files(
336 fs_root: VirtualPath,
337 _unused1: Any,
338 _unused2: Any,
339) -> None:
340 for path in _la_files(fs_root):
341 buffer = []
342 with path.open(byte_io=True) as fd:
343 replace_file = False
344 for line in fd:
345 if line.startswith(b"dependency_libs"):
346 replacement = _LA_DEP_LIB_RE.sub(b"''", line)
347 if replacement != line:
348 replace_file = True
349 line = replacement
350 buffer.append(line)
352 if not replace_file:
353 continue
354 _info(f"Clearing the dependency_libs line in {path.path}")
355 with path.replace_fs_path_content() as fs_path, open(fs_path, "wb") as wfd:
356 wfd.writelines(buffer)