| 107 | |
| 108 | |
| 109 | class BinaryLogStream(io.RawIOBase): |
| 110 | def __init__(self, prio, tag, fileno=None): |
| 111 | self.prio = prio |
| 112 | self.tag = tag |
| 113 | self._fileno = fileno |
| 114 | |
| 115 | def __repr__(self): |
| 116 | return f"<BinaryLogStream {self.tag!r}>" |
| 117 | |
| 118 | def writable(self): |
| 119 | return True |
| 120 | |
| 121 | def write(self, b): |
| 122 | if type(b) is not bytes: |
| 123 | try: |
| 124 | b = bytes(memoryview(b)) |
| 125 | except TypeError: |
| 126 | raise TypeError( |
| 127 | f"write() argument must be bytes-like, not {type(b).__name__}" |
| 128 | ) from None |
| 129 | |
| 130 | # Writing an empty string to the stream should have no effect. |
| 131 | if b: |
| 132 | logcat.write(self.prio, self.tag, b) |
| 133 | return len(b) |
| 134 | |
| 135 | # This is needed by the test suite --timeout option, which uses faulthandler. |
| 136 | def fileno(self): |
| 137 | if self._fileno is None: |
| 138 | raise io.UnsupportedOperation("fileno") |
| 139 | return self._fileno |
| 140 | |
| 141 | |
| 142 | # When a large volume of data is written to logcat at once, e.g. when a test |
no outgoing calls
no test coverage detected
searching dependent graphs…