1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
|
"""
QMP Data Models
This module provides simplistic data classes that represent the few
structures that the QMP spec mandates; they are used to verify incoming
data to make sure it conforms to spec.
"""
# pylint: disable=too-few-public-methods
from collections import abc
import copy
from typing import (
Any,
Dict,
Mapping,
Optional,
Sequence,
)
class Model:
"""
Abstract data model, representing some QMP object of some kind.
:param raw: The raw object to be validated.
:raise KeyError: If any required fields are absent.
:raise TypeError: If any required fields have the wrong type.
"""
def __init__(self, raw: Mapping[str, Any]):
self._raw = raw
def _check_key(self, key: str) -> None:
if key not in self._raw:
raise KeyError(f"'{self._name}' object requires '{key}' member")
def _check_value(self, key: str, type_: type, typestr: str) -> None:
assert key in self._raw
if not isinstance(self._raw[key], type_):
raise TypeError(
f"'{self._name}' member '{key}' must be a {typestr}"
)
def _check_member(self, key: str, type_: type, typestr: str) -> None:
self._check_key(key)
self._check_value(key, type_, typestr)
@property
def _name(self) -> str:
return type(self).__name__
def __repr__(self) -> str:
return f"{self._name}({self._raw!r})"
class Greeting(Model):
"""
Defined in qmp-spec.txt, section 2.2, "Server Greeting".
:param raw: The raw Greeting object.
:raise KeyError: If any required fields are absent.
:raise TypeError: If any required fields have the wrong type.
"""
def __init__(self, raw: Mapping[str, Any]):
super().__init__(raw)
#: 'QMP' member
self.QMP: QMPGreeting # pylint: disable=invalid-name
self._check_member('QMP', abc.Mapping, "JSON object")
self.QMP = QMPGreeting(self._raw['QMP'])
def _asdict(self) -> Dict[str, object]:
"""
For compatibility with the iotests sync QMP wrapper.
The legacy QMP interface needs Greetings as a garden-variety Dict.
This interface is private in the hopes that it will be able to
be dropped again in the near-future. Caller beware!
"""
return dict(copy.deepcopy(self._raw))
class QMPGreeting(Model):
"""
Defined in qmp-spec.txt, section 2.2, "Server Greeting".
:param raw: The raw QMPGreeting object.
:raise KeyError: If any required fields are absent.
:raise TypeError: If any required fields have the wrong type.
"""
def __init__(self, raw: Mapping[str, Any]):
super().__init__(raw)
#: 'version' member
self.version: Mapping[str, object]
#: 'capabilities' member
self.capabilities: Sequence[object]
self._check_member('version', abc.Mapping, "JSON object")
self.version = self._raw['version']
self._check_member('capabilities', abc.Sequence, "JSON array")
self.capabilities = self._raw['capabilities']
class ErrorResponse(Model):
"""
Defined in qmp-spec.txt, section 2.4.2, "error".
:param raw: The raw ErrorResponse object.
:raise KeyError: If any required fields are absent.
:raise TypeError: If any required fields have the wrong type.
"""
def __init__(self, raw: Mapping[str, Any]):
super().__init__(raw)
#: 'error' member
self.error: ErrorInfo
#: 'id' member
self.id: Optional[object] = None # pylint: disable=invalid-name
self._check_member('error', abc.Mapping, "JSON object")
self.error = ErrorInfo(self._raw['error'])
if 'id' in raw:
self.id = raw['id']
class ErrorInfo(Model):
"""
Defined in qmp-spec.txt, section 2.4.2, "error".
:param raw: The raw ErrorInfo object.
:raise KeyError: If any required fields are absent.
:raise TypeError: If any required fields have the wrong type.
"""
def __init__(self, raw: Mapping[str, Any]):
super().__init__(raw)
#: 'class' member, with an underscore to avoid conflicts in Python.
self.class_: str
#: 'desc' member
self.desc: str
self._check_member('class', str, "string")
self.class_ = self._raw['class']
self._check_member('desc', str, "string")
self.desc = self._raw['desc']
|
