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