| Server IP : 74.208.127.88 / Your IP : 216.73.216.214 Web Server : Apache/2.4.41 (Ubuntu) System : Linux ubuntu 5.4.0-163-generic #180-Ubuntu SMP Tue Sep 5 13:21:23 UTC 2023 x86_64 User : www-data ( 33) PHP Version : 7.4.3-4ubuntu2.29 Disable Function : pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,pcntl_unshare, MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : OFF | Sudo : ON | Pkexec : ON Directory : /proc/self/root/snap/core20/current/usr/share/subiquity/console_conf/models/ |
Upload File : |
#!/usr/bin/env python3
# Copyright 2020 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 <http://www.gnu.org/licenses/>.
import logging
import json
import attr
import jsonschema
log = logging.getLogger("console_conf.models.systems")
# This json schema describes the recovery systems data. Allow additional
# properties at each level so that console-conf does not have to be exactly in
# sync with snapd.
_RECOVERY_SYSTEMS_SCHEMA = {
"$schema": "http://json-schema.org/draft-04/schema#",
"title": "systems",
"type": "object",
"additionalProperties": True,
"required": ["systems"],
"properties": {
"systems": {
"type": "array",
"items": {
"type": "object",
"required": ["label", "brand", "model"],
"properties": {
"label": {"type": "string"},
"actions": {
"type": "array",
"items": {
"type": "object",
"additionalProperties": True,
"required": ["title", "mode"],
"properties": {
"title": {"type": "string"},
"mode": {"type": "string"},
},
},
},
"brand": {
"type": "object",
"additionalProperties": True,
"required": ["id", "username", "display-name"],
"properties": {
"id": {"type": "string"},
"username": {"type": "string"},
"display-name": {"type": "string"},
"validation": {"type": "string"},
},
},
"model": {
"type": "object",
"additionalProperties": True,
"required": ["model", "brand-id", "display-name"],
"properties": {
"model": {"type": "string"},
"brand-id": {"type": "string"},
"display-name": {"type": "string"},
},
},
},
},
},
},
}
class RecoverySystemsModel:
"""Recovery chooser data"""
def __init__(self, systems_data):
self.systems = systems_data
# current selection
self._selection = None
self._current = None
# find which system is current one, but be robust if none is marked as
# such
cs = [s for s in systems_data if s.current]
if cs:
self._current = cs[0]
def select(self, system, action):
self._selection = SelectedSystemAction(system=system, action=action)
def unselect(self):
self._selection = None
@property
def selection(self):
return self._selection
@property
def current(self):
return self._current
@staticmethod
def from_systems(recovery_systems):
systems = []
for syst in recovery_systems:
m = syst["model"]
b = syst["brand"]
model = SystemModel(
model=m["model"],
brand_id=m["brand-id"],
display_name=m["display-name"]
)
brand = Brand(
ID=b["id"],
username=b["username"],
display_name=b["display-name"],
validation=b.get("validation", "unproven"),
)
actions = []
for a in syst.get("actions", []):
actions.append(SystemAction(title=a["title"], mode=a["mode"]))
s = RecoverySystem(
current=syst.get("current", False),
label=syst["label"],
model=model,
brand=brand,
actions=actions,
)
systems.append(s)
return RecoverySystemsModel(systems)
@staticmethod
def from_systems_stream(chooser_input):
"""Deserialize recovery systems from input JSON stream."""
try:
dec = json.load(chooser_input)
jsonschema.validate(dec, _RECOVERY_SYSTEMS_SCHEMA)
systems = dec.get("systems", [])
except json.JSONDecodeError:
log.exception("cannot decode recovery systems info")
raise
except jsonschema.ValidationError:
log.exception("cannot validate recovery systems data")
raise
return RecoverySystemsModel.from_systems(systems)
@staticmethod
def to_response_stream(obj, chooser_output):
"""Serialize an object with selected action as JSON to the given output
stream.
"""
if not isinstance(obj, SelectedSystemAction):
raise TypeError("unexpected type: {}".format(type(obj)))
choice = {
"label": obj.system.label,
"action": {
"mode": obj.action.mode,
"title": obj.action.title,
},
}
return json.dump(choice, fp=chooser_output)
@attr.s
class RecoverySystem:
current = attr.ib()
label = attr.ib()
model = attr.ib()
brand = attr.ib()
actions = attr.ib()
def __eq__(self, other):
return self.current == other.current and \
self.label == other.label and \
self.model == other.model and \
self.brand == other.brand and \
self.actions == other.actions
@attr.s
class Brand:
ID = attr.ib()
username = attr.ib()
display_name = attr.ib()
validation = attr.ib()
def __eq__(self, other):
return self.ID == other.ID and \
self.username == other.username and \
self.display_name == other.display_name and \
self.validation == other.validation
@attr.s
class SystemModel:
model = attr.ib()
brand_id = attr.ib()
display_name = attr.ib()
def __eq__(self, other):
return self.model == other.model and \
self.brand_id == other.brand_id and \
self.display_name == other.display_name
@attr.s
class SystemAction:
title = attr.ib()
mode = attr.ib()
def __eq__(self, other):
return self.title == other.title and \
self.mode == other.mode
@attr.s
class SelectedSystemAction:
system = attr.ib()
action = attr.ib()
def __eq__(self, other):
return self.system == other.system and \
self.action == other.action