1#!/usr/bin/env python3 2# 3# Copyright 2024, 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"""Unit test for the Snapshot module.""" 18 19import json 20import os 21import pathlib 22import shutil 23import tempfile 24import unittest 25 26from atest.integration_tests.snapshot import Snapshot 27# Disable pylint here because it conflicts with google format 28# pylint: disable=wrong-import-order 29from pyfakefs import fake_filesystem_unittest 30 31 32class SnapshotTest(fake_filesystem_unittest.TestCase): 33 """Snapshot test class unit test.""" 34 35 def setUp(self): 36 self.setUpPyfakefs() 37 self.temp_dir = pathlib.Path(tempfile.mkdtemp()) 38 super().setUp() 39 40 def tearDown(self): 41 shutil.rmtree(self.temp_dir) 42 super().tearDown() 43 44 def test_restore_dir_and_env_and_obj_from_snapshot(self): 45 """Test take and restore directory, environ, and objects together.""" 46 workspace = self.temp_dir / 'workspace' 47 restore_dst_workspace = self.temp_dir / 'workspace2' 48 self.fs.create_dir(workspace) 49 self.fs.create_dir(workspace.joinpath('dir1')) 50 self.fs.create_file(workspace.joinpath('dir1', 'file1'), contents='test') 51 env_key = 'env_name' 52 env_val = 'value' 53 obj_key = 'obj_name' 54 obj_val = 123 55 os.environ[env_key] = env_val 56 snapshot = Snapshot(self.temp_dir / 'db') 57 snapshot_name = 'a_snapshot_name' 58 snapshot.take_snapshot( 59 snapshot_name, 60 workspace, 61 ['*'], 62 env_keys=[env_key], 63 objs={obj_key: obj_val}, 64 ) 65 66 environ, objs = snapshot.restore_snapshot( 67 snapshot_name, restore_dst_workspace.as_posix() 68 ) 69 70 self.assertTrue(restore_dst_workspace.joinpath('dir1', 'file1').exists()) 71 self.assertEqual(environ[env_key], env_val) 72 self.assertEqual(objs[obj_key], obj_val) 73 74 def test_restore_objects_from_snapshot(self): 75 """Test objects is restored from a snapshot.""" 76 workspace = self.temp_dir / 'workspace' 77 self.fs.create_dir(workspace) 78 snapshot = Snapshot(self.temp_dir / 'db') 79 snapshot_name = 'a_snapshot_name' 80 objs = {'a': {'b': False}} 81 snapshot.take_snapshot('a_snapshot_name', workspace, ['*'], objs=objs) 82 83 _, actual_objs = snapshot.restore_snapshot( 84 snapshot_name, workspace.as_posix() 85 ) 86 87 self.assertEqual(objs, actual_objs) 88 89 def test_take_snapshot_empty_dir_snapshot_is_created(self): 90 """Test taking a snapshot of an empty directory.""" 91 workspace = self.temp_dir / 'workspace' 92 self.fs.create_dir(workspace) 93 snapshot = Snapshot(self.temp_dir / 'db') 94 snapshot_name = 'a_snapshot_name' 95 96 snapshot.take_snapshot(snapshot_name, workspace, ['*']) 97 98 self.assertTrue( 99 self.temp_dir.joinpath(f'db/{snapshot_name}_metadata.json').exists() 100 ) 101 102 def test_take_snapshot_with_files_files_are_recorded(self): 103 """Test taking a snapshot of a directory with files.""" 104 workspace = self.temp_dir / 'workspace' 105 self.fs.create_dir(workspace) 106 self.fs.create_dir(workspace.joinpath('dir1')) 107 self.fs.create_file(workspace.joinpath('dir1', 'file1'), contents='test') 108 snapshot = Snapshot(self.temp_dir / 'db') 109 snapshot_name = 'a_snapshot_name' 110 111 snapshot.take_snapshot('a_snapshot_name', workspace, ['*']) 112 113 with open( 114 self.temp_dir / 'db' / f'{snapshot_name}_metadata.json', 115 encoding='utf-8', 116 ) as f: 117 file_infos = json.load(f) 118 self.assertEqual(len(file_infos), 2) 119 self.assertIn('dir1/file1', file_infos) 120 self.assertIn('dir1', file_infos) 121 122 def test_take_snapshot_with_include_paths(self): 123 """Test taking a snapshot with include paths.""" 124 workspace = self.temp_dir / 'workspace' 125 self.fs.create_dir(workspace) 126 self.fs.create_dir(workspace.joinpath('dir1')) 127 self.fs.create_file(workspace.joinpath('dir1', 'file1'), contents='test') 128 self.fs.create_dir(workspace.joinpath('dir2')) 129 self.fs.create_file(workspace.joinpath('dir2', 'file2'), contents='test') 130 snapshot = Snapshot(self.temp_dir / 'db') 131 snapshot_name = 'a_snapshot_name' 132 133 snapshot.take_snapshot('a_snapshot_name', workspace, include_paths=['dir2']) 134 135 with open( 136 self.temp_dir / 'db' / f'{snapshot_name}_metadata.json', 137 encoding='utf-8', 138 ) as f: 139 file_infos = json.load(f) 140 self.assertEqual(len(file_infos), 2) 141 self.assertIn('dir2', file_infos) 142 self.assertIn('dir2/file2', file_infos) 143 144 def test_take_snapshot_with_exclude_paths(self): 145 """Test taking a snapshot with exclude paths.""" 146 workspace = self.temp_dir / 'workspace' 147 self.fs.create_dir(workspace) 148 self.fs.create_dir(workspace.joinpath('dir1')) 149 self.fs.create_file(workspace.joinpath('dir1', 'file1'), contents='test') 150 self.fs.create_dir(workspace.joinpath('dir2')) 151 self.fs.create_file(workspace.joinpath('dir2', 'file2'), contents='test') 152 snapshot = Snapshot(self.temp_dir / 'db') 153 snapshot_name = 'a_snapshot_name' 154 155 snapshot.take_snapshot( 156 'a_snapshot_name', workspace, ['*'], exclude_paths=['dir1'] 157 ) 158 159 with open( 160 self.temp_dir / 'db' / f'{snapshot_name}_metadata.json', 161 encoding='utf-8', 162 ) as f: 163 file_infos = json.load(f) 164 self.assertEqual(len(file_infos), 2) 165 self.assertIn('dir2', file_infos) 166 self.assertIn('dir2/file2', file_infos) 167 168 def test_take_snapshot_with_include_and_exclude_paths(self): 169 """Test taking a snapshot with include and exclude paths.""" 170 workspace = self.temp_dir / 'workspace' 171 self.fs.create_dir(workspace) 172 self.fs.create_dir(workspace.joinpath('dir1')) 173 self.fs.create_file(workspace.joinpath('dir1', 'file1'), contents='test') 174 self.fs.create_dir(workspace.joinpath('dir2')) 175 self.fs.create_file(workspace.joinpath('dir2', 'file2'), contents='test') 176 snapshot = Snapshot(self.temp_dir / 'db') 177 snapshot_name = 'a_snapshot_name' 178 179 snapshot.take_snapshot( 180 snapshot_name, 181 workspace, 182 include_paths=['dir1'], 183 exclude_paths=['dir1/file1'], 184 ) 185 186 with open( 187 self.temp_dir / 'db' / f'{snapshot_name}_metadata.json', 188 encoding='utf-8', 189 ) as f: 190 file_infos = json.load(f) 191 self.assertEqual(len(file_infos), 1) 192 self.assertIn('dir1', file_infos) 193 194 def test_restore_snapshot_preserve_dangling_relative_links(self): 195 """Test restoring a dangling link relative to the workspace.""" 196 workspace = self.temp_dir / 'workspace' 197 self.fs.create_dir(workspace) 198 snapshot = Snapshot(self.temp_dir / 'db') 199 snapshot_name = 'a_snapshot_name' 200 restore_dir = self.temp_dir / 'restore' 201 link_file_name = 'link' 202 target_file_name = pathlib.Path('non-existent-path') 203 workspace.joinpath(link_file_name).symlink_to(target_file_name) 204 snapshot.take_snapshot(snapshot_name, workspace, ['*']) 205 206 snapshot.restore_snapshot(snapshot_name, restore_dir) 207 208 self.assertEqual( 209 restore_dir.joinpath(link_file_name).readlink(), 210 target_file_name, 211 ) 212 213 def test_restore_snapshot_preserve_dangling_absolute_links(self): 214 """Test restoring a dangling absolute link within workspace.""" 215 workspace = self.temp_dir / 'workspace' 216 self.fs.create_dir(workspace) 217 snapshot = Snapshot(self.temp_dir / 'db') 218 snapshot_name = 'a_snapshot_name' 219 restore_dir = self.temp_dir / 'restore' 220 link_file_name = 'link' 221 target_file_name = 'non-existent-path' 222 internal_dangling_target = workspace / target_file_name 223 workspace.joinpath(link_file_name).symlink_to(internal_dangling_target) 224 snapshot.take_snapshot(snapshot_name, workspace, ['*']) 225 226 snapshot.restore_snapshot(snapshot_name, restore_dir) 227 228 self.assertEqual( 229 restore_dir.joinpath(link_file_name).resolve(), 230 restore_dir / target_file_name, 231 ) 232 233 def test_restore_snapshot_preserve_dangling_external_links(self): 234 """Test restoring a dangling link pointing to outside the workspace.""" 235 workspace = self.temp_dir / 'workspace' 236 self.fs.create_dir(workspace) 237 snapshot = Snapshot(self.temp_dir / 'db') 238 snapshot_name = 'a_snapshot_name' 239 restore_dir = self.temp_dir / 'restore' 240 link_file_name = 'link' 241 external_dangling_target = pathlib.Path('/external/non-existent-path') 242 workspace.joinpath(link_file_name).symlink_to(external_dangling_target) 243 snapshot.take_snapshot(snapshot_name, workspace, ['*']) 244 245 snapshot.restore_snapshot(snapshot_name, restore_dir) 246 247 self.assertEqual( 248 restore_dir.joinpath(link_file_name).resolve(), external_dangling_target 249 ) 250 251 def test_restore_snapshot_empty_dir(self): 252 """Test restoring a snapshot of an empty directory.""" 253 workspace = self.temp_dir / 'workspace' 254 self.fs.create_dir(workspace) 255 self.fs.create_dir(workspace.joinpath('dir1')) 256 snapshot = Snapshot(self.temp_dir / 'db') 257 snapshot_name = 'a_snapshot_name' 258 snapshot.take_snapshot(snapshot_name, workspace, ['*']) 259 restore_dir = self.temp_dir / 'restore' 260 261 snapshot.restore_snapshot(snapshot_name, restore_dir) 262 263 self.assertTrue(restore_dir.joinpath('dir1').exists()) 264 265 def test_restore_snapshot_with_deleted_files(self): 266 """Test restoring a snapshot with deleted files.""" 267 workspace = self.temp_dir / 'workspace' 268 self.fs.create_dir(workspace) 269 self.fs.create_dir(workspace.joinpath('dir1')) 270 self.fs.create_file(workspace.joinpath('dir1', 'file1'), contents='test') 271 snapshot = Snapshot(self.temp_dir / 'db') 272 snapshot_name = 'a_snapshot_name' 273 snapshot.take_snapshot(snapshot_name, workspace, ['*']) 274 self.fs.remove(workspace.joinpath('dir1', 'file1')) 275 276 snapshot.restore_snapshot(snapshot_name, workspace.as_posix()) 277 278 self.assertTrue(workspace.joinpath('dir1', 'file1').exists()) 279 280 def test_restore_snapshot_with_extra_files(self): 281 """Test restoring a snapshot with extra files.""" 282 workspace = self.temp_dir / 'workspace' 283 self.fs.create_dir(workspace) 284 self.fs.create_dir(workspace.joinpath('dir1')) 285 self.fs.create_file(workspace.joinpath('dir1', 'file1'), contents='test') 286 snapshot = Snapshot(self.temp_dir / 'db') 287 snapshot_name = 'a_snapshot_name' 288 snapshot.take_snapshot('a_snapshot_name', workspace, ['*']) 289 self.fs.create_file(workspace.joinpath('dir1', 'file2'), contents='test') 290 291 snapshot.restore_snapshot(snapshot_name, workspace.as_posix()) 292 293 self.assertTrue(workspace.joinpath('dir1', 'file1').exists()) 294 self.assertFalse(workspace.joinpath('dir1', 'file2').exists()) 295 296 def test_restore_snapshot_with_modified_files(self): 297 """Test restoring a snapshot with modified files.""" 298 workspace = self.temp_dir / 'workspace' 299 self.fs.create_dir(workspace) 300 self.fs.create_dir(workspace.joinpath('dir1')) 301 file_path = workspace.joinpath('dir1', 'file1') 302 self.fs.create_file(file_path, contents='test1') 303 snapshot = Snapshot(self.temp_dir / 'db') 304 snapshot_name = 'a_snapshot_name' 305 snapshot.take_snapshot(snapshot_name, workspace, ['*']) 306 file_path.write_text('test2', encoding='utf-8') 307 # Increment file's modified time by 10 milliseconds 308 mtime = os.path.getmtime(file_path) 309 os.utime(file_path, (mtime, mtime + 0.01)) 310 311 snapshot.restore_snapshot(snapshot_name, workspace.as_posix()) 312 313 self.assertEqual( 314 workspace.joinpath('dir1', 'file1').read_text('utf-8'), 'test1' 315 ) 316 317 318if __name__ == '__main__': 319 unittest.main() 320