Coverage for frappe_manager / site_manager / domain_conflict.py: 48%

1from pathlib import Path 

2 

3from frappe_manager import CLI_BENCHES_DIRECTORY 

4from frappe_manager.site_manager.bench_config import BenchConfig 

5 

6 

7class DomainConflict: 

8 def __init__(self, domain: str, owner_bench: str, is_primary: bool = False): 

9 self.domain = domain 

10 self.owner_bench = owner_bench 

11 self.is_primary = is_primary 

12 

13 def __str__(self): 

14 type_str = "primary domain" if self.is_primary else "alias" 

15 return f"'{self.domain}' → already used as {type_str} by bench '{self.owner_bench}'" 

16 

17 

18class DomainConflictError(Exception): 

19 def __init__(self, conflicts: list[DomainConflict]): 

20 self.conflicts = conflicts 

21 conflict_msgs = "\n - ".join(str(c) for c in conflicts) 

22 super().__init__(f"Domain conflicts detected:\n - {conflict_msgs}") 

23 

24 

25def build_global_domain_map( 

26 benches_root: Path = CLI_BENCHES_DIRECTORY, 

27 exclude_bench: str | None = None, 

28) -> dict[str, tuple[str, bool]]: 

29 domain_map = {} 

30 

31 if not benches_root.exists(): 

32 return domain_map 

33 

34 for bench_dir in benches_root.iterdir(): 

35 if not bench_dir.is_dir(): 

36 continue 

37 

38 if exclude_bench and bench_dir.name == exclude_bench: 

39 continue 

40 

41 config_file = bench_dir / "bench_config.toml" 

42 if not config_file.exists(): 

43 continue 

44 

45 try: 

46 config = BenchConfig.import_from_toml(config_file) 

47 bench_name = config.name 

48 

49 domain_map[config.name.lower()] = (bench_name, True) 

50 

51 for alias in config.alias_domains or []: 

52 domain_map[alias.lower()] = (bench_name, False) 

53 

54 except Exception: 

55 continue 

56 

57 return domain_map 

58 

59 

60def check_domain_conflicts( 

61 candidate_domains: set[str] | list[str], 

62 benches_root: Path = CLI_BENCHES_DIRECTORY, 

63 exclude_bench: str | None = None, 

64) -> list[DomainConflict]: 

65 domain_map = build_global_domain_map(benches_root, exclude_bench) 

66 

67 conflicts = [] 

68 for domain in candidate_domains: 

69 normalized = domain.lower() 

70 if normalized in domain_map: 

71 owner_bench, is_primary = domain_map[normalized] 

72 conflicts.append(DomainConflict(domain, owner_bench, is_primary)) 

73 

74 return conflicts 

75 

76 

77def validate_domains_unique( 

78 candidate_domains: set[str] | list[str], 

79 benches_root: Path = CLI_BENCHES_DIRECTORY, 

80 exclude_bench: str | None = None, 

81 skip_check: bool = False, 

82) -> None: 

83 if skip_check: 

84 return 

85 

86 conflicts = check_domain_conflicts(candidate_domains, benches_root, exclude_bench) 

87 

88 if conflicts: 

89 raise DomainConflictError(conflicts)