Coverage for frappe_manager / metadata_manager.py: 50%

1from pathlib import Path 

2 

3import tomlkit 

4from pydantic import BaseModel, EmailStr, Field 

5 

6from frappe_manager import CLI_FM_CONFIG_PATH 

7from frappe_manager.migration_manager.version import Version 

8from frappe_manager.utils.helpers import get_current_fm_version 

9 

10 

11class FMCloudflareConfig(BaseModel): 

12 """Cloudflare DNS API credentials for DNS-01 challenge.""" 

13 

14 email: EmailStr | None = Field(None, description="Cloudflare account email (required for Global API Key).") 

15 api_token: str | None = Field(None, description="Cloudflare API Token (recommended - scoped permissions).") 

16 api_key: str | None = Field(None, description="Cloudflare Global API Key (legacy - full account access).") 

17 

18 @property 

19 def exists(self) -> bool: 

20 """Check if any Cloudflare credentials are configured.""" 

21 return bool(self.api_token or self.api_key) 

22 

23 def get_toml_doc(self): 

24 model_dict = self.model_dump(exclude_none=True) 

25 toml_doc = tomlkit.document() 

26 

27 for key, value in model_dict.items(): 

28 if isinstance(value, Path): 

29 toml_doc[key] = str(value.absolute()) 

30 else: 

31 toml_doc[key] = value 

32 return toml_doc 

33 

34 @classmethod 

35 def import_from_toml_doc(cls, toml_doc): 

36 config_object = cls(**toml_doc) 

37 return config_object 

38 

39 

40class FMLetsencryptConfig(BaseModel): 

41 """Let's Encrypt configuration for certificate registration.""" 

42 

43 email: EmailStr | None = Field( 

44 None, 

45 description="Email for Let's Encrypt certificate registration and notifications.", 

46 ) 

47 

48 @property 

49 def exists(self) -> bool: 

50 """Check if Let's Encrypt email is configured.""" 

51 return bool(self.email and self.email != "dummy@fm.fm") 

52 

53 def get_toml_doc(self): 

54 model_dict = self.model_dump(exclude_none=True) 

55 toml_doc = tomlkit.document() 

56 

57 for key, value in model_dict.items(): 

58 if isinstance(value, Path): 

59 toml_doc[key] = str(value.absolute()) 

60 else: 

61 toml_doc[key] = value 

62 return toml_doc 

63 

64 @classmethod 

65 def import_from_toml_doc(cls, toml_doc): 

66 config_object = cls(**toml_doc) 

67 return config_object 

68 

69 

70class FMValidationConfig(BaseModel): 

71 """Validation settings for Frappe Manager operations.""" 

72 

73 enforce_domain_uniqueness: bool = Field(default=True, description="Enforce domain uniqueness across benches") 

74 

75 def get_toml_doc(self): 

76 model_dict = self.model_dump(exclude_none=True) 

77 toml_doc = tomlkit.document() 

78 for key, value in model_dict.items(): 

79 toml_doc[key] = value 

80 return toml_doc 

81 

82 @classmethod 

83 def import_from_toml_doc(cls, toml_doc): 

84 return cls(**toml_doc) 

85 

86 

87class FMLogsConfig(BaseModel): 

88 """Logging configuration for file and console output.""" 

89 

90 file_level: str = Field( 

91 default="DEBUG", 

92 description="Log level for file logs (DEBUG, INFO, WARNING, ERROR, CRITICAL)", 

93 ) 

94 

95 def get_toml_doc(self): 

96 model_dict = self.model_dump(exclude_none=True) 

97 toml_doc = tomlkit.document() 

98 for key, value in model_dict.items(): 

99 toml_doc[key] = value 

100 return toml_doc 

101 

102 @classmethod 

103 def import_from_toml_doc(cls, toml_doc): 

104 return cls(**toml_doc) 

105 

106 

107class FMConfigManager(BaseModel): 

108 root_path: Path 

109 version: Version 

110 cloudflare: FMCloudflareConfig = Field(default=FMCloudflareConfig()) 

