1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
|
#!/usr/bin/env python3
#
# Copyright (C) 2023 The Android Open Source Project
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import re
import subprocess
import unittest
KNOWN_NON_LOGGING_SERVICES = [
"vendor.ir-default",
"SELF_TEST_SERVICE_DOES_NOT_EXIST",
]
KNOWN_LOGGING_SERVICES = [
"zygote",
# b/210919187 - main log is too busy, gets dropped off
# "statsd",
# "vendor.audio-hal-aidl",
"SELF_TEST_SERVICE_DOES_NOT_EXIST",
]
def device_log(log):
ret = subprocess.check_output(["adb", "shell", "log", "-t", "logd_integration_test", log]).decode()
assert len(ret) == 0, f"Expected no output, but found '{ret}'"
def get_service_pid(svc):
return int(subprocess.check_output(["adb", "shell", "getprop", "init.svc_debug_pid." + svc]))
def get_pid_logs(pid):
return subprocess.check_output(["adb", "logcat", "--pid", str(pid), "-d"]).decode()
def get_product_name():
return subprocess.check_output(["adb", "shell", "getprop", "ro.product.name"]).decode()
def iter_service_pids(test_case, services):
a_service_worked = False
for service in services:
try:
yield service, get_service_pid(service)
a_service_worked = True
except subprocess.CalledProcessError:
continue
except ValueError:
continue
test_case.assertTrue(a_service_worked)
def get_dropped_logs(test_case, buffer):
output = subprocess.check_output(["adb", "logcat", "-b", buffer, "--statistics"]).decode()
lines = iter(output.split("\n"))
res = []
# Search for these lines, in order. Consider output:
# :) adb logcat -b system -S | grep -E "Total|Now"
# size/num system Total
# Total 883973/6792 883973/6792
# Now 883973/6792 883973/6792
for indication in ["Total", "Now"]:
reLineCount = re.compile(f"^{indication}.*\s+[0-9]+/([0-9]+)")
while True:
line = next(lines)
match = reLineCount.match(line)
if match:
res.append(int(match.group(1)))
break
total, now = res
return total, now, output
class LogdIntegrationTest(unittest.TestCase):
def subTest(self, subtest_name):
"""install logger for all subtests"""
class SubTestLogger:
def __init__(self, testcase, subtest_name):
self.subtest_name = subtest_name
self.subtest = testcase.subTest(subtest_name)
def __enter__(self):
device_log(f"Starting subtest {subtest_name}")
return self.subtest.__enter__()
def __exit__(self, *args):
device_log(f"Ending subtest {subtest_name}")
return self.subtest.__exit__(*args)
return SubTestLogger(super(), subtest_name)
def test_no_logs(self):
for service, pid in iter_service_pids(self, KNOWN_NON_LOGGING_SERVICES):
with self.subTest(service + "_no_logs"):
lines = get_pid_logs(pid)
self.assertFalse("\n" in lines, f"{service} ({pid}) shouldn't have logs, but found: {lines}")
def test_has_logs(self):
for service, pid in iter_service_pids(self, KNOWN_LOGGING_SERVICES):
with self.subTest(service + "_has_logs"):
lines = get_pid_logs(pid)
self.assertTrue("\n" in lines, f"{service} ({pid}) should have logs, but found: {lines}")
def test_no_dropped_logs(self):
dropped_buffer_allowed = {
"crash": 0,
"kernel": 0,
"main": 4000,
"system": 0 if get_product_name().startswith("aosp") else 10000,
}
for buffer, allowed in dropped_buffer_allowed.items():
with self.subTest(buffer + "_buffer_not_dropped"):
total, now, output = get_dropped_logs(self, buffer)
dropped = total - now
self.assertLessEqual(dropped, allowed,
f"Buffer {buffer} has {dropped} dropped logs (now {now} out of {total} total logs), but expecting <= {allowed}. {output}")
def main():
unittest.main(verbosity=3)
if __name__ == "__main__":
main()
|
