Assert that a layer emits the expected commands in reaction to a given sequence of events. For example, the following code asserts that the TCP layer emits an OpenConnection command immediately after starting and does not yield any further commands as a reaction to successful connec
| 107 | |
| 108 | |
| 109 | class Playbook: |
| 110 | """ |
| 111 | Assert that a layer emits the expected commands in reaction to a given sequence of events. |
| 112 | For example, the following code asserts that the TCP layer emits an OpenConnection command |
| 113 | immediately after starting and does not yield any further commands as a reaction to successful |
| 114 | connection establishment. |
| 115 | |
| 116 | assert playbook(tcp.TCPLayer(tctx)) \ |
| 117 | << commands.OpenConnection(tctx.server) |
| 118 | >> reply(None) |
| 119 | << None # this line is optional. |
| 120 | |
| 121 | This is syntactic sugar for the following: |
| 122 | |
| 123 | t = tcp.TCPLayer(tctx) |
| 124 | x1 = list(t.handle_event(events.Start())) |
| 125 | assert x1 == [commands.OpenConnection(tctx.server)] |
| 126 | x2 = list(t.handle_event(events.OpenConnectionReply(x1[-1]))) |
| 127 | assert x2 == [] |
| 128 | """ |
| 129 | |
| 130 | layer: Layer |
| 131 | """The base layer""" |
| 132 | expected: PlaybookEntryList |
| 133 | """expected command/event sequence""" |
| 134 | actual: PlaybookEntryList |
| 135 | """actual command/event sequence""" |
| 136 | _errored: bool |
| 137 | """used to check if playbook as been fully asserted""" |
| 138 | logs: bool |
| 139 | """If False, the playbook specification doesn't contain log commands.""" |
| 140 | hooks: bool |
| 141 | """If False, the playbook specification doesn't include hooks or hook replies. They are automatically replied to.""" |
| 142 | |
| 143 | def __init__( |
| 144 | self, |
| 145 | layer: Layer, |
| 146 | hooks: bool = True, |
| 147 | logs: bool = False, |
| 148 | expected: PlaybookEntryList | None = None, |
| 149 | ): |
| 150 | if expected is None: |
| 151 | expected = [events.Start()] |
| 152 | |
| 153 | self.layer = layer |
| 154 | self.expected = expected |
| 155 | self.actual = [] |
| 156 | self._errored = False |
| 157 | self.logs = logs |
| 158 | self.hooks = hooks |
| 159 | |
| 160 | def __rshift__(self, e): |
| 161 | """Add an event to send""" |
| 162 | if isinstance(e, collections.abc.Iterable): |
| 163 | for ev in e: |
| 164 | self.__rshift__(ev) |
| 165 | return self |
| 166 | assert isinstance(e, events.Event) |
no outgoing calls
searching dependent graphs…