111 ngrok_auth_token: str | None = Field(None, description="Ngrok authentication token") 

112 validation: FMValidationConfig = Field(default=FMValidationConfig()) 

113 logs: FMLogsConfig = Field(default=FMLogsConfig()) 

114 

115 def __init__(self, **data): 

116 super().__init__(**data) 

117 self._raw_config = {} 

118 

119 def get_system_migration_version(self) -> Version: 

120 """Get version system is migrated to.""" 

121 if hasattr(self, "_raw_config") and "migration_state" in self._raw_config: 

122 version_str = self._raw_config["migration_state"].get("system_migrated_to") 

123 if version_str: 

124 return Version(version_str) 

125 return self.version 

126 

127 def set_system_migration_version(self, version: Version) -> None: 

128 """Update system migration version.""" 

129 if not hasattr(self, "_raw_config"): 

130 self._raw_config = {} 

131 

132 if "migration_state" not in self._raw_config: 

133 self._raw_config["migration_state"] = {} 

134 

135 self._raw_config["migration_state"]["system_migrated_to"] = str(version.version) 

136 self.export_to_toml() 

137 

138 def _ensure_migration_state(self) -> None: 

139 """Ensure migration_state exists in config.""" 

140 if not hasattr(self, "_raw_config"): 

141 self._raw_config = {} 

142 

143 if "migration_state" not in self._raw_config: 

144 self._raw_config["migration_state"] = { 

145 "system_migrated_to": str(self.version.version), 

146 } 

147 self.export_to_toml() 

148 

149 _config_data: dict | None = None 

150 

151 def export_to_toml(self, path: Path = CLI_FM_CONFIG_PATH) -> None: 

152 exclude = {"root_path"} 

153 

154 if not self.cloudflare.exists: 

155 exclude.add("cloudflare") 

156 

157 fm_config_dict = self.model_dump(exclude=exclude, exclude_none=True) 

158 

159 fm_config_dict["version"] = self.version.version 

160 

161 if hasattr(self, "_raw_config") and "migration_state" in self._raw_config: 

162 fm_config_dict["migration_state"] = self._raw_config["migration_state"] 

163 

164 toml_doc = tomlkit.document() 

165 

166 for key, value in fm_config_dict.items(): 

167 toml_doc[key] = value 

168 

169 try: 

170 with open(path, "w") as f: 

171 f.write(tomlkit.dumps(toml_doc)) 

172 except Exception as e: 

173 raise RuntimeError(f"Failed to write FM config to {path}: {e}") from e 

174 

175 @classmethod 

176 def import_from_toml(cls, path: Path = CLI_FM_CONFIG_PATH) -> "FMConfigManager": 

177 input_data = {} 

178 

179 input_data["version"] = Version(get_current_fm_version()) 

180 input_data["cloudflare"] = FMCloudflareConfig(email=None, api_key=None, api_token=None) 

181 input_data["root_path"] = str(path) 

182 input_data["ngrok_auth_token"] = None 

183 input_data["validation"] = FMValidationConfig() 

184 input_data["logs"] = FMLogsConfig() 

185 

186 raw_config_data = {} 

187 

188 if path.exists(): 

189 data = tomlkit.parse(path.read_text()) 

190 input_data["version"] = Version(data.get("version", get_current_fm_version())) 

191 

192 if "cloudflare" in data: 

193 input_data["cloudflare"] = FMCloudflareConfig(**data["cloudflare"]) 

194 

195 input_data["ngrok_auth_token"] = data.get("ngrok_auth_token", None) 

196 

197 if "validation" in data: 

198 input_data["validation"] = FMValidationConfig(**data["validation"]) 

199 

200 if "logs" in data: 

201 input_data["logs"] = FMLogsConfig(**data["logs"]) 

202 

203 if "migration_state" in data: 

204 import json 

205 

206 raw_config_data["migration_state"] = json.loads(json.dumps(data["migration_state"])) 

207 

208 fm_config_instance = cls(**input_data) 

209 fm_config_instance._raw_config = raw_config_data 

210 return fm_config_instance