Determine if playbook is correct.
(self)
| 191 | return self |
| 192 | |
| 193 | def __bool__(self): |
| 194 | """Determine if playbook is correct.""" |
| 195 | already_asserted = len(self.actual) |
| 196 | i = already_asserted |
| 197 | while i < len(self.expected): |
| 198 | x = self.expected[i] |
| 199 | if isinstance(x, commands.Command): |
| 200 | pass |
| 201 | else: |
| 202 | if hasattr(x, "playbook_eval"): |
| 203 | try: |
| 204 | x = self.expected[i] = x.playbook_eval(self) |
| 205 | except Exception: |
| 206 | self.actual.append(_TracebackInPlaybook(traceback.format_exc())) |
| 207 | break |
| 208 | for name, value in vars(x).items(): |
| 209 | if isinstance(value, _Placeholder): |
| 210 | setattr(x, name, value()) |
| 211 | if isinstance(x, events.OpenConnectionCompleted) and not x.reply: |
| 212 | x.command.connection.state = ConnectionState.OPEN |
| 213 | x.command.connection.timestamp_start = 1624544785 |
| 214 | elif isinstance(x, events.ConnectionClosed): |
| 215 | x.connection.state &= ~ConnectionState.CAN_READ |
| 216 | x.connection.timestamp_end = 1624544787 |
| 217 | |
| 218 | self.actual.append(x) |
| 219 | cmds: list[commands.Command] = [] |
| 220 | try: |
| 221 | # consume them one by one so that we can extend the log with all commands until traceback. |
| 222 | for cmd in self.layer.handle_event(x): |
| 223 | cmds.append(cmd) |
| 224 | except Exception: |
| 225 | self.actual.extend(cmds) |
| 226 | self.actual.append(_TracebackInPlaybook(traceback.format_exc())) |
| 227 | break |
| 228 | |
| 229 | cmds = list( |
| 230 | _merge_sends( |
| 231 | cmds, ignore_hooks=not self.hooks, ignore_logs=not self.logs |
| 232 | ) |
| 233 | ) |
| 234 | |
| 235 | self.actual.extend(cmds) |
| 236 | pos = len(self.actual) - len(cmds) - 1 |
| 237 | hook_replies = [] |
| 238 | for cmd in cmds: |
| 239 | pos += 1 |
| 240 | assert self.actual[pos] == cmd |
| 241 | if isinstance(cmd, commands.CloseTcpConnection) and cmd.half_close: |
| 242 | cmd.connection.state &= ~ConnectionState.CAN_WRITE |
| 243 | elif isinstance(cmd, commands.CloseConnection): |
| 244 | cmd.connection.state = ConnectionState.CLOSED |
| 245 | elif isinstance(cmd, commands.Log): |
| 246 | need_to_emulate_log = ( |
| 247 | not self.logs |
| 248 | and cmd.level in (logging.DEBUG, logging.INFO) |
| 249 | and ( |
| 250 | pos >= len(self.expected) |
nothing calls this directly
no test coverage detected