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

1import json 7 ctx1abcdefg

2from pathlib import Path 7 ctx1abcdefg

3from unittest.mock import patch 7 ctx1abcdefg

4 

5import pytest 7 ctx1abcdefg

6import respx 7 ctx1abcdefg

7from httpx import Response 7 ctx1abcdefg

8from inline_snapshot import snapshot 7 ctx1abcdefg

9 

10from fastapi_cloud_cli.cli import cloud_app as app 7 ctx1abcdefg

11from fastapi_cloud_cli.config import Settings 7 ctx1abcdefg

12from fastapi_cloud_cli.utils.apps import AppConfig, write_app_config 7 ctx1abcdefg

13from tests.utils import SnapshotCliRunner, changing_dir 7 ctx1abcdefg

14 

15runner = SnapshotCliRunner() 7 ctx1abcdefg

16 

17APP_ID = "00000000-0000-4000-8000-000000000001" 7 ctx1abcdefg

18LINKED_APP_ID = "00000000-0000-4000-8000-000000000002" 7 ctx1abcdefg

19TEAM_ID = "00000000-0000-4000-8000-000000000003" 7 ctx1abcdefg

20TEAM_SLUG = "acme" 7 ctx1abcdefg

21APP_SLUG = "api" 7 ctx1abcdefg

22CONNECT_URL = ( 7 ctx1abcdefg

23 f"{Settings().dashboard_base_url}/{TEAM_SLUG}/apps/{APP_SLUG}/integrations/connect" 

24) 

25 

26 

27def _app(*, app_id: str = APP_ID) -> dict[str, object]: 7 ctx1abcdefg

28 return { 21 ctx1hqjkrsltumvwixnoyzpAB

29 "id": app_id, 

30 "team_id": TEAM_ID, 

31 "slug": APP_SLUG, 

32 "name": "API", 

33 "directory": None, 

34 } 

35 

36 

37def _team() -> dict[str, str]: 7 ctx1abcdefg

38 return { 21 ctx1hqjkrsltumvwixnoyzpAB

39 "id": TEAM_ID, 

40 "slug": TEAM_SLUG, 

41 "name": "Acme", 

42 } 

43 

44 

45def _mock_app_and_team( 7 ctx1abcdefg

46 respx_mock: respx.MockRouter, 

47 *, 

48 app_id: str = APP_ID, 

49) -> None: 

50 respx_mock.get(f"/apps/{app_id}").mock( 21 ctx1hqjkrsltumvwixnoyzpAB

51 return_value=Response(200, json=_app(app_id=app_id)) 

52 ) 

53 respx_mock.get(f"/teams/{TEAM_ID}").mock(return_value=Response(200, json=_team())) 21 ctx1hqjkrsltumvwixnoyzpAB

54 

55 

56def test_connect_resource_json_returns_not_logged_in_when_logged_out( 7 ctx1abcdefg

57 logged_out_cli: None, 

58) -> None: 

59 result = runner.invoke( 7 ctx1QRSTUVW

60 app, 

61 ["integrations", "resources", "connect", "--app-id", APP_ID, "--json"], 

62 ) 

63 

64 assert result.exit_code == 1 7 ctx1QRSTUVW

65 assert json.loads(result.stdout) == { 7 ctx1QRSTUVW

66 "error": { 

67 "code": "not_logged_in", 

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

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

70 } 

71 } 

72 assert result.stderr == "" 7 ctx1QRSTUVW

73 

74 

75@pytest.mark.respx 7 ctx1abcdefg

76def test_connect_resource_json_returns_url_without_opening_browser( 7 ctx1abcdefg

77 logged_in_cli: None, 

78 respx_mock: respx.MockRouter, 

79 tmp_path: Path, 

80) -> None: 

81 write_app_config( 7 ctx1hklmiop

82 tmp_path, 

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

84 ) 

85 _mock_app_and_team(respx_mock) 7 ctx1hklmiop

86 

87 with ( 2 ctx1hi

88 changing_dir(tmp_path), 

89 patch( 

90 "fastapi_cloud_cli.commands.integrations.resources.connect.typer.launch" 

91 ) as mock_launch, 

92 ): 

93 result = runner.invoke( 7 ctx1hklmiop

94 app, 

95 [ 

96 "integrations", 

97 "resources", 

98 "connect", 

99 "--app-id", 

100 APP_ID, 

101 "--json", 

102 ], 

103 ) 

104 

105 assert result.exit_code == 0 7 ctx1hklmiop

106 assert json.loads(result.stdout) == { 7 ctx1hklmiop

107 "data": { 

108 "app_id": APP_ID, 

109 "connect_url": CONNECT_URL, 

110 } 

111 } 

