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

80 statements  

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

1import json 1abcdefg

2from datetime import datetime, timezone 1abcdefg

3from pathlib import Path 1abcdefg

4from typing import Literal 1abcdefg

5 

6import pytest 1abcdefg

7import respx 1abcdefg

8import time_machine 1abcdefg

9from httpx import Response 1abcdefg

10from inline_snapshot import snapshot 1abcdefg

11 

12from fastapi_cloud_cli.cli import cloud_app as app 1abcdefg

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

14from tests.utils import SnapshotCliRunner, changing_dir 1abcdefg

15 

16runner = SnapshotCliRunner() 1abcdefg

17 

18DatabaseProvider = Literal["neon", "redis", "supabase"] 1abcdefg

19 

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

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

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

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

24INTEGRATION_ID = "00000000-0000-4000-8000-000000000005" 1abcdefg

25CONSOLE_URL = "https://provider.example/resource" 1abcdefg

26CREATED_AT = "2026-08-18T10:00:00Z" 1abcdefg

27UPDATED_AT = "2026-08-19T11:00:00Z" 1abcdefg

28 

29 

30def _database_resource( 1abcdefg

31 *, 

32 provider: DatabaseProvider = "neon", 

33 environment_variables: list[str] | None = None, 

34) -> dict[str, object]: 

35 if environment_variables is None: 1hviwjxkylzmAnB

36 environment_variables = ["DATABASE_URL", "DATABASE_PASSWORD"] 1hviwjxkylzmAnB

37 

38 return { 1hviwjxkylzmAnB

39 "id": RESOURCE_ID, 

40 "name": "production", 

41 "app_id": APP_ID, 

42 "integration_id": INTEGRATION_ID, 

43 "provider_metadata": { 

44 "type": provider, 

45 "database_name": "app_db", 

46 "project_id": "unused-project-id", 

47 "branch_id": "unused-branch-id", 

48 }, 

49 "console_url": CONSOLE_URL, 

50 "environment_variables": [{"name": name} for name in environment_variables], 

51 "created_at": CREATED_AT, 

52 "updated_at": UPDATED_AT, 

53 } 

54 

55 

56def _logfire_resource( 1abcdefg

57 *, 

58 environment_variables: list[str] | None = None, 

59) -> dict[str, object]: 

60 if environment_variables is None: 1oCpDqErFsGtHuI

61 environment_variables = ["LOGFIRE_TOKEN"] 1opqrstu

62 

63 return { 1oCpDqErFsGtHuI

64 "id": RESOURCE_ID, 

65 "name": "observability", 

66 "app_id": APP_ID, 

67 "integration_id": INTEGRATION_ID, 

68 "provider_metadata": { 

69 "type": "logfire", 

70 "project_id": "unused-project-id", 

71 "project_name": "fastapi-api", 

72 "organization_name": "acme", 

73 "write_token_id": "unused-token-id", 

74 }, 

75 "console_url": CONSOLE_URL, 

76 "environment_variables": [{"name": name} for name in environment_variables], 

77 "created_at": CREATED_AT, 

78 "updated_at": UPDATED_AT, 

79 } 

80 

81 

82def test_gets_resource_json_returns_not_logged_in_when_logged_out( 1abcdefg

83 logged_out_cli: None, 

84) -> None: 

85 result = runner.invoke( 1XYZ0123

86 app, 

87 [ 

88 "integrations", 

89 "resources", 

90 "get", 

91 RESOURCE_ID, 

92 "--app-id", 

93 APP_ID, 

94 "--json", 

95 ], 

96 ) 

97 

98 assert result.exit_code == 1 1XYZ0123

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

100 "error": { 

101 "code": "not_logged_in", 

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

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

104 } 

105 } 

106 assert result.stderr == "" 1XYZ0123

107 

108 

109@pytest.mark.respx 1abcdefg

110@pytest.mark.parametrize("provider", ["neon", "redis", "supabase"]) 1abcdefg

111def test_gets_database_resource_as_json_with_explicit_app_overriding_linked_app( 1abcdefg

112 logged_in_cli: None, 

113 respx_mock: respx.MockRouter, 

114 tmp_path: Path, 

115 provider: DatabaseProvider, 

116) -> None: 

117 write_app_config( 1hijklmn

118 tmp_path, 

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

120 ) 

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

122 return_value=Response(200, json=_database_resource(provider=provider)) 

123 ) 

124 

125 with changing_dir(tmp_path): 1hijklmn

126 result = runner.invoke( 1hijklmn

127 app, 

128 [ 

129 "integrations", 

130 "resources", 

131 "get", 

132 RESOURCE_ID, 

133 "--app-id", 

134 APP_ID, 

135 "--json", 

136 ], 

137 ) 

138 

139 assert result.exit_code == 0 1hijklmn

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

141 "data": { 

142 "app_id": APP_ID, 

143 "resource": { 

144 "id": RESOURCE_ID, 

145 "name": "production", 

146 "provider": provider, 

147 "console_url": CONSOLE_URL, 

148 "environment_variables": ["DATABASE_URL", "DATABASE_PASSWORD"], 

149 "created_at": CREATED_AT, 

150 "updated_at": UPDATED_AT, 

151 "database_name": "app_db", 

152 }, 

153 } 

