cpython/Lib/test/test_android.py

333 lines
13 KiB
Python
Raw Normal View History

import platform
import queue
import re
import subprocess
import sys
import unittest
from array import array
from contextlib import contextmanager
from threading import Thread
from test.support import LOOPBACK_TIMEOUT
from time import time
if sys.platform != "android":
raise unittest.SkipTest("Android-specific")
api_level = platform.android_ver().api_level
# Test redirection of stdout and stderr to the Android log.
@unittest.skipIf(
api_level < 23 and platform.machine() == "aarch64",
"SELinux blocks reading logs on older ARM64 emulators"
)
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()
Thread(target=logcat_thread).start()
from ctypes import CDLL, c_char_p, c_int
android_log_write = getattr(CDLL("liblog.so"), "__android_log_write")
android_log_write.argtypes = (c_int, c_char_p, c_char_p)
ANDROID_LOG_INFO = 4
# Separate tests using a marker line with a different tag.
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, timeout=5)
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, timeout=0.5):
deadline = time() + timeout
while True:
try:
line = self.logcat_queue.get(timeout=(deadline - time()))
except queue.Empty:
self.fail(f"line not found: {expected!r}")
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)
@contextmanager
def unbuffered(self, stream):
stream.reconfigure(write_through=True)
try:
yield
finally:
stream.reconfigure(write_through=False)
def test_str(self):
for stream_name, level in [("stdout", "I"), ("stderr", "W")]:
with self.subTest(stream=stream_name):
stream = getattr(sys, stream_name)
tag = f"python.{stream_name}"
self.assertEqual(f"<TextLogStream '{tag}'>", repr(stream))
self.assertTrue(stream.writable())
self.assertFalse(stream.readable())
self.assertEqual("UTF-8", stream.encoding)
self.assertTrue(stream.line_buffering)
self.assertFalse(stream.write_through)
# stderr is backslashreplace by default; stdout is configured
# that way by libregrtest.main.
self.assertEqual("backslashreplace", stream.errors)
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.unbuffered(stream):
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. Avoid identical consecutive lines, as
# they may activate "chatty" filtering and break the tests.
write("\nx", [""])
write("\na\n", ["x", "a"])
write("\n", [""])
write("b\n", ["b"])
write("c\n\n", ["c", ""])
write("d\ne", ["d"])
write("xx", [])
write("f\n\ng", ["exxf", ""])
write("\n", ["g"])
with self.unbuffered(stream):
write("\nx", ["", "x"])
write("\na\n", ["", "a"])
write("\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",
[("foobar" * 666) + "foob", # 4000 bytes
"ar" + ("foobar" * 33)]) # 200 bytes
# "Full-width" digits 0-9 (3 bytes per character)
s = "\uff10\uff11\uff12\uff13\uff14\uff15\uff16\uff17\uff18\uff19"
write((s * 150) + "\n",
[s * 100, # 3000 bytes
s * 50]) # 1500 bytes
s = "0123456789"
write(s * 200, [])
write(s * 150, [])
write(s * 51, [s * 350]) # 3500 bytes
write("\n", [s * 51]) # 510 bytes
def test_bytes(self):
for stream_name, level in [("stdout", "I"), ("stderr", "W")]:
with self.subTest(stream=stream_name):
stream = getattr(sys, stream_name).buffer
tag = f"python.{stream_name}"
self.assertEqual(f"<BinaryLogStream '{tag}'>", repr(stream))
self.assertTrue(stream.writable())
self.assertFalse(stream.readable())
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.
#
# On newer versions of Android, all three of the above tools (or
# maybe Logcat itself) will also strip any number of leading
# newlines.
write(b"\nx", ["", "x"] if api_level < 30 else ["x"])
write(b"\na\n", ["", "a"] if api_level < 30 else ["a"])
write(b"\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)