(self)
| 141 | Flow.__types[cls.type] = cls |
| 142 | |
| 143 | def get_state(self) -> serializable.State: |
| 144 | state = { |
| 145 | "version": version.FLOW_FORMAT_VERSION, |
| 146 | "type": self.type, |
| 147 | "id": self.id, |
| 148 | "error": self.error.get_state() if self.error else None, |
| 149 | "client_conn": self.client_conn.get_state(), |
| 150 | "server_conn": self.server_conn.get_state(), |
| 151 | "intercepted": self.intercepted, |
| 152 | "is_replay": self.is_replay, |
| 153 | "marked": self.marked, |
| 154 | "metadata": copy.deepcopy(self.metadata), |
| 155 | "comment": self.comment, |
| 156 | "timestamp_created": self.timestamp_created, |
| 157 | } |
| 158 | state["backup"] = copy.deepcopy(self._backup) if self._backup != state else None |
| 159 | return state |
| 160 | |
| 161 | def set_state(self, state: serializable.State) -> None: |
| 162 | assert state.pop("version") == version.FLOW_FORMAT_VERSION |
no outgoing calls
no test coverage detected