Coverage for src / rhiza / models / bundle.py: 100%

62 statements  

« prev     ^ index     » next       coverage.py v7.13.5, created at 2026-04-02 07:04 +0000

1"""Bundle models for Rhiza configuration.""" 

2 

3from dataclasses import dataclass, field 

4from typing import Any 

5 

6from rhiza.models._base import YamlSerializable 

7from rhiza.models._git_utils import _normalize_to_list 

8 

9 

10@dataclass(frozen=True, kw_only=True) 

11class BundleDefinition: 

12 """Represents a single bundle from template-bundles.yml. 

13 

14 Attributes: 

15 description: Human-readable description of the bundle. 

16 files: List of file paths included in this bundle. 

17 requires: List of bundle names that this bundle requires. 

18 standalone: Whether this bundle is standalone (no dependencies). 

19 """ 

20 

21 # name: str 

22 description: str 

23 standalone: bool = True 

24 files: list[str] = field(default_factory=list) 

25 requires: list[str] = field(default_factory=list) 

26 

27 

28@dataclass(frozen=True, kw_only=True) 

29class RhizaBundles(YamlSerializable): 

30 """Represents the structure of template-bundles.yml. 

31 

32 Attributes: 

33 version: Optional version string of the bundles configuration format. 

34 bundles: Dictionary mapping bundle names to their definitions. 

35 """ 

36 

37 version: str | None = None 

38 bundles: dict[str, BundleDefinition] = field(default_factory=dict) 

39 

40 @property 

41 def config(self) -> dict[str, Any]: 

42 """Return the bundles' current state as a configuration dictionary.""" 

43 config: dict[str, Any] = {} 

44 

45 if self.version is not None: 

46 config["version"] = self.version 

47 

48 bundles_dict: dict[str, Any] = {} 

49 for name, bundle in self.bundles.items(): 

50 bundle_entry: dict[str, Any] = {"description": bundle.description} 

51 if bundle.files: 

52 bundle_entry["files"] = bundle.files 

53 if bundle.requires: 

54 bundle_entry["requires"] = bundle.requires 

55 if bundle.standalone: 

56 bundle_entry["standalone"] = bundle.standalone 

57 bundles_dict[name] = bundle_entry 

58 

59 config["bundles"] = bundles_dict 

60 return config 

61 

62 @classmethod 

63 def from_config(cls, config: dict[str, Any]) -> "RhizaBundles": 

64 """Create a RhizaBundles instance from a configuration dictionary. 

65 

66 Args: 

67 config: Dictionary containing bundles configuration. 

68 

69 Returns: 

70 A new RhizaBundles instance. 

71 

72 Raises: 

73 TypeError: If bundle data has invalid types. 

74 """ 

75 version = config.get("version") 

76 

77 bundles_config = config.get("bundles", {}) 

78 if not isinstance(bundles_config, dict): 

79 msg = "Bundles must be a dictionary" 

80 raise TypeError(msg) 

81 

82 bundles: dict[str, BundleDefinition] = {} 

83 for bundle_name, bundle_data in bundles_config.items(): 

84 if not isinstance(bundle_data, dict): 

85 msg = f"Bundle '{bundle_name}' must be a dictionary" 

86 raise TypeError(msg) 

87 

88 files = _normalize_to_list(bundle_data.get("files")) 

89 requires = _normalize_to_list(bundle_data.get("requires")) 

90 

91 bundles[bundle_name] = BundleDefinition( 

92 description=bundle_data.get("description", ""), 

93 files=files, 

94 requires=requires, 

95 standalone=bundle_data.get("standalone", True), 

96 ) 

97 

98 return cls(version=version, bundles=bundles) 

99 

100 def resolve_to_paths(self, bundle_names: list[str]) -> list[str]: 

101 """Convert bundle names to deduplicated file paths. 

102 

103 Args: 

104 bundle_names: List of bundle names to resolve. 

105 

106 Returns: 

107 Deduplicated list of file paths from all bundles and their dependencies. 

108 

109 Raises: 

110 ValueError: If a bundle doesn't exist or circular dependency detected. 

111 """ 

112 bundles: set[str] = set() 

113 paths: list[str] = [] 

114 seen: set[str] = set() 

115 

116 for bundle_name in bundle_names: 

117 bundles.add(bundle_name) 

118 for bundle in self.bundles[bundle_name].requires: 

119 bundles.add(bundle) 

120 

121 for bundle_name in bundles: 

122 bundle = self.bundles[bundle_name] 

123 for path in bundle.files: 

124 if path not in seen: 

125 paths.append(path) 

126 seen.add(path) 

127 

128 return paths