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