(cls, data: Dict[str, Any])
| 33 | |
| 34 | @classmethod |
| 35 | def from_dict(cls, data: Dict[str, Any]) -> CephadmConfig: |
| 36 | for key in REQUIRED_CEPHADM_KEYS: |
| 37 | if key not in data: |
| 38 | raise ValueError(f"Missing required cephadm config key: {key}") |
| 39 | # Normalize key with dot to attribute name |
| 40 | node_proxy_config_path = data.get("node_proxy_config") or os.environ.get( |
| 41 | "NODE_PROXY_CONFIG", "/etc/ceph/node-proxy.yml" |
| 42 | ) |
| 43 | assert node_proxy_config_path is not None |
| 44 | return cls( |
| 45 | target_ip=data["target_ip"], |
| 46 | target_port=data["target_port"], |
| 47 | keyring=data["keyring"], |
| 48 | root_cert_pem=data["root_cert.pem"], |
| 49 | listener_crt=data["listener.crt"], |
| 50 | listener_key=data["listener.key"], |
| 51 | name=data["name"], |
| 52 | node_proxy_config_path=node_proxy_config_path, |
| 53 | ) |
| 54 | |
| 55 | |
| 56 | def load_cephadm_config(path: str) -> CephadmConfig: |
no test coverage detected