Merge pull request #427 from advaithm/mypy
some type hint fixes and a bad catch fix
This commit is contained in:
commit
df6c4e77f7
|
|
@ -1,3 +1,4 @@
|
||||||
|
from typing import Optional
|
||||||
import glob, re, os, json, time, hashlib
|
import glob, re, os, json, time, hashlib
|
||||||
import pathlib, traceback, logging
|
import pathlib, traceback, logging
|
||||||
from collections import OrderedDict
|
from collections import OrderedDict
|
||||||
|
|
@ -205,7 +206,7 @@ class Partition():
|
||||||
return f'Partition(path={self.path}, size={self.size}, fs={self.filesystem}{mount_repr})'
|
return f'Partition(path={self.path}, size={self.size}, fs={self.filesystem}{mount_repr})'
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def uuid(self) -> str:
|
def uuid(self) -> Optional[str]:
|
||||||
"""
|
"""
|
||||||
Returns the PARTUUID as returned by lsblk.
|
Returns the PARTUUID as returned by lsblk.
|
||||||
This is more reliable than relying on /dev/disk/by-partuuid as
|
This is more reliable than relying on /dev/disk/by-partuuid as
|
||||||
|
|
@ -214,7 +215,7 @@ class Partition():
|
||||||
lsblk = b''.join(sys_command(f'lsblk -J -o+PARTUUID {self.path}'))
|
lsblk = b''.join(sys_command(f'lsblk -J -o+PARTUUID {self.path}'))
|
||||||
for partition in json.loads(lsblk.decode('UTF-8'))['blockdevices']:
|
for partition in json.loads(lsblk.decode('UTF-8'))['blockdevices']:
|
||||||
return partition.get('partuuid', None)
|
return partition.get('partuuid', None)
|
||||||
|
return None
|
||||||
@property
|
@property
|
||||||
def encrypted(self):
|
def encrypted(self):
|
||||||
return self._encrypted
|
return self._encrypted
|
||||||
|
|
|
||||||
|
|
@ -5,6 +5,7 @@ from subprocess import Popen, STDOUT, PIPE, check_output
|
||||||
from select import epoll, EPOLLIN, EPOLLHUP
|
from select import epoll, EPOLLIN, EPOLLHUP
|
||||||
from .exceptions import *
|
from .exceptions import *
|
||||||
from .output import log
|
from .output import log
|
||||||
|
from typing import Optional, Union
|
||||||
|
|
||||||
def gen_uid(entropy_length=256):
|
def gen_uid(entropy_length=256):
|
||||||
return hashlib.sha512(os.urandom(entropy_length)).hexdigest()
|
return hashlib.sha512(os.urandom(entropy_length)).hexdigest()
|
||||||
|
|
@ -160,16 +161,15 @@ class sys_command():#Thread):
|
||||||
'exit_code': self.exit_code
|
'exit_code': self.exit_code
|
||||||
}
|
}
|
||||||
|
|
||||||
def peak(self, output :str):
|
def peak(self, output : Union[str, bytes]) -> bool:
|
||||||
if type(output) == bytes:
|
if type(output) == bytes:
|
||||||
try:
|
try:
|
||||||
output = output.decode('UTF-8')
|
output = output.decode('UTF-8')
|
||||||
except UnicodeDecodeError:
|
except UnicodeDecodeError:
|
||||||
return None
|
return False
|
||||||
|
|
||||||
output = output.strip('\r\n ')
|
output = output.strip('\r\n ')
|
||||||
if len(output) <= 0:
|
if len(output) <= 0:
|
||||||
return None
|
return False
|
||||||
|
|
||||||
if self.peak_output:
|
if self.peak_output:
|
||||||
from .user_interaction import get_terminal_width
|
from .user_interaction import get_terminal_width
|
||||||
|
|
@ -191,6 +191,7 @@ class sys_command():#Thread):
|
||||||
# And print the new output we're peaking on:
|
# And print the new output we're peaking on:
|
||||||
sys.stdout.write(output)
|
sys.stdout.write(output)
|
||||||
sys.stdout.flush()
|
sys.stdout.flush()
|
||||||
|
return True
|
||||||
|
|
||||||
def run(self):
|
def run(self):
|
||||||
self.status = 'running'
|
self.status = 'running'
|
||||||
|
|
|
||||||
|
|
@ -91,6 +91,7 @@ def cpuVendor()-> Optional[str]:
|
||||||
if info.get('field',None):
|
if info.get('field',None):
|
||||||
if info.get('field',None) == "Vendor ID:":
|
if info.get('field',None) == "Vendor ID:":
|
||||||
return info.get('data',None)
|
return info.get('data',None)
|
||||||
|
return None
|
||||||
|
|
||||||
def isVM() -> bool:
|
def isVM() -> bool:
|
||||||
try:
|
try:
|
||||||
|
|
|
||||||
|
|
@ -19,7 +19,7 @@ class journald(dict):
|
||||||
@abc.abstractmethod
|
@abc.abstractmethod
|
||||||
def log(message, level=logging.DEBUG):
|
def log(message, level=logging.DEBUG):
|
||||||
try:
|
try:
|
||||||
import systemd.journal
|
import systemd.journal # type: ignore
|
||||||
except ModuleNotFoundError:
|
except ModuleNotFoundError:
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -1,3 +1,4 @@
|
||||||
|
from typing import Optional
|
||||||
import os, urllib.request, urllib.parse, ssl, json, re
|
import os, urllib.request, urllib.parse, ssl, json, re
|
||||||
import importlib.util, sys, glob, hashlib, logging
|
import importlib.util, sys, glob, hashlib, logging
|
||||||
from collections import OrderedDict
|
from collections import OrderedDict
|
||||||
|
|
@ -49,7 +50,7 @@ def list_profiles(filter_irrelevant_macs=True, subpath='', filter_top_level_prof
|
||||||
except urllib.error.HTTPError as err:
|
except urllib.error.HTTPError as err:
|
||||||
print(f'Error: Listing profiles on URL "{profiles_url}" resulted in:', err)
|
print(f'Error: Listing profiles on URL "{profiles_url}" resulted in:', err)
|
||||||
return cache
|
return cache
|
||||||
except:
|
except json.decoder.JSONDecodeError as err:
|
||||||
print(f'Error: Could not decode "{profiles_url}" result as JSON:', err)
|
print(f'Error: Could not decode "{profiles_url}" result as JSON:', err)
|
||||||
return cache
|
return cache
|
||||||
|
|
||||||
|
|
@ -215,7 +216,7 @@ class Profile(Script):
|
||||||
return True
|
return True
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def packages(self) -> list:
|
def packages(self) -> Optional[list]:
|
||||||
"""
|
"""
|
||||||
Returns a list of packages baked into the profile definition.
|
Returns a list of packages baked into the profile definition.
|
||||||
If no package definition has been done, .packages() will return None.
|
If no package definition has been done, .packages() will return None.
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue