(self, s)
| 869 | _WRITE_LOG_PREFIX = "write: " |
| 870 | |
| 871 | def write(self, s): |
| 872 | from _pydevd_bundle.pydevd_comm import ID_TO_MEANING |
| 873 | |
| 874 | meaning = ID_TO_MEANING.get(re.search(r"\d+", s).group(), "") |
| 875 | if meaning: |
| 876 | meaning += ": " |
| 877 | |
| 878 | self.log.append( |
| 879 | self._WRITE_LOG_PREFIX |
| 880 | + "%s%s" |
| 881 | % ( |
| 882 | meaning, |
| 883 | s, |
| 884 | ) |
| 885 | ) |
| 886 | |
| 887 | if SHOW_WRITES_AND_READS: |
| 888 | print( |
| 889 | "Test Writer Thread Written %s%s" |
| 890 | % ( |
| 891 | meaning, |
| 892 | s, |
| 893 | ) |
| 894 | ) |
| 895 | msg = s + "\n" |
| 896 | |
| 897 | if not hasattr(self, "sock"): |
| 898 | print("%s.sock not available when sending: %s" % (self, msg)) |
| 899 | return |
| 900 | |
| 901 | msg = msg.encode("utf-8") |
| 902 | |
| 903 | self.sock.send(msg) |
| 904 | |
| 905 | def get_next_message(self, context_message, timeout=None): |
| 906 | return self.reader_thread.get_next_message(context_message, timeout=timeout) |
no test coverage detected