1#!/usr/bin/env python
2#
3# Copyright (C) 2018 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
18from vts.runners.host import asserts
19from vts.runners.host import base_test
20from vts.runners.host import const
21from vts.runners.host import test_runner
22
23import time
24
25class VtsKernelCheckpointTest(base_test.BaseTestClass):
26
27    _CHECKPOINTTESTFILE = "/data/local/tmp/checkpointtest"
28    _ORIGINALVALUE = "original value"
29    _MODIFIEDVALUE = "modified value"
30
31    def getFstab(self):
32        self.dut.adb.root()
33
34        for prop in ["fstab_suffix", "hardware", "hardware.platform"]:
35            extension = self.dut.adb.shell("getprop ro.boot." + prop, no_except = True)
36            extension = extension[const.STDOUT]
37            if not extension:
38                continue
39
40            for filename in ["/odm/etc/fstab.", "/vendor/etc/fstab.", "/fstab."]:
41                contents = self.dut.adb.shell("cat " + filename + extension, no_except = True)
42                if contents[const.EXIT_CODE] != 0:
43                    continue
44
45                return contents[const.STDOUT]
46
47        return ""
48
49    def isCheckpoint(self):
50        fstab = self.getFstab().splitlines()
51        for line in fstab:
52            parts = line.split()
53            if len(parts) != 5: # fstab has five parts for each entry:
54                                # [device-name] [mount-point] [type] [mount_flags] [fsmgr_flags]
55                continue
56
57            flags = parts[4]    # the fsmgr_flags field is the fifth one, thus index 4
58            flags = flags.split(',')
59            if any(flag.startswith("checkpoint=") for flag in flags):
60                return True
61
62        return False
63
64    def setUpClass(self):
65        self.dut = self.android_devices[0]
66        self.shell = self.dut.shell
67        self.isCheckpoint_ = self.isCheckpoint()
68
69    def reboot(self):
70        self.dut.adb.reboot()
71        try:
72          self.dut.adb.wait_for_device(timeout=900)
73        except self.dut.adb.AdbError as e:
74          asserts.fail("Exception thrown waiting for device:" + e.msg())
75
76        # Should not be necessary, but without these retries, test fails
77        # regularly on taimen with Android Q
78        for i in range(1, 30):
79          try:
80            self.dut.adb.root()
81            break
82          except:
83            time.sleep(1)
84
85        for i in range(1, 30):
86          try:
87            self.dut.adb.shell("ls");
88            break
89          except:
90            time.sleep(1)
91
92    def checkBooted(self):
93        for i in range(1, 900):
94          result = self.dut.adb.shell("getprop sys.boot_completed")
95          try:
96            boot_completed = int(result)
97            asserts.assertEqual(boot_completed, 1)
98            return
99          except:
100            time.sleep(1)
101
102        asserts.fail("sys.boot_completed not set")
103
104    def testCheckpointEnabled(self):
105        result = self.dut.adb.shell("getprop ro.product.first_api_level")
106        try:
107          first_api_level = int(result)
108          asserts.assertTrue(first_api_level < 29 or self.isCheckpoint_,
109                             "User Data Checkpoint is disabled")
110        except:
111          pass
112
113    def testRollback(self):
114        if not self.isCheckpoint_:
115            return
116
117        self.dut.adb.root()
118
119        # Make sure that we are fully booted so we don't get entangled in
120        # someone else's checkpoint
121        self.checkBooted()
122
123        # Create a file and initiate checkpoint
124        self.dut.adb.shell("setprop persist.vold.dont_commit_checkpoint 1")
125        self.dut.adb.shell("echo " + self._ORIGINALVALUE + " > " + self._CHECKPOINTTESTFILE)
126        result = self.dut.adb.shell("vdc checkpoint startCheckpoint 1", no_except = True)
127        asserts.assertEqual(result[const.EXIT_CODE], 0)
128        self.reboot()
129
130        # Modify the file but do not commit
131        self.dut.adb.shell("echo " + self._MODIFIEDVALUE + " > " + self._CHECKPOINTTESTFILE)
132        self.reboot()
133
134        # Check the file is unchanged
135        result = self.dut.adb.shell("cat " + self._CHECKPOINTTESTFILE)
136        asserts.assertEqual(result.strip(), self._ORIGINALVALUE)
137
138        # Clean up
139        self.dut.adb.shell("setprop persist.vold.dont_commit_checkpoint 0")
140        result = self.dut.adb.shell("vdc checkpoint commitChanges", no_except = True)
141        asserts.assertEqual(result[const.EXIT_CODE], 0)
142        self.reboot()
143        self.dut.adb.shell("rm " + self._CHECKPOINTTESTFILE)
144
145    def testCommit(self):
146        if not self.isCheckpoint_:
147            return
148
149        self.dut.adb.root()
150
151        # Make sure that we are fully booted so we don't get entangled in
152        # someone else's checkpoint
153        self.checkBooted()
154
155        # Create a file and initiate checkpoint
156        self.dut.adb.shell("setprop persist.vold.dont_commit_checkpoint 1")
157        self.dut.adb.shell("echo " + self._ORIGINALVALUE + " > " + self._CHECKPOINTTESTFILE)
158        result = self.dut.adb.shell("vdc checkpoint startCheckpoint 1", no_except = True)
159        asserts.assertEqual(result[const.EXIT_CODE], 0)
160        self.reboot()
161
162        # Modify the file and commit the checkpoint
163        self.dut.adb.shell("echo " + self._MODIFIEDVALUE + " > " + self._CHECKPOINTTESTFILE)
164        self.dut.adb.shell("setprop persist.vold.dont_commit_checkpoint 0")
165        result = self.dut.adb.shell("vdc checkpoint commitChanges", no_except = True)
166        asserts.assertEqual(result[const.EXIT_CODE], 0)
167        self.reboot()
168
169        # Check file has changed
170        result = self.dut.adb.shell("cat " + self._CHECKPOINTTESTFILE)
171        asserts.assertEqual(result.strip(), self._MODIFIEDVALUE)
172
173        # Clean up
174        self.dut.adb.shell("rm " + self._CHECKPOINTTESTFILE)
175
176if __name__ == "__main__":
177    test_runner.main()
178