(self, message)
| 200 | ) |
| 201 | |
| 202 | def process_message(self, message): |
| 203 | if message['message_type'] == 'heartbeat' and not self._heartbeat_is_animating: |
| 204 | self._heartbeat_is_animating = True |
| 205 | self._animate_heartbeat() |
| 206 | |
| 207 | if not self.seen_message and message['message_type'] == 'certificate_update': |
| 208 | self.right_text = urwid.Text('') |
| 209 | self.frame.set_body( |
| 210 | urwid.Columns( |
| 211 | widget_list=[ |
| 212 | urwid.Pile( |
| 213 | [ |
| 214 | SidelessLineBox( |
| 215 | self.list_box, |
| 216 | title="CertStream Messages", |
| 217 | bline="", |
| 218 | ), |
| 219 | ( |
| 220 | 'flow', |
| 221 | urwid.Columns([ |
| 222 | ('fixed', 6, urwid.Text(u'└─────')), |
| 223 | ('flow', self.counter_text), |
| 224 | Divider('─'), |
| 225 | ('fixed', 1, urwid.Text(u'┘')), |
| 226 | ]) |
| 227 | ) |
| 228 | ] |
| 229 | ), |
| 230 | SidelessLineBox( |
| 231 | urwid.Filler( |
| 232 | self.right_text, |
| 233 | valign=urwid.TOP |
| 234 | ), |
| 235 | title="Parsed JSON" |
| 236 | ) |
| 237 | ], |
| 238 | ) |
| 239 | ) |
| 240 | self.seen_message = True |
| 241 | |
| 242 | if message['message_type'] == 'certificate_update': |
| 243 | _, original_offset = self.list_box.get_focus() |
| 244 | self.list_walker.insert(0, |
| 245 | urwid.AttrMap( |
| 246 | FauxButton( |
| 247 | "[{}] {} - {}".format( |
| 248 | message['data']['cert_index'], |
| 249 | message['data']['source']['url'], |
| 250 | message['data']['leaf_cert']['subject']['CN'], |
| 251 | ), |
| 252 | user_data=message, |
| 253 | on_press=self.focus_right_panel |
| 254 | ), |
| 255 | '', |
| 256 | focus_map='buttons' |
| 257 | ) |
| 258 | ) |
| 259 |
no test coverage detected