154 } 

155 assert result.stderr == "" 1hijklmn

156 

157 

158@pytest.mark.respx 1abcdefg

159def test_gets_logfire_resource_as_json_using_linked_app( 1abcdefg

160 logged_in_cli: None, 

161 respx_mock: respx.MockRouter, 

162 tmp_path: Path, 

163) -> None: 

164 write_app_config( 1opqrstu

165 tmp_path, 

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

167 ) 

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

169 return_value=Response(200, json=_logfire_resource()) 

170 ) 

171 

172 with changing_dir(tmp_path): 1opqrstu

173 result = runner.invoke( 1opqrstu

174 app, 

175 ["integrations", "resources", "get", RESOURCE_ID, "--json"], 

176 ) 

177 

178 assert result.exit_code == 0 1opqrstu

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

180 "data": { 

181 "app_id": APP_ID, 

182 "resource": { 

183 "id": RESOURCE_ID, 

184 "name": "observability", 

185 "provider": "logfire", 

186 "console_url": CONSOLE_URL, 

187 "environment_variables": ["LOGFIRE_TOKEN"], 

188 "created_at": CREATED_AT, 

189 "updated_at": UPDATED_AT, 

190 "project_name": "fastapi-api", 

191 "organization_name": "acme", 

192 }, 

193 } 

194 } 

195 

196 

197def test_gets_resource_json_requires_app_without_linked_app( 1abcdefg

198 logged_in_cli: None, 

199 tmp_path: Path, 

200) -> None: 

201 with changing_dir(tmp_path): 1JKLMNOP

202 result = runner.invoke( 1JKLMNOP

203 app, 

204 ["integrations", "resources", "get", RESOURCE_ID, "--json"], 

205 ) 

206 

207 assert result.exit_code == 1 1JKLMNOP

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

209 "error": { 

210 "code": "missing_required_input", 

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

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

213 } 

214 } 

215 assert result.stderr == "" 1JKLMNOP

216 

217 

218@pytest.mark.respx 1abcdefg

219@time_machine.travel(datetime(2026, 9, 1, 10, tzinfo=timezone.utc), tick=False) 1abcdefg

220def test_gets_database_resource_in_human_output( 1abcdefg

221 logged_in_cli: None, 

222 respx_mock: respx.MockRouter, 

223) -> None: 

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

225 return_value=Response(200, json=_database_resource()) 

226 ) 

227 

228 result = runner.invoke( 1vwxyzAB

229 app, 

230 ["integrations", "resources", "get", RESOURCE_ID, "--app-id", APP_ID], 

231 ) 

232 

233 assert result.exit_code == 0 1vwxyzAB

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

235connected resource 

236 

237🔌 production 

238 

239 id 00000000-0000-4000-8000-000000000004 

240 provider Neon 

241 database app_db 

242 console https://provider.example/resource 

243 environment variables DATABASE_URL 

244 DATABASE_PASSWORD 

245 connected 14 days ago 

246 last updated 12 days ago\ 

247""") 

248 

249 

250@pytest.mark.respx 1abcdefg

251@time_machine.travel(datetime(2026, 9, 1, 10, tzinfo=timezone.utc), tick=False) 1abcdefg

252def test_gets_logfire_resource_in_human_output_without_environment_variables( 1abcdefg

253 logged_in_cli: None, 

254 respx_mock: respx.MockRouter, 

255) -> None: 

256 respx_mock.get(f"/apps/{APP_ID}/connected-resources/{RESOURCE_ID}").mock( 1CDEFGHI

257 return_value=Response( 

258 200, 

259 json=_logfire_resource(environment_variables=[]), 

260 ) 

261 ) 

262 

263 result = runner.invoke( 1CDEFGHI

264 app, 

265 ["integrations", "resources", "get", RESOURCE_ID, "--app-id", APP_ID], 

266 ) 

267 

268 assert result.exit_code == 0 1CDEFGHI

269 assert result.output == snapshot("""\ 1CDEFGHI

270connected resource 

271 

272🔌 observability 

273 

274 id 00000000-0000-4000-8000-000000000004 

275 provider Logfire 

276 project fastapi-api 

277 organization acme 

278 console https://provider.example/resource 

279 environment variables - 

280 connected 14 days ago 

281 last updated 12 days ago\ 

282""") 

283 

284 

285@pytest.mark.respx 1abcdefg

286def test_gets_resource_json_returns_not_found( 1abcdefg

287 logged_in_cli: None, 

288 respx_mock: respx.MockRouter, 

289) -> None: 

290 respx_mock.get(f"/apps/{APP_ID}/connected-resources/{RESOURCE_ID}").mock( 1QRSTUVW

291 return_value=Response(404) 

292 ) 

293 

294 result = runner.invoke( 1QRSTUVW

295 app, 

296 [ 

297 "integrations", 

298 "resources", 

299 "get", 

300 RESOURCE_ID, 

301 "--app-id", 

302 APP_ID, 

303 "--json", 

304 ], 

305 ) 

306 

307 assert result.exit_code == 1 1QRSTUVW

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

309 "error": { 

310 "code": "not_found", 

311 "message": "Connected resource not found.", 

312 "hint": None, 

313 } 

314 } 

315 assert result.stderr == "" 1QRSTUVW