FM / subiquitycore

/snap/core20/2866/usr/share/subiquity/subiquitycore/

/snap/core20/2866/usr/share/subiquity/subiquitycore

Name	Size	Mode	Actions
controllers/	-	0755	rm
models/	-	0755	rm
testing/	-	0755	rm
tests/	-	0755	rm
ui/	-	0755	rm
__pycache__/	-	0755	rm
async_helpers.py	2327	0644	edit dl rm
context.py	4414	0644	edit dl rm
controller.py	4624	0644	edit dl rm
controllerset.py	1898	0644	edit dl rm
core.py	16738	0644	edit dl rm
file_util.py	832	0644	edit dl rm
gettext38.py	27565	0644	edit dl rm
i18n.py	1780	0644	edit dl rm
log.py	1516	0644	edit dl rm
lsb_release.py	734	0644	edit dl rm
netplan.py	5471	0644	edit dl rm
palette.py	4641	0644	edit dl rm
prober.py	2048	0644	edit dl rm
screen.py	5017	0644	edit dl rm
signals.py	2017	0644	edit dl rm
snapd.py	5932	0644	edit dl rm
ssh.py	3062	0644	edit dl rm
utils.py	5366	0644	edit dl rm
view.py	3433	0644	edit dl rm
__init__.py	710	0644	edit dl rm

Edit: /snap/core20/2866/usr/share/subiquity/subiquitycore/prober.py (2048B)

import logging
import time
import yaml

from probert.network import (
    StoredDataObserver,
    UdevObserver,
    )

log = logging.getLogger('subiquitycore.prober')

class Prober():
    def __init__(self, machine_config, debug_flags):
        self.saved_config = None
        if machine_config:
            with open(machine_config) as mc:
                self.saved_config = yaml.safe_load(mc)
        self.debug_flags = debug_flags
        log.debug('Prober() init finished, data:{}'.format(self.saved_config))

def probe_network(self, receiver):
        if self.saved_config is not None:
            observer = StoredDataObserver(
                self.saved_config['network'], receiver)
        else:
            observer = UdevObserver(receiver)
        return observer, observer.start()

def get_storage(self, probe_types=None):
        if self.saved_config is not None:
            flag = 'bpfail-full'
            if probe_types is not None:
                flag = 'bpfail-restricted'
            if flag in self.debug_flags:
                time.sleep(2)
                1/0
            r = self.saved_config['storage'].copy()
            if probe_types is not None:
                for k in self.saved_config['storage']:
                    if k not in probe_types:
                        r[k] = {}
            return r
        from probert.storage import Storage
        return Storage().probe(probe_types=probe_types)

# Copyright 2015 Canonical, Ltd. # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, version 3. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see . import logging import time import yaml from probert.network import ( StoredDataObserver, UdevObserver, ) log = logging.getLogger('subiquitycore.prober') class Prober(): def __init__(self, machine_config, debug_flags): self.saved_config = None if machine_config: with open(machine_config) as mc: self.saved_config = yaml.safe_load(mc) self.debug_flags = debug_flags log.debug('Prober() init finished, data:{}'.format(self.saved_config)) def probe_network(self, receiver): if self.saved_config is not None: observer = StoredDataObserver( self.saved_config['network'], receiver) else: observer = UdevObserver(receiver) return observer, observer.start() def get_storage(self, probe_types=None): if self.saved_config is not None: flag = 'bpfail-full' if probe_types is not None: flag = 'bpfail-restricted' if flag in self.debug_flags: time.sleep(2) 1/0 r = self.saved_config['storage'].copy() if probe_types is not None: for k in self.saved_config['storage']: if k not in probe_types: r[k] = {} return r from probert.storage import Storage return Storage().probe(probe_types=probe_types)