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

78 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 unittest.mock import patch 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 Keys, SnapshotCliRunner, changing_dir 1abcdefg

13 

14runner = SnapshotCliRunner() 1abcdefg

15 

16TEAM_ID = "00000000-0000-4000-8000-000000000001" 1abcdefg

17LINKED_TEAM_ID = "00000000-0000-4000-8000-000000000002" 1abcdefg

18INTEGRATION_ID = "00000000-0000-4000-8000-000000000003" 1abcdefg

19 

20NEON_AVAILABLE = { 1abcdefg

21 "id": "neon", 

22 "name": "Neon", 

23 "description": "Serverless Postgres database for your applications.", 

24 "category": "database", 

25 "status": "available", 

26 "default_env_var_name": "DATABASE_URL", 

27 "suggested_env_var_name": "NEON_DATABASE_URL", 

28 "connection_string_template": ( 

29 "postgresql://<user>:<password>@<host>/<database>" 

30 "?sslmode=require&channel_binding=require" 

31 ), 

32} 

33NEON_CONNECTED = { 1abcdefg

34 **NEON_AVAILABLE, 

35 "status": "connected", 

36 "connected_integration": { 

37 "id": INTEGRATION_ID, 

38 "type": "neon", 

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

40 }, 

41} 

42REDIS_AVAILABLE = { 1abcdefg

43 "id": "redis", 

44 "name": "Redis Cloud", 

45 "description": "In-memory data store for caching and real-time features.", 

46 "category": "database", 

47 "status": "available", 

48 "default_env_var_name": "REDIS_URL", 

49 "suggested_env_var_name": "REDIS_CLOUD_URL", 

50 "connection_string_template": None, 

51} 

52LOGFIRE_COMING_SOON = { 1abcdefg

53 "id": "logfire", 

54 "name": "Logfire", 

55 "description": "Observability for traces, metrics, and logs.", 

56 "category": "observability", 

57 "status": "coming_soon", 

58 "default_env_var_name": "LOGFIRE_TOKEN", 

59 "suggested_env_var_name": "LOGFIRE_TOKEN_2", 

60 "connection_string_template": None, 

61} 

62NEON_CONNECTED_OUTPUT = { 1abcdefg

63 "id": "neon", 

64 "name": "Neon", 

65 "status": "connected", 

66 "connected_integration": {"id": INTEGRATION_ID}, 

67} 

68REDIS_AVAILABLE_OUTPUT = { 1abcdefg

69 "id": "redis", 

70 "name": "Redis Cloud", 

71 "status": "available", 

72 "connected_integration": None, 

73} 

74 

75 

76def test_lists_providers_json_returns_not_logged_in_when_logged_out( 1abcdefg

77 logged_out_cli: None, 

78) -> None: 

79 result = runner.invoke( 1XYZ0123

80 app, 

81 ["integrations", "providers", "list", "--team-id", TEAM_ID, "--json"], 

82 ) 

83 

84 assert result.exit_code == 1 1XYZ0123

85 assert json.loads(result.stdout) == { 1XYZ0123

86 "error": { 

87 "code": "not_logged_in", 

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

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

90 } 

91 } 

92 assert result.stderr == "" 1XYZ0123

93 

94 

95@pytest.mark.respx 1abcdefg

96def test_lists_providers_as_json_with_explicit_team_overriding_linked_app( 1abcdefg

97 logged_in_cli: None, 

98 respx_mock: respx.MockRouter, 

99 tmp_path: Path, 

100) -> None: 

101 write_app_config( 1jklmnop

102 tmp_path, 

103 AppConfig( 

104 app_id="00000000-0000-4000-8000-000000000004", team_id=LINKED_TEAM_ID 

105 ), 

106 ) 

107 respx_mock.get(f"/teams/{TEAM_ID}/integrations").mock( 1jklmnop

108 return_value=Response( 

109 200, 

110 json={"data": [NEON_CONNECTED, REDIS_AVAILABLE], "count": 2}, 

111 ) 

112 ) 

113 

114 with changing_dir(tmp_path): 1jklmnop

115 result = runner.invoke( 1jklmnop

116 app, 

117 [ 

118 "integrations", 

119 "providers", 

120 "list", 

121 "--team-id", 

122 TEAM_ID, 

123 "--json", 

124 ], 

125 ) 

126 

127 assert result.exit_code == 0 1jklmnop

128 assert json.loads(result.stdout) == { 1jklmnop

129 "data": { 

130 "team_id": TEAM_ID, 

131 "providers": [NEON_CONNECTED_OUTPUT, REDIS_AVAILABLE_OUTPUT], 

132 } 

133 } 

134 assert result.stderr == "" 1jklmnop

135 

136 

137@pytest.mark.respx 1abcdefg

138def test_lists_providers_uses_linked_app_team( 1abcdefg

139 logged_in_cli: None, 

140 respx_mock: respx.MockRouter, 

141 tmp_path: Path, 

142) -> None: 

143 write_app_config( 1vwxyzAB

144 tmp_path, 

145 AppConfig(app_id="00000000-0000-4000-8000-000000000004", team_id=TEAM_ID), 

146 ) 

147 respx_mock.get(f"/teams/{TEAM_ID}/integrations").mock( 1vwxyzAB

148 return_value=Response( 

149 200, 

150 json={"data": [REDIS_AVAILABLE], "count": 1}, 

151 ) 

152 ) 

