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 checkpoint_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(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.Execute(["root"])
42
43        for prop in ["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.Execute(["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        try:
111          first_api_level = int(out)
112          self.assertTrue(first_api_level < 29 or self.isCheckpoint_,
113                             "User Data Checkpoint is disabled")
114        except:
115          pass
116
117    def testRollback(self):
118        if not self.isCheckpoint_:
119            return
120
121        self.adb.Execute(["root"])
122
123        # Make sure that we are fully booted so we don't get entangled in
124        # someone else's checkpoint
125        self.checkBooted()
126
127        # Create a file and initiate checkpoint
128        self.dut.Execute("setprop persist.vold.dont_commit_checkpoint 1")
129        self.dut.Execute("echo " + self._ORIGINALVALUE + " > " + self._CHECKPOINTTESTFILE)
130        out, err, return_code = self.dut.Execute("vdc checkpoint startCheckpoint 1")
131        self.assertEqual(return_code, 0)
132        self.reboot()
133
134        # Modify the file but do not commit
135        self.dut.Execute("echo " + self._MODIFIEDVALUE + " > " + self._CHECKPOINTTESTFILE)
136        self.reboot()
137
138        # Check the file is unchanged
139        out, err, return_code = self.dut.Execute("cat " + self._CHECKPOINTTESTFILE)
140        self.assertEqual(out.strip(), self._ORIGINALVALUE)
141
142        # Clean up
143        self.dut.Execute("setprop persist.vold.dont_commit_checkpoint 0")
144        out, err, return_code = self.dut.Execute("vdc checkpoint commitChanges")
145        self.assertEqual(return_code, 0)
146        self.reboot()
147        self.dut.Execute("rm " + self._CHECKPOINTTESTFILE)
148
149    def testCommit(self):
150        if not self.isCheckpoint_:
151            return
152
153        self.adb.Execute(["root"])
154
155        # Make sure that we are fully booted so we don't get entangled in
156        # someone else's checkpoint
157        self.checkBooted()
158
159        # Create a file and initiate checkpoint
160        self.dut.Execute("setprop persist.vold.dont_commit_checkpoint 1")
161        self.dut.Execute("echo " + self._ORIGINALVALUE + " > " + self._CHECKPOINTTESTFILE)
162        out, err, return_code = self.dut.Execute("vdc checkpoint startCheckpoint 1")
163        self.assertEqual(return_code, 0)
164        self.reboot()
165
166        # Modify the file and commit the checkpoint
167        self.dut.Execute("echo " + self._MODIFIEDVALUE + " > " + self._CHECKPOINTTESTFILE)
168        self.dut.Execute("setprop persist.vold.dont_commit_checkpoint 0")
169        out, err, return_code = self.dut.Execute("vdc checkpoint commitChanges")
170        self.assertEqual(return_code, 0)
171        self.reboot()
172
173        # Check file has changed
174        out, err, return_code = self.dut.Execute("cat " + self._CHECKPOINTTESTFILE)
175        self.assertEqual(out.strip(), self._MODIFIEDVALUE)
176
177        # Clean up
178        self.dut.Execute("rm " + self._CHECKPOINTTESTFILE)
179
180if __name__ == "__main__":
181    # Setting verbosity is required to generate output that the TradeFed test
182    # runner can parse.
183    unittest.main(verbosity=3)
184