Coverage for tests / conftest.py: 43%

1""" 

2Pytest configuration and shared fixtures for superset-io tests. 

3""" 

4 

5import logging 

6import os 

7import subprocess 

8import time 

9import zipfile 

10from collections.abc import Generator 

11from pathlib import Path 

12 

13import pytest 

14import requests 

15import yaml 

16 

17from superset_io.api import SuperSetApiClient, SupersetApiSession 

18 

19from .docker_compose import SupersetDockerCompose, get_docker_compose_path 

20 

21logger = logging.getLogger(__name__) 

22 

23 

24# ---------------------------------- Config ---------------------------------- # 

25 

26 

27def pytest_addoption(parser): 

28 """Add command-line options for pytest.""" 

29 parser.addoption( 

30 "--integration", 

31 action="store_true", 

32 default=False, 

33 help="Run integration tests (requires Superset instance)", 

34 ) 

35 parser.addoption( 

36 "--superset-url", 

37 action="store", 

38 default=os.environ.get("SUPERSET_TEST_URL", "http://localhost:8088"), 

39 help="Superset base URL for integration tests", 

40 ) 

41 parser.addoption( 

42 "--superset-username", 

43 action="store", 

44 default=os.environ.get("SUPERSET_TEST_USERNAME", "admin"), 

45 help="Superset username for integration tests", 

46 ) 

47 parser.addoption( 

48 "--superset-password", 

49 action="store", 

50 default=os.environ.get("SUPERSET_TEST_PASSWORD", "admin"), 

51 help="Superset password for integration tests", 

52 ) 

53 

54 

55def pytest_configure(config): 

56 """Configure pytest based on options.""" 

57 config.addinivalue_line( 

58 "markers", "integration: mark test as integration test (requires Superset)" 

59 ) 

60 

61 

62def pytest_collection_modifyitems(config, items): 

63 """Skip integration tests unless --integration flag is provided. 

64 

65 Mark a test by adding `@pytest.mark.integration` to it. 

66 """ 

67 if not config.getoption("--integration"): 

68 skip_integration = pytest.mark.skip( 

69 reason="Integration tests require --integration flag" 

70 ) 

71 for item in items: 

72 if "integration" in item.keywords: 

73 item.add_marker(skip_integration) 

74 

75 

76# --------------------------------- Fixtures --------------------------------- # 

77 

78 

79@pytest.fixture(scope="session") 

80def superset_url(request) -> str: 

81 """Return the Superset base URL for integration tests.""" 

82 return request.config.getoption("--superset-url") 

83 

84 

85@pytest.fixture(scope="session") 

86def superset_credentials(request): 

87 """Return Superset credentials as a tuple (username, password).""" 

88 username = request.config.getoption("--superset-username") 

89 password = request.config.getoption("--superset-password") 

90 return username, password 

91 

92 

93@pytest.fixture(scope="session") 

94def superset(request) -> Generator[SupersetDockerCompose, None, None]: 

95 """ 

96 Start a Superset docker-compose stack if --start-docker flag is provided. 

97 

98 Yields a SupersetDockerCompose instance. The stack is automatically 

99 stopped after the test session. 

100 """ 

101 compose_path = get_docker_compose_path() 

102 if not compose_path.exists(): 

103 pytest.skip(f"Docker compose file not found at {compose_path}") 

104 

105 # Check if docker-compose is available 

106 try: 

107 subprocess.run(["docker-compose", "--version"], capture_output=True, check=True) 

108 except (subprocess.CalledProcessError, FileNotFoundError): 

109 pytest.skip("docker-compose not available") 

110 

111 # Check if Docker is running 

112 try: 

113 subprocess.run(["docker", "info"], capture_output=True, check=True) 

114 except (subprocess.CalledProcessError, FileNotFoundError): 

115 pytest.skip("Docker not available") 

116 

117 docker_compose = SupersetDockerCompose(compose_path) 

118 try: 

119 docker_compose.start() 

120 yield docker_compose 

121 finally: 

122 docker_compose.stop() 

