Coverage for tests/integrations/resources/test_list.py: 100%

66 statements  

« prev     ^ index     » next       coverage.py v7.16.0, created at 2026-11-09 00:25 +0000

1import json 1abcdefg

2from pathlib import Path 1abcdefg

3from typing import Literal 1abcdefg

4 

5import pytest 1abcdefg

6import respx 1abcdefg

7from httpx import Response 1abcdefg

8from inline_snapshot import snapshot 1abcdefg

9 

10from fastapi_cloud_cli.cli import cloud_app as app 1abcdefg

11from fastapi_cloud_cli.utils.apps import AppConfig, write_app_config 1abcdefg

12from tests.utils import SnapshotCliRunner, changing_dir 1abcdefg

13 

14runner = SnapshotCliRunner() 1abcdefg

15 

16Provider = Literal["neon", "redis", "supabase", "logfire"] 1abcdefg

17 

18APP_ID = "00000000-0000-4000-8000-000000000001" 1abcdefg

19LINKED_APP_ID = "00000000-0000-4000-8000-000000000002" 1abcdefg

20TEAM_ID = "00000000-0000-4000-8000-000000000003" 1abcdefg

21RESOURCE_ID = "00000000-0000-4000-8000-000000000004" 1abcdefg

22 

23 

24def _resource( 1abcdefg

25 *, 

26 resource_id: str = RESOURCE_ID, 

27 name: str = "production", 

28 provider: Provider = "neon", 

29) -> dict[str, object]: 

30 return { 1hvoiwpjxqkyrlzsmAtnBu

31 "id": resource_id, 

32 "name": name, 

33 "app_id": APP_ID, 

34 "integration_id": "00000000-0000-4000-8000-000000000005", 

35 "provider_metadata": { 

36 "type": provider, 

37 "database_name": "unused-by-list", 

38 }, 

39 "console_url": "https://provider.example/resource", 

40 "created_at": "2026-08-18T10:00:00Z", 

41 "updated_at": "2026-08-18T10:00:00Z", 

42 } 

43 

44 

45def test_lists_resources_json_returns_not_logged_in_when_logged_out( 1abcdefg

46 logged_out_cli: None, 

47) -> None: 

48 result = runner.invoke( 1QRSTUVW

49 app, 

50 ["integrations", "resources", "list", "--app-id", APP_ID, "--json"], 

51 ) 

52 

53 assert result.exit_code == 1 1QRSTUVW

54 assert json.loads(result.stdout) == { 1QRSTUVW

55 "error": { 

56 "code": "not_logged_in", 

57 "message": "No credentials found.", 

58 "hint": "Run `fastapi cloud login`.", 

59 } 

60 } 

61 assert result.stderr == "" 1QRSTUVW

62 

63 

64@pytest.mark.respx 1abcdefg

65def test_lists_resources_as_json_with_explicit_app_overriding_linked_app( 1abcdefg

66 logged_in_cli: None, 

67 respx_mock: respx.MockRouter, 

68 tmp_path: Path, 

69) -> None: 

70 write_app_config( 1hijklmn

71 tmp_path, 

72 AppConfig(app_id=LINKED_APP_ID, team_id=TEAM_ID), 

73 ) 

74 respx_mock.get(f"/apps/{APP_ID}/connected-resources").mock( 1hijklmn

75 return_value=Response( 

76 200, 

77 json={"data": [_resource()], "count": 1}, 

78 ) 

79 ) 

80 

81 with changing_dir(tmp_path): 1hijklmn

82 result = runner.invoke( 1hijklmn

83 app, 

84 [ 

85 "integrations", 

86 "resources", 

87 "list", 

88 "--app-id", 

89 APP_ID, 

90 "--json", 

91 ], 

92 ) 

93 

94 assert result.exit_code == 0 1hijklmn

95 assert json.loads(result.stdout) == { 1hijklmn

96 "data": { 

97 "app_id": APP_ID, 

98 "resources": [ 

99 { 

100 "id": RESOURCE_ID, 

101 "name": "production", 

102 "provider": "neon", 

103 } 

104 ], 

105 } 

106 } 

107 assert result.stderr == "" 1hijklmn

108 

109 

110@pytest.mark.respx 1abcdefg

111def test_lists_resources_uses_linked_app( 1abcdefg

112 logged_in_cli: None, 

113 respx_mock: respx.MockRouter, 

114 tmp_path: Path, 

115) -> None: 

116 write_app_config( 1opqrstu

117 tmp_path, 

118 AppConfig(app_id=APP_ID, team_id=TEAM_ID), 

119 ) 

120 respx_mock.get(f"/apps/{APP_ID}/connected-resources").mock( 1opqrstu

121 return_value=Response( 

122 200, 

123 json={"data": [_resource(provider="redis")], "count": 1}, 

124 ) 

125 ) 

