Coverage for app/services/clouds_yaml_parser.py: 81.13%

1"""Parse a raw `clouds.yaml` blob into an `OpenStackCredentialUpsert`. 

2 

3Used by the `/me/openstack-credentials/from-yaml` convenience endpoint so 

4users can paste/upload the file Horizon hands them instead of filling 

5the form manually. The parser is strict: anything weird gets a 422 with 

6a message the UI can render verbatim. 

7""" 

8from __future__ import annotations 

9 

10import yaml 

11from fastapi import HTTPException, status 

12 

13from app.models import OpenStackAuthType 

14from app.schemas import OpenStackCredentialUpsert 

15 

16_SUPPORTED_AUTH_TYPES = { 

17 "v3applicationcredential": OpenStackAuthType.APPLICATION_CREDENTIAL, 

18 "password": OpenStackAuthType.PASSWORD, 

19 # `clouds.yaml` from Horizon often omits `auth_type` for password; 

20 # treat the absence as password. 

21 None: OpenStackAuthType.PASSWORD, 

22} 

23 

24 

25def _bad(msg: str) -> HTTPException: 

26 return HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=msg) 

27 

28 

29def parse(clouds_yaml: str, cloud_name: str | None = None) -> OpenStackCredentialUpsert: 

30 try: 

31 doc = yaml.safe_load(clouds_yaml) 

32 except yaml.YAMLError as e: 

33 raise _bad(f"Invalid YAML: {e}") 

34 

35 if not isinstance(doc, dict) or "clouds" not in doc or not isinstance(doc["clouds"], dict): 

36 raise _bad("clouds.yaml must contain a top-level 'clouds:' mapping") 

37 

38 clouds = doc["clouds"] 

39 if not clouds: 

40 raise _bad("clouds.yaml has no clouds defined") 

41 

42 if cloud_name is None: 

43 if len(clouds) == 1: 

44 cloud_name = next(iter(clouds.keys())) 

45 else: 

46 names = ", ".join(sorted(clouds.keys())) 

47 raise _bad(f"Multiple clouds in YAML; specify cloud_name (one of: {names})") 

48 

49 if cloud_name not in clouds: 

50 names = ", ".join(sorted(clouds.keys())) 

51 raise _bad(f"Cloud '{cloud_name}' not found in YAML (available: {names})") 

52 

53 cloud = clouds[cloud_name] 

54 if not isinstance(cloud, dict): 

55 raise _bad(f"Cloud '{cloud_name}' is not a mapping") 

56 

57 auth = cloud.get("auth") 

58 if not isinstance(auth, dict): 

59 raise _bad(f"Cloud '{cloud_name}' is missing 'auth:' block") 

60 

61 raw_auth_type = cloud.get("auth_type") 

62 if raw_auth_type not in _SUPPORTED_AUTH_TYPES: 

63 raise _bad( 

64 f"Unsupported auth_type '{raw_auth_type}'. " 

65 f"Supported: v3applicationcredential, password" 

66 ) 

67 auth_type = _SUPPORTED_AUTH_TYPES[raw_auth_type] 

68 

69 auth_url = auth.get("auth_url") 

70 if not auth_url: 

71 raise _bad("auth.auth_url is required") 

72 

73 common = { 

74 "auth_url": auth_url, 

75 "region_name": cloud.get("region_name"), 

76 "interface": cloud.get("interface", "public"), 

77 "identity_api_version": str(cloud.get("identity_api_version", "3")), 

78 "project_id": auth.get("project_id"), 

79 "project_name": auth.get("project_name"), 

80 "user_domain_name": auth.get("user_domain_name"), 

81 "project_domain_name": auth.get("project_domain_name"), 

82 } 

83 

84 if auth_type == OpenStackAuthType.APPLICATION_CREDENTIAL: 

85 identifier = auth.get("application_credential_id") 

86 secret = auth.get("application_credential_secret") 

87 if not identifier or not secret: 

88 raise _bad( 

89 "Application credential requires application_credential_id " 

90 "and application_credential_secret" 

91 ) 

92 else: 

93 identifier = auth.get("username") 

94 secret = auth.get("password") 

95 if not identifier or not secret: 

96 raise _bad("Password auth requires auth.username and auth.password") 

97 

98 try: 

99 return OpenStackCredentialUpsert( 

100 auth_type=auth_type, 

101 identifier=identifier, 

102 secret=secret, 

103 **common, 

104 ) 

105 except ValueError as e: 

106 raise _bad(str(e))