| import io | |
| import platform | |
| import queue | |
| import re | |
| import subprocess | |
| import sys | |
| import unittest | |
| from _android_support import TextLogStream | |
| from array import array | |
| from contextlib import ExitStack, contextmanager | |
| from threading import Thread | |
| from test.support import LOOPBACK_TIMEOUT | |
| from time import time | |
| from unittest.mock import patch | |
| if sys.platform != "android": | |
| raise unittest.SkipTest("Android-specific") | |
| # Some API levels filter out consecutive identical lines and replace them with a | |
| # "chatty" marker. | |
| api_level = platform.android_ver().api_level | |
| chatty_issue = (26 <= api_level <= 30) | |
| # (name, level, fileno) | |
| STREAM_INFO = [("stdout", "I", 1), ("stderr", "W", 2)] | |
| # Test redirection of stdout and stderr to the Android log. | |
| class TestAndroidOutput(unittest.TestCase): | |
| maxDiff = None | |
| def setUp(self): | |
| self.logcat_process = subprocess.Popen( | |
| ["logcat", "-v", "tag"], stdout=subprocess.PIPE, | |
| errors="backslashreplace" | |
| ) | |
| self.logcat_queue = queue.Queue() | |
| def logcat_thread(): | |
| for line in self.logcat_process.stdout: | |
| self.logcat_queue.put(line.rstrip("\n")) | |
| self.logcat_process.stdout.close() | |
| self.logcat_thread = Thread(target=logcat_thread) | |
| self.logcat_thread.start() | |
| try: | |
| from ctypes import CDLL, c_char_p, c_int | |
| from ctypes.util import wrap_dll_function | |
| liblog = CDLL("liblog.so") | |
| @wrap_dll_function(liblog) | |
| def __android_log_write(prio: c_int, tag: c_char_p, | |
| text: c_char_p) -> c_int: | |
| pass | |
| # Separate tests using a marker line with a different tag. | |
| ANDROID_LOG_INFO = 4 | |
| tag, message = "python.test", f"{self.id()} {time()}" | |
| __android_log_write( | |
| ANDROID_LOG_INFO, tag.encode("UTF-8"), message.encode("UTF-8")) | |
| self.assert_log("I", tag, message, skip=True) | |
| except: | |
| # If setUp throws an exception, tearDown is not automatically | |
| # called. Avoid leaving a dangling thread which would keep the | |
| # Python process alive indefinitely. | |
| self.tearDown() | |
| raise | |
| def assert_logs(self, level, tag, expected, **kwargs): | |
| for line in expected: | |
| self.assert_log(level, tag, line, **kwargs) | |
| def assert_log(self, level, tag, expected, *, skip=False): | |
| deadline = time() + LOOPBACK_TIMEOUT | |
| while True: | |
| try: | |
| line = self.logcat_queue.get(timeout=(deadline - time())) | |
| except queue.Empty: | |
| raise self.failureException( | |
| f"line not found: {expected!r}" | |
| ) from None | |
| if match := re.fullmatch(fr"(.)/{tag}: (.*)", line): | |
| try: | |
| self.assertEqual(level, match[1]) | |
| self.assertEqual(expected, match[2]) | |
| break | |
| except AssertionError: | |
| if not skip: | |
| raise | |
| def tearDown(self): | |
| self.logcat_process.terminate() | |
| self.logcat_process.wait(LOOPBACK_TIMEOUT) | |
| self.logcat_thread.join(LOOPBACK_TIMEOUT) | |
| # Avoid an irrelevant warning about threading._dangling. | |
| self.logcat_thread = None | |
| @contextmanager | |
| def reconfigure(self, stream, **settings): | |
| original_settings = {key: getattr(stream, key, None) for key in settings.keys()} | |
| stream.reconfigure(**settings) | |
| try: | |
| yield | |
| finally: | |
| stream.reconfigure(**original_settings) | |
| def stream_context(self, stream_name, level): | |
| stack = ExitStack() | |
| stack.enter_context(self.subTest(stream_name)) | |
| # In --verbose3 mode, sys.stdout and sys.stderr are captured, so we can't | |
| # test them directly. Detect this mode and use some temporary streams with | |
| # the same properties. | |
| stream = getattr(sys, stream_name) | |
| native_stream = getattr(sys, f"__{stream_name}__") | |
| if isinstance(stream, io.StringIO): | |
| # https://developer.android.com/ndk/reference/group/logging | |
| prio = {"I": 4, "W": 5}[level] | |
| stack.enter_context( | |
| patch( | |
| f"sys.{stream_name}", | |
| stream := TextLogStream( | |
| prio, f"python.{stream_name}", native_stream, | |
| ), | |
| ) | |
| ) | |
| # The tests assume the stream is initially buffered. | |
| stack.enter_context(self.reconfigure(stream, write_through=False)) | |
| return stack | |
| def test_str(self): | |
| for stream_name, level, fileno in STREAM_INFO: | |
| with self.stream_context(stream_name, level): | |
| stream = getattr(sys, stream_name) | |
| tag = f"python.{stream_name}" | |
| self.assertEqual(f"<TextLogStream '{tag}'>", repr(stream)) | |
| self.assertIs(stream.writable(), True) | |
| self.assertIs(stream.readable(), False) | |
| self.assertEqual(stream.fileno(), fileno) | |
| self.assertEqual("UTF-8", stream.encoding) | |
| self.assertEqual("backslashreplace", stream.errors) | |
| self.assertIs(stream.line_buffering, True) | |
| self.assertIs(stream.write_through, False) | |
| def write(s, lines=None, *, write_len=None): | |
| if write_len is None: | |
| write_len = len(s) | |
| self.assertEqual(write_len, stream.write(s)) | |
| if lines is None: | |
| lines = [s] | |
| self.assert_logs(level, tag, lines) | |
| # Single-line messages, | |
| with self.reconfigure(stream, write_through=True): | |
| write("", []) | |
| write("a") | |
| write("Hello") | |
| write("Hello world") | |
| write(" ") | |
| write(" ") | |
| # Non-ASCII text | |
| write("ol\u00e9") # Spanish | |
| write("\u4e2d\u6587") # Chinese | |
| # Non-BMP emoji | |
| write("\U0001f600") | |
| # Non-encodable surrogates | |
| write("\ud800\udc00", [r"\ud800\udc00"]) | |
| # Code used by surrogateescape (which isn't enabled here) | |
| write("\udc80", [r"\udc80"]) | |
| # Null characters are logged using "modified UTF-8". | |
| write("\u0000", [r"\xc0\x80"]) | |
| write("a\u0000", [r"a\xc0\x80"]) | |
| write("\u0000b", [r"\xc0\x80b"]) | |
| write("a\u0000b", [r"a\xc0\x80b"]) | |
| # Multi-line messages. Additional spaces will appear in the output where | |
| # necessary to protect leading newlines. | |
| write("\nx", [" "]) | |
| write("\na\n", ["x", "a"]) | |
| write("\n", [" "]) | |
| if not chatty_issue: | |
| write("\n\n", [" ", " "]) | |
| write("b\n", ["b"]) | |
| write("c\n\n", ["c", " "]) | |
| write("d\ne", ["d"]) | |
| write("xx", []) | |
| write("f\n\ng", ["exxf", " "]) | |
| write("\n", ["g"]) | |
| # Since this is a line-based logging system, line buffering | |
| # cannot be turned off, i.e. a newline always causes a flush. | |
| stream.reconfigure(line_buffering=False) | |
| self.assertIs(stream.line_buffering, True) | |
| # However, buffering can be turned off completely if you want a | |
| # flush after every write. | |
| with self.reconfigure(stream, write_through=True): | |
| write("\nx", [" ", "x"]) | |
| write("\na\n", [" ", "a"]) | |
| write("\n", [" "]) | |
| if not chatty_issue: | |
| write("\n\n", [" ", " "]) | |
| write("b\n", ["b"]) | |
| write("c\n\n", ["c", " "]) | |
| write("d\ne", ["d", "e"]) | |
| write("xx", ["xx"]) | |
| write("f\n\ng", ["f", " ", "g"]) | |
| write("\n", [" "]) | |
| # "\r\n" should be translated into "\n". | |
| write("hello\r\n", ["hello"]) | |
| write("hello\r\nworld\r\n", ["hello", "world"]) | |
| write("\r\n", [""]) | |
| # Non-standard line separators should be preserved. | |
| write("before form feed\x0cafter form feed\n", | |
| ["before form feed\x0cafter form feed"]) | |
| write("before line separator\u2028after line separator\n", | |
| ["before line separator\u2028after line separator"]) | |
| # String subclasses are accepted, but they should be converted | |
| # to a standard str without calling any of their methods. | |
| class CustomStr(str): | |
| def splitlines(self, *args, **kwargs): | |
| raise AssertionError() | |
| def __len__(self): | |
| raise AssertionError() | |
| def __str__(self): | |
| raise AssertionError() | |
| write(CustomStr("custom\n"), ["custom"], write_len=7) | |
| # Non-string classes are not accepted. | |
| for obj in [b"", b"hello", None, 42]: | |
| with self.subTest(obj=obj): | |
| with self.assertRaisesRegex( | |
| TypeError, | |
| fr"write\(\) argument must be str, not " | |
| fr"{type(obj).__name__}" | |
| ): | |
| stream.write(obj) | |
| # Manual flushing is supported. | |
| write("hello", []) | |
| stream.flush() | |
| self.assert_log(level, tag, "hello") | |
| write("hello", []) | |
| write("world", []) | |
| stream.flush() | |
| self.assert_log(level, tag, "helloworld") | |
| # Long lines are split into blocks of 1000 characters | |
| # (MAX_CHARS_PER_WRITE in _android_support.py), but | |
| # TextIOWrapper should then join them back together as much as | |
| # possible without exceeding 4000 UTF-8 bytes | |
| # (MAX_BYTES_PER_WRITE). | |
| # | |
| # ASCII (1 byte per character) | |
| write(("foobar" * 700) + "\n", # 4200 bytes in | |
| [("foobar" * 666) + "foob", # 4000 bytes out | |
| "ar" + ("foobar" * 33)]) # 200 bytes out | |
| # "Full-width" digits 0-9 (3 bytes per character) | |
| s = "\uff10\uff11\uff12\uff13\uff14\uff15\uff16\uff17\uff18\uff19" | |
| write((s * 150) + "\n", # 4500 bytes in | |
| [s * 100, # 3000 bytes out | |
| s * 50]) # 1500 bytes out | |
| s = "0123456789" | |
| write(s * 200, []) # 2000 bytes in | |
| write(s * 150, []) # 1500 bytes in | |
| write(s * 51, [s * 350]) # 510 bytes in, 3500 bytes out | |
| write("\n", [s * 51]) # 0 bytes in, 510 bytes out | |
| def test_bytes(self): | |
| for stream_name, level, fileno in STREAM_INFO: | |
| with self.stream_context(stream_name, level): | |
| stream = getattr(sys, stream_name).buffer | |
| tag = f"python.{stream_name}" | |
| self.assertEqual(f"<BinaryLogStream '{tag}'>", repr(stream)) | |
| self.assertIs(stream.writable(), True) | |
| self.assertIs(stream.readable(), False) | |
| self.assertEqual(stream.fileno(), fileno) | |
| def write(b, lines=None, *, write_len=None): | |
| if write_len is None: | |
| write_len = len(b) | |
| self.assertEqual(write_len, stream.write(b)) | |
| if lines is None: | |
| lines = [b.decode()] | |
| self.assert_logs(level, tag, lines) | |
| # Single-line messages, | |
| write(b"", []) | |
| write(b"a") | |
| write(b"Hello") | |
| write(b"Hello world") | |
| write(b" ") | |
| write(b" ") | |
| # Non-ASCII text | |
| write(b"ol\xc3\xa9") # Spanish | |
| write(b"\xe4\xb8\xad\xe6\x96\x87") # Chinese | |
| # Non-BMP emoji | |
| write(b"\xf0\x9f\x98\x80") | |
| # Null bytes are logged using "modified UTF-8". | |
| write(b"\x00", [r"\xc0\x80"]) | |
| write(b"a\x00", [r"a\xc0\x80"]) | |
| write(b"\x00b", [r"\xc0\x80b"]) | |
| write(b"a\x00b", [r"a\xc0\x80b"]) | |
| # Invalid UTF-8 | |
| write(b"\xff", [r"\xff"]) | |
| write(b"a\xff", [r"a\xff"]) | |
| write(b"\xffb", [r"\xffb"]) | |
| write(b"a\xffb", [r"a\xffb"]) | |
| # Log entries containing newlines are shown differently by | |
| # `logcat -v tag`, `logcat -v long`, and Android Studio. We | |
| # currently use `logcat -v tag`, which shows each line as if it | |
| # was a separate log entry, but strips a single trailing | |
| # newline. | |
| write(b"\nx", [" ", "x"]) | |
| write(b"\na\n", [" ", "a"]) | |
| write(b"\n", [" "]) | |
| write(b"\n\n", [" ", ""]) | |
| write(b"b\n", ["b"]) | |
| write(b"c\n\n", ["c", ""]) | |
| write(b"d\ne", ["d", "e"]) | |
| write(b"xx", ["xx"]) | |
| write(b"f\n\ng", ["f", "", "g"]) | |
| write(b"\n", [" "]) | |
| # "\r\n" should be translated into "\n". | |
| write(b"hello\r\n", ["hello"]) | |
| write(b"hello\r\nworld\r\n", ["hello", "world"]) | |
| write(b"\r\n", [""]) | |
| # Other bytes-like objects are accepted. | |
| write(bytearray(b"bytearray")) | |
| mv = memoryview(b"memoryview") | |
| write(mv, ["memoryview"]) # Continuous | |
| write(mv[::2], ["mmrve"]) # Discontinuous | |
| write( | |
| # Android only supports little-endian architectures, so the | |
| # bytes representation is as follows: | |
| array("H", [ | |
| 0, # 00 00 | |
| 1, # 01 00 | |
| 65534, # FE FF | |
| 65535, # FF FF | |
| ]), | |
| # After encoding null bytes with modified UTF-8, the only | |
| # valid UTF-8 sequence is \x01. All other bytes are handled | |
| # by backslashreplace. | |
| ["\\xc0\\x80\\xc0\\x80" | |
| "\x01\\xc0\\x80" | |
| "\\xfe\\xff" | |
| "\\xff\\xff"], | |
| write_len=8, | |
| ) | |
| # Non-bytes-like classes are not accepted. | |
| for obj in ["", "hello", None, 42]: | |
| with self.subTest(obj=obj): | |
| with self.assertRaisesRegex( | |
| TypeError, | |
| fr"write\(\) argument must be bytes-like, not " | |
| fr"{type(obj).__name__}" | |
| ): | |
| stream.write(obj) | |
| class TestAndroidRateLimit(unittest.TestCase): | |
| def test_rate_limit(self): | |
| # https://cs.android.com/android/platform/superproject/+/android-14.0.0_r1:system/logging/liblog/include/log/log_read.h;l=39 | |
| PER_MESSAGE_OVERHEAD = 28 | |
| # https://developer.android.com/ndk/reference/group/logging | |
| ANDROID_LOG_DEBUG = 3 | |
| # To avoid flooding the test script output, use a different tag rather | |
| # than stdout or stderr. | |
| tag = "python.rate_limit" | |
| stream = TextLogStream(ANDROID_LOG_DEBUG, tag) | |
| # Make a test message which consumes 1 KB of the logcat buffer. | |
| message = "Line {:03d} " | |
| message += "." * ( | |
| 1024 - PER_MESSAGE_OVERHEAD - len(tag) - len(message.format(0)) | |
| ) + "\n" | |
| # To avoid depending on the performance of the test device, we mock the | |
| # passage of time. | |
| mock_now = time() | |
| def mock_time(): | |
| # Avoid division by zero by simulating a small delay. | |
| mock_sleep(0.0001) | |
| return mock_now | |
| def mock_sleep(duration): | |
| nonlocal mock_now | |
| mock_now += duration | |
| # See _android_support.py. The default values of these parameters work | |
| # well across a wide range of devices, but we'll use smaller values to | |
| # ensure a quick and reliable test that doesn't flood the log too much. | |
| MAX_KB_PER_SECOND = 100 | |
| BUCKET_KB = 10 | |
| with ( | |
| patch("_android_support.MAX_BYTES_PER_SECOND", MAX_KB_PER_SECOND * 1024), | |
| patch("_android_support.BUCKET_SIZE", BUCKET_KB * 1024), | |
| patch("_android_support.sleep", mock_sleep), | |
| patch("_android_support.time", mock_time), | |
| ): | |
| # Make sure the token bucket is full. | |
| stream.write("Initial message to reset _prev_write_time") | |
| mock_sleep(BUCKET_KB / MAX_KB_PER_SECOND) | |
| line_num = 0 | |
| # Write BUCKET_KB messages, and return the rate at which they were | |
| # accepted in KB per second. | |
| def write_bucketful(): | |
| nonlocal line_num | |
| start = mock_time() | |
| max_line_num = line_num + BUCKET_KB | |
| while line_num < max_line_num: | |
| stream.write(message.format(line_num)) | |
| line_num += 1 | |
| return BUCKET_KB / (mock_time() - start) | |
| # The first bucketful should be written with minimal delay. The | |
| # factor of 2 here is not arbitrary: it verifies that the system can | |
| # write fast enough to empty the bucket within two bucketfuls, which | |
| # the next part of the test depends on. | |
| self.assertGreater(write_bucketful(), MAX_KB_PER_SECOND * 2) | |
| # Write another bucketful to empty the token bucket completely. | |
| write_bucketful() | |
| # The next bucketful should be written at the rate limit. | |
| self.assertAlmostEqual( | |
| write_bucketful(), MAX_KB_PER_SECOND, | |
| delta=MAX_KB_PER_SECOND * 0.1 | |
| ) | |
| # Once the token bucket refills, we should go back to full speed. | |
| mock_sleep(BUCKET_KB / MAX_KB_PER_SECOND) | |
| self.assertGreater(write_bucketful(), MAX_KB_PER_SECOND * 2) |