1- # Copyright (c) TurnKey GNU/Linux - http ://www.turnkeylinux.org
1+ # Copyright (c) TurnKey GNU/Linux - https ://www.turnkeylinux.org
22#
33# This file is part of Fab
44#
77# Free Software Foundation; either version 3 of the License, or (at your
88# option) any later version.
99
10- import io
1110import os
1211from os .path import join , exists , basename
1312import shutil
14- from typing import (
15- Iterable , Optional , Dict , Tuple , List , TextIO , IO , AnyStr , cast
16- )
13+ from typing import Iterable , TextIO , cast
1714import logging
18- logger = logging .getLogger ('fab.installer' )
1915
2016import hashlib
2117from debian import debfile
2218
2319from chroot import Chroot
2420from fablib import common
2521
22+ logger = logging .getLogger ("fab.installer" )
23+
2624
2725class Error (Exception ):
2826 pass
2927
3028
3129class RevertibleFile :
3230 """File that automatically reverts to previous state on destruction
33- or if the revert method is invoked"""
31+ or if the revert method is invoked"""
3432
3533 @staticmethod
3634 def _get_orig_path (path : str ) -> str :
@@ -43,16 +41,16 @@ def _get_orig_path(path: str) -> str:
4341 i += 1
4442
4543 def __init__ (self , path : str ):
46- self .orig_path : Optional [ str ] = None
44+ self .orig_path : str | None = None
4745 if exists (path ):
4846 self .orig_path = self ._get_orig_path (path )
4947 shutil .move (path , self .orig_path )
50- self .path : Optional [ str ] = path
48+ self .path : str | None = path
5149
52- self ._inner = open (path , 'w' )
50+ self ._inner = open (path , "w" )
5351
5452 def revert (self ) -> None :
55- ''' revert file to original state '''
53+ """ revert file to original state"""
5654 if self .orig_path is not None :
5755 assert self .path is not None
5856 shutil .move (self .orig_path , self .path )
@@ -74,7 +72,8 @@ def __del__(self) -> None:
7472
7573
7674class RevertibleScript (RevertibleFile ):
77- ''' RevertibleFile that ensures file is executable '''
75+ """RevertibleFile that ensures file is executable"""
76+
7877 def __init__ (self , path : str , lines : Iterable [str ]):
7978 super ().__init__ (path )
8079 self .write ("\n " .join (lines ))
@@ -84,10 +83,7 @@ def __init__(self, path: str, lines: Iterable[str]):
8483
8584
8685class Installer :
87- def __init__ (
88- self , chroot_path : str ,
89- environ : dict [str , str ] = None
90- ):
86+ def __init__ (self , chroot_path : str , environ : dict [str , str ] | None = None ):
9187 if environ is None :
9288 environ = {}
9389 env = {"DEBIAN_FRONTEND" : "noninteractive" , "DEBIAN_PRIORITY" : "critical" }
@@ -98,7 +94,7 @@ def __init__(
9894 @staticmethod
9995 def _get_packages_priority (packages : list [str ]) -> tuple [list [str ], list [str ]]:
10096 """high priority packages must be installed before regular packages
101- APT should handle this, but in some circumstances it chokes...
97+ APT should handle this, but in some circumstances it chokes...
10298 """
10399 HIGH_PRIORITY = "linux-image"
104100
@@ -114,22 +110,25 @@ def _get_packages_priority(packages: list[str]) -> tuple[list[str], list[str]]:
114110 return high , regular
115111
116112 def _install (
117- self , packages : list [str ],
118- ignore_errors : list [str ] = None ,
119- extra_apt_args : list [str ] = None ) -> None :
120-
113+ self ,
114+ packages : list [str ],
115+ ignore_errors : list [str ] | None = None ,
116+ extra_apt_args : list [str ] | None = None ,
117+ ) -> None :
121118 if ignore_errors is None :
122119 ignore_errors = []
123120 if extra_apt_args is None :
124121 extra_apt_args = []
125122 high , regular = self ._get_packages_priority (packages )
126123
127124 lines = ["#!/bin/sh" , "echo" , 'echo "Warning: Fake invoke-rc.d called"' ]
125+ # TODO fake_invoke_rcd not accessed
128126 fake_invoke_rcd = RevertibleScript (
129127 join (self .chroot .path , "usr/sbin/invoke-rc.d" ), lines
130128 )
131129
132130 lines = ["#!/bin/sh" , "echo" , 'echo "Warning: Fake start-stop-daemon called"' ]
131+ # TODO fake_start_stop not accessed
133132 fake_start_stop = RevertibleScript (
134133 join (self .chroot .path , "sbin/start-stop-daemon" ), lines
135134 )
@@ -139,15 +138,21 @@ def _install(
139138 "#!/bin/sh" ,
140139 "echo" ,
141140 'echo "Warning: Deferring update-initramfs $@"' ,
142- f'echo "update-initramfs $@" >> /{ defer_log } '
141+ f'echo "update-initramfs $@" >> /{ defer_log } ' ,
143142 ]
144143
145144 for packages in (high , regular ):
146145 if packages :
147- args = ["-o" , "Debug::pkgProblemResolver=true" , "install" , "--assume-yes" ]
146+ args = [
147+ "-o" ,
148+ "Debug::pkgProblemResolver=true" ,
149+ "install" ,
150+ "--assume-yes" ,
151+ ]
148152 args .extend (extra_apt_args )
149153 apt_return_code = self .chroot .system (
150- f"apt-get { ' ' .join ((args + packages ))} " )
154+ f"apt-get { ' ' .join ((args + packages ))} "
155+ )
151156 if apt_return_code != 0 :
152157
153158 def get_last_log (path : str ) -> list [str ]:
@@ -184,8 +189,7 @@ def get_errors(log: list[str], error_str: str) -> list[str]:
184189 if apt_return_code == 100 :
185190 # always seems to return 100 when hitting
186191 # 'E: Unable to locate package ...'
187- raise Error (
188- 'Errors encountered installing packages' )
192+ raise Error ("Errors encountered installing packages" )
189193 else :
190194 continue
191195
@@ -195,13 +199,15 @@ def get_errors(log: list[str], error_str: str) -> list[str]:
195199 errors = set (errors ) - set (ignore_errors )
196200
197201 if ignored_errors :
198- print (f"Warning: ignoring package installation errors"
199- f" ({ ' ' .join (ignored_errors )} )" )
202+ print (
203+ f"Warning: ignoring package installation errors"
204+ f" ({ ' ' .join (ignored_errors )} )"
205+ )
200206
201207 if errors :
202208 for error in errors :
203209 common .error (error )
204- raise Error (' package installation errors' )
210+ raise Error (" package installation errors" )
205211
206212 defer_log = join (self .chroot .path , defer_log )
207213 if exists (defer_log ):
@@ -216,22 +222,26 @@ def get_errors(log: list[str], error_str: str) -> list[str]:
216222 if self .chroot .system ("update-initramfs -u" ) != 0 :
217223 self .chroot .system ("live-update-initramfs -u" )
218224 else :
219- if self .chroot .system (
220- f"update-initramfs -c -k { kversion } " ) != 0 :
225+ if self .chroot .system (f"update-initramfs -c -k { kversion } " ) != 0 :
221226 self .chroot .system (f"live-update-initramfs -c -k { kversion } " )
222227
223228 os .remove (defer_log )
224229
225230 def install (
226- self , packages : list [str ],
227- ignore_errors : list [str ] = None ) -> None :
231+ self , packages : list [str ], ignore_errors : list [str ] | None = None
232+ ) -> None :
233+ # TODO implement or remove
228234 raise NotImplementedError ()
229235
230236
231237class PoolInstaller (Installer ):
232238 def __init__ (
233- self , chroot_path : str , pool_path : str ,
234- arch : str , environ : dict [str , str ] = None ):
239+ self ,
240+ chroot_path : str ,
241+ pool_path : str ,
242+ arch : str ,
243+ environ : dict [str , str ] | None = None ,
244+ ):
235245 super (PoolInstaller , self ).__init__ (chroot_path , environ )
236246
237247 from pool_lib import Pool
@@ -247,17 +257,18 @@ def filesize(path: str) -> str:
247257 return str (os .stat (path ).st_size )
248258
249259 def md5sum (path : str ) -> str :
250- with open (path , 'rb' ) as fob :
260+ with open (path , "rb" ) as fob :
251261 return str (hashlib .md5 (fob .read ()).hexdigest ())
252262
253263 def sha256sum (path : str ) -> str :
254- with open (path , 'rb' ) as fob :
264+ with open (path , "rb" ) as fob :
255265 return str (hashlib .sha256 (fob .read ()).hexdigest ())
256266
257267 index = []
258268 for package in os .listdir (packagedir ):
259269 path = os .path .join (packagedir , package )
260- # dl_path would best be calculated; but we don't have access to chroot_path here...
270+ # dl_path would best be calculated; but we don't
271+ # have access to chroot_path here...
261272 dl_path = os .path .join ("var/cache/apt/archives" , package )
262273 if path .endswith (".deb" ):
263274 control = debfile .DebFile (path ).debcontrol ()
@@ -273,8 +284,7 @@ def sha256sum(path: str) -> str:
273284 return index
274285
275286 def install (
276- self , packages : list [str ],
277- ignore_errors : list [str ] = None
287+ self , packages : list [str ], ignore_errors : list [str ] | None = None
278288 ) -> None :
279289 """install packages into chroot via pool"""
280290
@@ -308,29 +318,22 @@ def install(
308318
309319class LiveInstaller (Installer ):
310320 def __init__ (
311- self , chroot_path : str ,
312- apt_proxy : str = None ,
313- environ : dict [str , str ] = None ):
321+ self ,
322+ chroot_path : str ,
323+ apt_proxy : str | None = None ,
324+ environ : dict [str , str ] | None = None ,
325+ ):
314326 super (LiveInstaller , self ).__init__ (chroot_path , environ )
315327
316328 self .apt_proxy = apt_proxy
317329
318330 def install (
319- self , packages : list [str ],
320- ignore_errors : list [ str ] = None ) -> None :
331+ self , packages : list [str ], ignore_errors : list [ str ] | None = None
332+ ) -> None :
321333 """install packages into chroot via live apt"""
322334 if ignore_errors is None :
323335 ignore_errors = []
324336
325- # For v17.x I've moved the apt setting to common. I think that is the
326- # right place for it, but haven't 100% committed yet. For now I'm
327- # leaving this here commented...
328- #if self.apt_proxy:
329- # print("setting apt proxy settings...")
330- # conf_path = join(self.chroot.path, "etc/apt/apt.conf.d/01proxy")
331- # with open(conf_path, "w") as fob:
332- # fob.write('Acquire::http::Proxy "%s";\n' % self.apt_proxy)
333-
334337 print ("updating package lists..." )
335338 self .chroot .system ("apt-get update" )
336339
0 commit comments