Coverage for gco/services/aws_ssm.py: 100%

1""" 

2SSM Parameter Store helpers shared across ``cli/``, ``mcp/`` and ``gco/services/``. 

3 

4Four free functions cover every shape callers in the tree currently 

5reach for: 

6 

7* :func:`get_ssm_parameter` — fetch a value, propagate errors verbatim. 

8 Use when the parameter is required and a missing parameter is a hard 

9 failure. The :class:`botocore.exceptions.ClientError` raised by the 

10 underlying ``ssm:GetParameter`` carries a ``ParameterNotFound`` code 

11 that callers can match if they need to map missing into a friendlier 

12 domain error. 

13* :func:`get_ssm_parameter_optional` — fetch a value, return ``None`` on 

14 the specific ``ParameterNotFound`` case while still propagating any 

15 other error (permission denied, throttled, etc.). Use when a missing 

16 parameter is a non-fatal "absent" signal but real errors should 

17 surface. 

18* :func:`check_ssm_parameter` — return ``(True, "")`` iff the parameter 

19 exists, or ``(False, str(exc))`` on any error including 

20 ``ParameterNotFound``. Use for diagnostics-style "is this thing here?" 

21 checks where a transport error is treated the same as absent. 

22* :func:`put_ssm_parameter` — write a parameter. Errors propagate 

23 verbatim. Use for the rare cases (e.g. ALB hostname drift correction 

24 in ``gco/services/health_monitor.py``) where the CLI / monitor needs 

25 to write back to SSM. 

26 

27Architectural rationale. ``mcp/`` is forbidden from importing ``cli/`` 

28directly (the runtime tool surface shells out via subprocess instead), 

29but ``mcp/`` already imports ``gco/services/...`` for shared service 

30helpers. Putting these helpers under ``gco/services/`` lets every 

31concrete callsite — the CLI's :class:`cli.models.ModelManager`, the 

32analytics helpers in :mod:`cli.analytics_user_mgmt`, the 

33``HealthMonitor`` in :mod:`gco.services.health_monitor`, and the 

34:class:`mcp.mission.state.DynamoDBBackend` — share one implementation 

35without re-introducing the ``mcp -> cli`` import edge. 

36 

37Each function lazy-imports ``boto3`` so the helper module's import 

38surface stays free of SDK dependencies; tests that don't exercise SSM 

39can monkeypatch ``boto3.client`` without dragging the whole CLI in. 

40The same lazy-import pattern is already used in 

41:func:`cli.analytics_user_mgmt.check_ssm_parameter` and the 

42:class:`mcp.mission.state.DynamoDBBackend` — this module just 

43consolidates it. 

44""" 

45 

46from __future__ import annotations 

47 

48from typing import Any 

49 

50__all__ = [ 

51 "check_ssm_parameter", 

52 "get_ssm_parameter", 

53 "get_ssm_parameter_optional", 

54 "put_ssm_parameter", 

55] 

56 

57 

58def _ssm_client(region: str | None) -> Any: 

59 """Construct a fresh ``ssm`` boto3 client. 

60 

61 ``boto3`` is imported inside the function so a caller that monkey- 

62 patches ``boto3.client`` for a test never has to also intercept a 

63 module-level cached client. The fresh client is also free of 

64 cross-test leakage: a permission-denied response in one test does 

65 not poison the credential cache for the next. 

66 """ 

67 import boto3 

68 

69 if region is None: 

70 return boto3.client("ssm") 

71 return boto3.client("ssm", region_name=region) 

72 

73 

74def get_ssm_parameter(name: str, *, region: str | None = None) -> str: 

75 """Fetch an SSM parameter value; propagate errors verbatim. 

76 

77 Args: 

78 name: Fully-qualified parameter name (e.g. ``"/gco/foo"``). 

79 region: Optional AWS region. ``None`` lets boto3's default 

80 chain (env var, config file, instance metadata) decide. 

81 

82 Returns: 

83 The parameter's ``Value`` field as a string. 

84 

85 Raises: 

86 botocore.exceptions.ClientError: ``ParameterNotFound`` when the 

87 parameter is missing, plus any other boto3 client error 

88 (throttled, access denied, etc.). 

89 botocore.exceptions.BotoCoreError: For transport / credential 

90 failures. 

91 

92 The function does not catch any exception — errors carry the 

93 underlying ``Code`` field that callers wanting domain-specific 

94 messages can match on. 

95 """ 