112 assert result.stderr == "" 7 ctx1hklmiop

113 mock_launch.assert_not_called() 7 ctx1hklmiop

114 

115 

116@pytest.mark.respx 7 ctx1abcdefg

117def test_connect_resource_opens_browser_for_linked_app( 7 ctx1abcdefg

118 logged_in_cli: None, 

119 respx_mock: respx.MockRouter, 

120 tmp_path: Path, 

121) -> None: 

122 write_app_config( 7 ctx1jsuwnzB

123 tmp_path, 

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

125 ) 

126 _mock_app_and_team(respx_mock) 7 ctx1jsuwnzB

127 

128 with ( 2 ctx1jn

129 changing_dir(tmp_path), 

130 patch( 

131 "fastapi_cloud_cli.commands.integrations.resources.connect.typer.launch", 

132 return_value=0, 

133 ) as mock_launch, 

134 ): 

135 result = runner.invoke(app, ["integrations", "resources", "connect"]) 7 ctx1jsuwnzB

136 

137 assert result.exit_code == 0 7 ctx1jsuwnzB

138 mock_launch.assert_called_once_with(CONNECT_URL) 7 ctx1jsuwnzB

139 assert result.output == snapshot("""\ 7 ctx1jsuwnzB

140connect resource 

141 

142🔌 Opened the integration setup for API in your browser. 

143 https://dashboard.fastapicloud.com/acme/apps/api/integrations/connect 

144 

145💡 When the connection is complete, run fastapi cloud integrations resources 

146 list to view it.\ 

147""") 

148 

149 

150@pytest.mark.respx 7 ctx1abcdefg

151def test_connect_resource_no_open_prints_url_without_opening_browser( 7 ctx1abcdefg

152 logged_in_cli: None, 

153 respx_mock: respx.MockRouter, 

154) -> None: 

155 _mock_app_and_team(respx_mock) 7 ctx1qrtvxyA

156 

157 with patch( 7 ctx1qrtvxyA

158 "fastapi_cloud_cli.commands.integrations.resources.connect.typer.launch" 

159 ) as mock_launch: 

160 result = runner.invoke( 7 ctx1qrtvxyA

161 app, 

162 [ 

163 "integrations", 

164 "resources", 

165 "connect", 

166 "--app-id", 

167 APP_ID, 

168 "--no-open", 

169 ], 

170 ) 

171 

172 assert result.exit_code == 0 7 ctx1qrtvxyA

173 mock_launch.assert_not_called() 7 ctx1qrtvxyA

174 assert result.output == snapshot("""\ 7 ctx1qrtvxyA

175connect resource 

176 

177🔌 Open the integration setup for API in your browser: 

178 https://dashboard.fastapicloud.com/acme/apps/api/integrations/connect 

179 

180💡 When the connection is complete, run fastapi cloud integrations resources 

181 list --app-id 00000000-0000-4000-8000-000000000001 to view it.\ 

182""") 

183 

184 

185def test_connect_resource_json_requires_app_without_linked_app( 7 ctx1abcdefg

186 logged_in_cli: None, 

187 tmp_path: Path, 

188) -> None: 

189 with changing_dir(tmp_path): 7 ctx1CDEFGHI

190 result = runner.invoke( 7 ctx1CDEFGHI

191 app, 

192 ["integrations", "resources", "connect", "--json"], 

193 ) 

194 

195 assert result.exit_code == 1 7 ctx1CDEFGHI

196 assert json.loads(result.stdout) == { 7 ctx1CDEFGHI

197 "error": { 

198 "code": "missing_required_input", 

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

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

201 } 

202 } 

203 assert result.stderr == "" 7 ctx1CDEFGHI

204 

205 

206@pytest.mark.respx 7 ctx1abcdefg

207def test_connect_resource_json_returns_not_found_for_unknown_app( 7 ctx1abcdefg

208 logged_in_cli: None, 

209 respx_mock: respx.MockRouter, 

210) -> None: 

211 respx_mock.get(f"/apps/{APP_ID}").mock(return_value=Response(404)) 7 ctx1JKLMNOP

212 

213 result = runner.invoke( 7 ctx1JKLMNOP

214 app, 

215 ["integrations", "resources", "connect", "--app-id", APP_ID, "--json"], 

216 ) 

217 

218 assert result.exit_code == 1 7 ctx1JKLMNOP

219 assert json.loads(result.stdout) == { 7 ctx1JKLMNOP

220 "error": { 

221 "code": "not_found", 

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

223 "hint": None, 

224 } 

225 } 

226 assert result.stderr == "" 7 ctx1JKLMNOP