123 

124 

125@pytest.fixture(scope="session") 

126def superset_client( 

127 superset_url, superset_credentials, superset_docker_compose 

128) -> SuperSetApiClient: 

129 """ 

130 Create an authenticated Superset API client for integration tests. 

131 

132 This fixture will skip if Superset is not reachable. 

133 If --start-docker is provided, it will start the stack before connecting. 

134 """ 

135 username, password = superset_credentials 

136 max_attempts = 5 

137 for attempt in range(max_attempts): 

138 try: 

139 # Quick health check 

140 resp = requests.get(f"{superset_url}/health", timeout=10) 

141 resp.raise_for_status() 

142 break 

143 except (requests.exceptions.ConnectionError, requests.exceptions.Timeout): 

144 if attempt == max_attempts - 1: 

145 pytest.skip(f"Superset not reachable at {superset_url}") 

146 time.sleep(5) 

147 

148 session = SupersetApiSession.from_credentials( 

149 base_url=superset_url, 

150 username=username, 

151 password=password, 

152 ) 

153 return SuperSetApiClient(session) 

154 

155 

156@pytest.fixture 

157def mock_superset_session(mocker): 

158 """Create a mock SupersetApiSession for unit tests.""" 

159 mock_session = mocker.Mock(spec=SupersetApiSession) 

160 mock_session.base_url = "http://localhost:8088" 

161 mock_session.bearer_token = "mock-token" 

162 mock_session.csrf_token = "mock-csrf" 

163 mock_session.headers = { 

164 "Authorization": "Bearer mock-token", 

165 "X-CSRFToken": "mock-csrf", 

166 } 

167 return mock_session 

168 

169 

170@pytest.fixture 

171def mock_superset_client(mock_superset_session): 

172 """Create a mock SuperSetApiClient for unit tests.""" 

173 return SuperSetApiClient(mock_superset_session) 

174 

175 

176@pytest.fixture(scope="session") 

177def test_assets_dir() -> Path: 

178 """Return the path to the test assets directory.""" 

179 assets_dir = Path(__file__).parent / "assets" 

180 assets_dir.mkdir(exist_ok=True) 

181 return assets_dir 

182 

183 

184def create_sample_dashboard_zip(zip_path: Path) -> None: 

185 """Create a minimal valid Superset dashboard export ZIP file.""" 

186 # Create a minimal valid Superset export structure 

187 metadata = { 

188 "version": 1.0, 

189 "type": "assets", 

190 "timestamp": "2024-01-01T00:00:00Z", 

191 } 

192 

193 dashboard = { 

194 "dashboard_title": "Sample Dashboard", 

195 "description": "A sample dashboard for testing", 

196 "css": "", 

197 "position_json": "{}", 

198 "published": True, 

199 } 

200 

201 with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf: 

202 # Add metadata.yaml 

203 zf.writestr( 

204 "assets_export_sample/metadata.yaml", 

205 yaml.dump(metadata, default_flow_style=False), 

206 ) 

207 # Add a dashboard file 

208 zf.writestr( 

209 "assets_export_sample/dashboards/sample_dashboard.yaml", 

210 yaml.dump(dashboard, default_flow_style=False), 

211 ) 

212 # Add empty directories for other asset types 

213 zf.writestr("assets_export_sample/charts/", "") 

214 zf.writestr("assets_export_sample/datasets/", "") 

215 zf.writestr("assets_export_sample/databases/", "") 

216 

217 

218@pytest.fixture(scope="session") 

219def sample_dashboard_zip(test_assets_dir) -> bytes: 

220 """ 

221 Provide a sample dashboard export ZIP for testing. 

222 

223 Creates a minimal valid Superset export if it doesn't exist. 

224 """ 

225 zip_path = test_assets_dir / "sample_dashboard.zip" 

226 

227 if not zip_path.exists(): 

228 logger.info(f"Creating sample dashboard ZIP at {zip_path}") 

229 create_sample_dashboard_zip(zip_path) 

230 

231 return zip_path.read_bytes()