126 

127 with changing_dir(tmp_path): 1opqrstu

128 result = runner.invoke( 1opqrstu

129 app, 

130 ["integrations", "resources", "list", "--json"], 

131 ) 

132 

133 assert result.exit_code == 0 1opqrstu

134 assert json.loads(result.stdout) == { 1opqrstu

135 "data": { 

136 "app_id": APP_ID, 

137 "resources": [ 

138 { 

139 "id": RESOURCE_ID, 

140 "name": "production", 

141 "provider": "redis", 

142 } 

143 ], 

144 } 

145 } 

146 

147 

148def test_lists_resources_json_requires_app_without_linked_app( 1abcdefg

149 logged_in_cli: None, 

150 tmp_path: Path, 

151) -> None: 

152 with changing_dir(tmp_path): 1CDEFGHI

153 result = runner.invoke( 1CDEFGHI

154 app, 

155 ["integrations", "resources", "list", "--json"], 

156 ) 

157 

158 assert result.exit_code == 1 1CDEFGHI

159 assert json.loads(result.stdout) == { 1CDEFGHI

160 "error": { 

161 "code": "missing_required_input", 

162 "message": "App ID is required.", 

163 "hint": "Pass --app-id or run `fastapi cloud apps create --link` first.", 

164 } 

165 } 

166 assert result.stderr == "" 1CDEFGHI

167 

168 

169@pytest.mark.respx 1abcdefg

170def test_lists_resources_in_human_output( 1abcdefg

171 logged_in_cli: None, 

172 respx_mock: respx.MockRouter, 

173) -> None: 

174 resources = [ 1vwxyzAB

175 _resource( 

176 resource_id="00000000-0000-4000-8000-000000000010", 

177 name="primary", 

178 provider="neon", 

179 ), 

180 _resource( 

181 resource_id="00000000-0000-4000-8000-000000000011", 

182 name="cache", 

183 provider="redis", 

184 ), 

185 _resource( 

186 resource_id="00000000-0000-4000-8000-000000000012", 

187 name="analytics", 

188 provider="supabase", 

189 ), 

190 _resource( 

191 resource_id="00000000-0000-4000-8000-000000000013", 

192 name="observability", 

193 provider="logfire", 

194 ), 

195 ] 

196 respx_mock.get(f"/apps/{APP_ID}/connected-resources").mock( 1vwxyzAB

197 return_value=Response(200, json={"data": resources, "count": len(resources)}) 

198 ) 

199 

200 result = runner.invoke( 1vwxyzAB

201 app, 

202 ["integrations", "resources", "list", "--app-id", APP_ID], 

203 ) 

204 

205 assert result.exit_code == 0 1vwxyzAB

206 assert result.output == snapshot("""\ 1vwxyzAB

207connected resources 

208 

209Name Provider Resource ID 

210 

211primary Neon 00000000-0000-4000-8000-000000000010 

212cache Redis Cloud 00000000-0000-4000-8000-000000000011 

213analytics Supabase 00000000-0000-4000-8000-000000000012 

214observability Logfire 00000000-0000-4000-8000-000000000013\ 

215""") 

216 

217 

218@pytest.mark.respx 1abcdefg

219def test_lists_resources_in_human_output_empty( 1abcdefg

220 logged_in_cli: None, 

221 respx_mock: respx.MockRouter, 

222) -> None: 

223 respx_mock.get(f"/apps/{APP_ID}/connected-resources").mock( 1XYZ0123

224 return_value=Response(200, json={"data": [], "count": 0}) 

225 ) 

226 

227 result = runner.invoke( 1XYZ0123

228 app, 

229 ["integrations", "resources", "list", "--app-id", APP_ID], 

230 ) 

231 

232 assert result.exit_code == 0 1XYZ0123

233 assert result.output == snapshot("""\ 1XYZ0123

234connected resources 

235 

236No connected resources found.\ 

237""") 

238 

239 

240@pytest.mark.respx 1abcdefg

241def test_lists_resources_json_returns_not_found_for_unknown_app( 1abcdefg

242 logged_in_cli: None, 

243 respx_mock: respx.MockRouter, 

244) -> None: 

245 respx_mock.get(f"/apps/{APP_ID}/connected-resources").mock( 1JKLMNOP

246 return_value=Response(404) 

247 ) 

248 

249 result = runner.invoke( 1JKLMNOP

250 app, 

251 ["integrations", "resources", "list", "--app-id", APP_ID, "--json"], 

252 ) 

253 

254 assert result.exit_code == 1 1JKLMNOP

255 assert json.loads(result.stdout) == { 1JKLMNOP

256 "error": { 

257 "code": "not_found", 

258 "message": "App not found.", 

259 "hint": None, 

260 } 

261 } 

262 assert result.stderr == "" 1JKLMNOP