96 response = _ssm_client(region).get_parameter(Name=name) 

97 return str(response["Parameter"]["Value"]) 

98 

99 

100def get_ssm_parameter_optional(name: str, *, region: str | None = None) -> str | None: 

101 """Fetch an SSM parameter value or return ``None`` if absent. 

102 

103 Distinguishes the ``ParameterNotFound`` case (returns ``None``) 

104 from every other error (re-raised verbatim). This matches the 

105 pattern :class:`gco.services.health_monitor.HealthMonitor` reaches 

106 for when it has to read-then-maybe-write a drift-tracker 

107 parameter — a missing value is not an error, but a permission 

108 denied or a throttle is. 

109 

110 Args: 

111 name: Fully-qualified parameter name. 

112 region: Optional AWS region. 

113 

114 Returns: 

115 The parameter's ``Value`` as a string, or ``None`` if the 

116 parameter does not exist. 

117 

118 Raises: 

119 botocore.exceptions.ClientError: Any error other than 

120 ``ParameterNotFound``. 

121 botocore.exceptions.BotoCoreError: Transport / credential 

122 failures. 

123 """ 

124 from botocore.exceptions import ClientError 

125 

126 client = _ssm_client(region) 

127 try: 

128 response = client.get_parameter(Name=name) 

129 except ClientError as exc: 

130 # ``ParameterNotFound`` is the only error this helper translates 

131 # to ``None``; every other code propagates so the caller sees 

132 # the real failure. 

133 if exc.response.get("Error", {}).get("Code") == "ParameterNotFound": 

134 return None 

135 raise 

136 return str(response["Parameter"]["Value"]) 

137 

138 

139def check_ssm_parameter(name: str, *, region: str | None = None) -> tuple[bool, str]: 

140 """Return ``(True, "")`` iff the parameter exists, ``(False, error)`` otherwise. 

141 

142 Diagnostic-style helper that flattens every kind of failure 

143 (missing, throttled, denied, transport error) into a single 

144 boolean. The returned error string is the underlying exception's 

145 ``str()`` so a caller logging the result has the original message. 

146 

147 Args: 

148 name: Fully-qualified parameter name. 

149 region: Optional AWS region. 

150 

151 Returns: 

152 ``(True, "")`` when the parameter resolves; ``(False, str(exc))`` 

153 on any error including ``ParameterNotFound``. 

154 """ 

155 from botocore.exceptions import BotoCoreError, ClientError 

156 

157 try: 

158 _ssm_client(region).get_parameter(Name=name) 

159 except (ClientError, BotoCoreError) as exc: 

160 return False, str(exc) 

161 return True, "" 

162 

163 

164def put_ssm_parameter( 

165 name: str, 

166 value: str, 

167 *, 

168 region: str | None = None, 

169 parameter_type: str = "String", 

170 overwrite: bool = True, 

171) -> None: 

172 """Write or overwrite an SSM parameter. Errors propagate verbatim. 

173 

174 Args: 

175 name: Fully-qualified parameter name. 

176 value: New string value to store. 

177 region: Optional AWS region. 

178 parameter_type: SSM parameter type — ``"String"`` (default), 

179 ``"StringList"``, or ``"SecureString"``. 

180 overwrite: When ``True`` (default), passes ``Overwrite=True`` 

181 to ``ssm:PutParameter`` so an existing parameter is 

182 replaced. When ``False``, the call fails with 

183 ``ParameterAlreadyExists`` if the name is taken. 

184 

185 Raises: 

186 botocore.exceptions.ClientError: For SSM-side errors (insufficient 

187 permission, parameter type mismatch, etc.). 

188 botocore.exceptions.BotoCoreError: For transport / credential 

189 failures. 

190 

191 The helper does not catch any exception — callers that need to 

192 classify failures (e.g. retry on throttle) match on the 

193 underlying ``Code`` field themselves. 

194 """ 

195 _ssm_client(region).put_parameter( 

196 Name=name, 

197 Value=value, 

198 Type=parameter_type, 

199 Overwrite=overwrite, 

200 )