153 

154 with changing_dir(tmp_path): 1vwxyzAB

155 result = runner.invoke( 1vwxyzAB

156 app, 

157 ["integrations", "providers", "list", "--json"], 

158 ) 

159 

160 assert result.exit_code == 0 1vwxyzAB

161 assert json.loads(result.stdout) == { 1vwxyzAB

162 "data": { 

163 "team_id": TEAM_ID, 

164 "providers": [REDIS_AVAILABLE_OUTPUT], 

165 } 

166 } 

167 

168 

169def test_lists_providers_json_requires_team_without_linked_app( 1abcdefg

170 logged_in_cli: None, 

171 tmp_path: Path, 

172) -> None: 

173 with changing_dir(tmp_path): 1CDEFGHI

174 result = runner.invoke( 1CDEFGHI

175 app, 

176 ["integrations", "providers", "list", "--json"], 

177 ) 

178 

179 assert result.exit_code == 1 1CDEFGHI

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

181 "error": { 

182 "code": "missing_required_input", 

183 "message": "Team ID is required.", 

184 "hint": "Pass --team-id or run the command from a linked app.", 

185 } 

186 } 

187 assert result.stderr == "" 1CDEFGHI

188 

189 

190@pytest.mark.respx 1abcdefg

191def test_lists_providers_prompts_for_team_and_renders_statuses( 1abcdefg

192 logged_in_cli: None, 

193 respx_mock: respx.MockRouter, 

194 tmp_path: Path, 

195) -> None: 

196 team = {"id": TEAM_ID, "slug": "acme", "name": "Acme"} 1hqrsitu

197 respx_mock.get("/teams/").mock( 1hqrsitu

198 return_value=Response(200, json={"data": [team], "count": 1}) 

199 ) 

200 respx_mock.get(f"/teams/{TEAM_ID}/integrations").mock( 1hqrsitu

201 return_value=Response( 

202 200, 

203 json={ 

204 "data": [LOGFIRE_COMING_SOON, NEON_CONNECTED, REDIS_AVAILABLE], 

205 "count": 3, 

206 }, 

207 ) 

208 ) 

209 

210 with ( 1hi

211 changing_dir(tmp_path), 

212 patch("rich_toolkit.container.getchar") as mock_getchar, 

213 ): 

214 mock_getchar.side_effect = [Keys.ENTER] 1hqrsitu

215 result = runner.invoke(app, ["integrations", "providers", "list"]) 1hqrsitu

216 

217 assert result.exit_code == 0 1hqrsitu

218 assert result.output == snapshot("""\ 1hqrsitu

219Select the team: 

220Filter: 

221 

222● Acme 

223 

224Select the team: Acme 

225 

226integrations 

227 

228Name Status Integration ID 

229 

230Logfire coming soon - 

231Neon connected 00000000-0000-4000-8000-000000000003 

232Redis Cloud not connected -\ 

233""") 

234 

235 

236@pytest.mark.respx 1abcdefg

237def test_lists_providers_returns_missing_input_when_no_teams_exist( 1abcdefg

238 logged_in_cli: None, 

239 respx_mock: respx.MockRouter, 

240 tmp_path: Path, 

241) -> None: 

242 respx_mock.get("/teams/").mock( 1JKLMNOP

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

244 ) 

245 

246 with changing_dir(tmp_path): 1JKLMNOP

247 result = runner.invoke(app, ["integrations", "providers", "list"]) 1JKLMNOP

248 

249 assert result.exit_code == 1 1JKLMNOP

250 assert result.output == snapshot("""\ 1JKLMNOP

251✗ error: No teams found. 

252 

253 hint: Create a team before listing integration providers.\ 

254""") 

255 

256 

257@pytest.mark.respx 1abcdefg

258def test_lists_providers_in_human_output_empty( 1abcdefg

259 logged_in_cli: None, 

260 respx_mock: respx.MockRouter, 

261) -> None: 

262 respx_mock.get(f"/teams/{TEAM_ID}/integrations").mock( 1456789!

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

264 ) 

265 

266 result = runner.invoke( 1456789!

267 app, 

268 ["integrations", "providers", "list", "--team-id", TEAM_ID], 

269 ) 

270 

271 assert result.exit_code == 0 1456789!

272 assert result.output == snapshot("""\ 1456789!

273integrations 

274 

275No integration providers available.\ 

276""") 

277 

278 

279@pytest.mark.respx 1abcdefg

280def test_lists_providers_json_returns_not_found_for_unknown_team( 1abcdefg

281 logged_in_cli: None, 

282 respx_mock: respx.MockRouter, 

283) -> None: 

284 respx_mock.get(f"/teams/{TEAM_ID}/integrations").mock(return_value=Response(404)) 1QRSTUVW

285 

286 result = runner.invoke( 1QRSTUVW

287 app, 

288 ["integrations", "providers", "list", "--team-id", TEAM_ID, "--json"], 

289 ) 

290 

291 assert result.exit_code == 1 1QRSTUVW

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

293 "error": { 

294 "code": "not_found", 

295 "message": "Team not found.", 

296 "hint": None, 

297 } 

298 } 

299 assert result.stderr == "" 1QRSTUVW