Coverage for opt/mealie/lib/python3.12/site-packages/mealie/services/backups_v2/backup_file.py: 29%

1import json 

2import shutil 

3import tempfile 

4from pathlib import Path 

5 

6 

7class BackupContents: 

8 _tables: dict | None = None 

9 

10 def __init__(self, file: Path) -> None: 

11 self.base = self._find_base(file) 

12 self.data_directory = self._find_data_dir_from_base(self.base) 

13 self.tables = self._find_database_from_base(self.base) 

14 

15 @classmethod 

16 def _find_base(cls, file: Path) -> Path: 

17 # Safari mangles our ZIP structure and adds a "__MACOSX" directory at the root along with 

18 # an arbitrarily-named directory containing the actual contents. So, if we find a dunder directory 

19 # at the root (i.e. __MACOSX) we traverse down the first non-dunder directory and assume this is the base. 

20 # This works because our backups never contain a directory that starts with "__". 

21 dirs = [d for d in file.iterdir() if d.is_dir()] 

22 dunder_dirs = [d for d in dirs if d.name.startswith("__")] 

23 normal_dirs = [d for d in dirs if not d.name.startswith("__")] 

24 

25 if not dunder_dirs: 

26 return file 

27 

28 # If the backup somehow adds a __MACOSX directory alongside the data directory, rather than in the 

29 # parent directory, we don't want to traverse down. We check for our database.json file, and if it exists, 

30 # we're already at the correct base. 

31 if cls._find_database_from_base(file).exists(): 

32 return file 

33 

34 # This ZIP file was mangled, so we return the first non-dunder directory (if it exists). 

35 return normal_dirs[0] if normal_dirs else file 

36 

37 @classmethod 

38 def _find_data_dir_from_base(cls, base: Path) -> Path: 

39 return base / "data" 

40 

41 @classmethod 

42 def _find_database_from_base(cls, base: Path) -> Path: 

43 return base / "database.json" 

44 

45 def validate(self) -> bool: 

46 if not self.base.is_dir(): 

47 return False 

48 

49 if not self.data_directory.is_dir(): 

50 return False 

51 

52 if not self.tables.is_file(): 

53 return False 

54 

55 return True 

56 

57 def schema_version(self) -> str: 

58 tables = self.read_tables() 

59 

60 alembic_version = tables.get("alembic_version", []) 

61 

62 if not alembic_version: 

63 return "" 

64 

65 return alembic_version[0].get("version_num", "") 

66 

67 def read_tables(self) -> dict: 

68 if self._tables is None: 

69 with open(self.tables) as f: 

70 self._tables = json.load(f) 

71 

72 return self._tables 

73 

74 

75class BackupFile: 

76 temp_dir: Path | None = None 

77 

78 def __init__(self, file: Path) -> None: 

79 self.zip = file 

80 

81 def __enter__(self) -> BackupContents: 

82 self.temp_dir = Path(tempfile.mkdtemp()) 

83 shutil.unpack_archive(str(self.zip), str(self.temp_dir)) 

84 return BackupContents(self.temp_dir) 

85 

86 def __exit__(self, exc_type, exc_val, exc_tb): 

87 if self.temp_dir and self.temp_dir.is_dir(): 

88 shutil.rmtree(self.temp_dir) 

89 

90 self.temp_dir = None