|
| 1 | +"""End-to-end tests for Grafana version display functionality. |
| 2 | +
|
| 3 | +These tests verify that: |
| 4 | +1. The Flask /version endpoint returns valid data |
| 5 | +2. Grafana dashboard has the version panel configured |
| 6 | +3. The version data flows through to Grafana correctly |
| 7 | +
|
| 8 | +Prerequisites: |
| 9 | +- Run with `pytest tests/e2e/test_grafana_version_display.py -v` |
| 10 | +- The monitoring stack must be running (docker compose up) |
| 11 | +- Default ports: Flask on 8000, Grafana on 3000 |
| 12 | +""" |
| 13 | + |
| 14 | +import os |
| 15 | +import json |
| 16 | +import pytest |
| 17 | +import requests |
| 18 | +from urllib.parse import urljoin |
| 19 | + |
| 20 | + |
| 21 | +# Configuration - can be overridden with environment variables |
| 22 | +FLASK_BASE_URL = os.environ.get('FLASK_URL', 'http://localhost:8000') |
| 23 | +GRAFANA_BASE_URL = os.environ.get('GRAFANA_URL', 'http://localhost:3000') |
| 24 | +GRAFANA_USER = os.environ.get('GRAFANA_USER', 'admin') |
| 25 | +GRAFANA_PASSWORD = os.environ.get('GRAFANA_PASSWORD', 'admin') |
| 26 | + |
| 27 | +# Dashboard UID for Self Monitoring Dashboard |
| 28 | +SELF_MONITORING_DASHBOARD_UID = 'self_monitoring_dashboard' |
| 29 | + |
| 30 | + |
| 31 | +@pytest.fixture |
| 32 | +def flask_session(): |
| 33 | + """Create a requests session for Flask API.""" |
| 34 | + session = requests.Session() |
| 35 | + session.headers.update({'Accept': 'application/json'}) |
| 36 | + return session |
| 37 | + |
| 38 | + |
| 39 | +@pytest.fixture |
| 40 | +def grafana_session(): |
| 41 | + """Create a requests session for Grafana API with authentication.""" |
| 42 | + session = requests.Session() |
| 43 | + session.auth = (GRAFANA_USER, GRAFANA_PASSWORD) |
| 44 | + session.headers.update({'Accept': 'application/json'}) |
| 45 | + return session |
| 46 | + |
| 47 | + |
| 48 | +def is_service_available(url: str, timeout: int = 5) -> bool: |
| 49 | + """Check if a service is available.""" |
| 50 | + try: |
| 51 | + response = requests.get(url, timeout=timeout) |
| 52 | + return response.status_code < 500 |
| 53 | + except requests.exceptions.RequestException: |
| 54 | + return False |
| 55 | + |
| 56 | + |
| 57 | +@pytest.mark.e2e |
| 58 | +class TestFlaskVersionEndpoint: |
| 59 | + """Tests for the Flask /version endpoint.""" |
| 60 | + |
| 61 | + @pytest.fixture(autouse=True) |
| 62 | + def check_flask_available(self): |
| 63 | + """Skip tests if Flask is not available.""" |
| 64 | + if not is_service_available(FLASK_BASE_URL): |
| 65 | + pytest.skip(f"Flask backend not available at {FLASK_BASE_URL}") |
| 66 | + |
| 67 | + def test_version_endpoint_returns_200(self, flask_session): |
| 68 | + """Test that /version endpoint returns HTTP 200.""" |
| 69 | + response = flask_session.get(urljoin(FLASK_BASE_URL, '/version')) |
| 70 | + assert response.status_code == 200, f"Expected 200, got {response.status_code}" |
| 71 | + |
| 72 | + def test_version_endpoint_returns_json(self, flask_session): |
| 73 | + """Test that /version returns valid JSON.""" |
| 74 | + response = flask_session.get(urljoin(FLASK_BASE_URL, '/version')) |
| 75 | + assert response.headers.get('Content-Type') == 'application/json' |
| 76 | + data = response.json() |
| 77 | + assert isinstance(data, dict) |
| 78 | + |
| 79 | + def test_version_endpoint_has_version_field(self, flask_session): |
| 80 | + """Test that /version response contains 'version' field.""" |
| 81 | + response = flask_session.get(urljoin(FLASK_BASE_URL, '/version')) |
| 82 | + data = response.json() |
| 83 | + assert 'version' in data, "Response missing 'version' field" |
| 84 | + assert data['version'] is not None |
| 85 | + assert len(data['version']) > 0 |
| 86 | + |
| 87 | + def test_version_endpoint_has_build_ts_field(self, flask_session): |
| 88 | + """Test that /version response contains 'build_ts' field.""" |
| 89 | + response = flask_session.get(urljoin(FLASK_BASE_URL, '/version')) |
| 90 | + data = response.json() |
| 91 | + assert 'build_ts' in data, "Response missing 'build_ts' field" |
| 92 | + assert data['build_ts'] is not None |
| 93 | + |
| 94 | + |
| 95 | +@pytest.mark.e2e |
| 96 | +class TestGrafanaDashboardConfiguration: |
| 97 | + """Tests for Grafana dashboard version panel configuration.""" |
| 98 | + |
| 99 | + @pytest.fixture(autouse=True) |
| 100 | + def check_grafana_available(self): |
| 101 | + """Skip tests if Grafana is not available.""" |
| 102 | + if not is_service_available(GRAFANA_BASE_URL): |
| 103 | + pytest.skip(f"Grafana not available at {GRAFANA_BASE_URL}") |
| 104 | + |
| 105 | + def test_grafana_api_accessible(self, grafana_session): |
| 106 | + """Test that Grafana API is accessible with authentication.""" |
| 107 | + response = grafana_session.get(urljoin(GRAFANA_BASE_URL, '/api/health')) |
| 108 | + assert response.status_code == 200 |
| 109 | + |
| 110 | + def test_self_monitoring_dashboard_exists(self, grafana_session): |
| 111 | + """Test that Self Monitoring Dashboard exists.""" |
| 112 | + url = urljoin(GRAFANA_BASE_URL, f'/api/dashboards/uid/{SELF_MONITORING_DASHBOARD_UID}') |
| 113 | + response = grafana_session.get(url) |
| 114 | + assert response.status_code == 200, f"Dashboard not found: {response.text}" |
| 115 | + |
| 116 | + def test_dashboard_has_version_panel(self, grafana_session): |
| 117 | + """Test that the dashboard contains a version panel.""" |
| 118 | + url = urljoin(GRAFANA_BASE_URL, f'/api/dashboards/uid/{SELF_MONITORING_DASHBOARD_UID}') |
| 119 | + response = grafana_session.get(url) |
| 120 | + assert response.status_code == 200 |
| 121 | + |
| 122 | + dashboard = response.json() |
| 123 | + panels = dashboard.get('dashboard', {}).get('panels', []) |
| 124 | + |
| 125 | + # Look for version panel by title or by URL pattern |
| 126 | + version_panel = None |
| 127 | + for panel in panels: |
| 128 | + # Check by title |
| 129 | + if 'version' in panel.get('title', '').lower(): |
| 130 | + version_panel = panel |
| 131 | + break |
| 132 | + # Check by URL in datasource (for Infinity datasource) |
| 133 | + targets = panel.get('targets', []) |
| 134 | + for target in targets: |
| 135 | + url = target.get('url', '') |
| 136 | + if '/version' in url: |
| 137 | + version_panel = panel |
| 138 | + break |
| 139 | + if version_panel: |
| 140 | + break |
| 141 | + |
| 142 | + assert version_panel is not None, "No version panel found in dashboard" |
| 143 | + |
| 144 | + def test_version_panel_uses_infinity_datasource(self, grafana_session): |
| 145 | + """Test that version panel uses Infinity datasource.""" |
| 146 | + url = urljoin(GRAFANA_BASE_URL, f'/api/dashboards/uid/{SELF_MONITORING_DASHBOARD_UID}') |
| 147 | + response = grafana_session.get(url) |
| 148 | + assert response.status_code == 200 |
| 149 | + |
| 150 | + dashboard = response.json() |
| 151 | + panels = dashboard.get('dashboard', {}).get('panels', []) |
| 152 | + |
| 153 | + version_panel = None |
| 154 | + for panel in panels: |
| 155 | + targets = panel.get('targets', []) |
| 156 | + for target in targets: |
| 157 | + if '/version' in target.get('url', ''): |
| 158 | + version_panel = panel |
| 159 | + break |
| 160 | + if version_panel: |
| 161 | + break |
| 162 | + |
| 163 | + if version_panel: |
| 164 | + datasource = version_panel.get('datasource', {}) |
| 165 | + ds_type = datasource.get('type', '') |
| 166 | + assert 'infinity' in ds_type.lower(), f"Expected Infinity datasource, got: {ds_type}" |
| 167 | + |
| 168 | + def test_version_panel_queries_flask_backend(self, grafana_session): |
| 169 | + """Test that version panel is configured to query Flask backend.""" |
| 170 | + url = urljoin(GRAFANA_BASE_URL, f'/api/dashboards/uid/{SELF_MONITORING_DASHBOARD_UID}') |
| 171 | + response = grafana_session.get(url) |
| 172 | + assert response.status_code == 200 |
| 173 | + |
| 174 | + dashboard = response.json() |
| 175 | + panels = dashboard.get('dashboard', {}).get('panels', []) |
| 176 | + |
| 177 | + flask_url_found = False |
| 178 | + for panel in panels: |
| 179 | + targets = panel.get('targets', []) |
| 180 | + for target in targets: |
| 181 | + target_url = target.get('url', '') |
| 182 | + # Check for Flask backend URL (container name or localhost) |
| 183 | + if '/version' in target_url and ('flask-pgss-api' in target_url or 'localhost' in target_url): |
| 184 | + flask_url_found = True |
| 185 | + # Verify port is 8000 (gunicorn port) |
| 186 | + assert ':8000' in target_url, f"Expected port 8000 in URL, got: {target_url}" |
| 187 | + break |
| 188 | + if flask_url_found: |
| 189 | + break |
| 190 | + |
| 191 | + assert flask_url_found, "No Flask backend version endpoint URL found in dashboard panels" |
| 192 | + |
| 193 | + |
| 194 | +@pytest.mark.e2e |
| 195 | +class TestEndToEndVersionFlow: |
| 196 | + """End-to-end tests verifying version data flows through the entire stack.""" |
| 197 | + |
| 198 | + @pytest.fixture(autouse=True) |
| 199 | + def check_services_available(self): |
| 200 | + """Skip tests if required services are not available.""" |
| 201 | + if not is_service_available(FLASK_BASE_URL): |
| 202 | + pytest.skip(f"Flask backend not available at {FLASK_BASE_URL}") |
| 203 | + if not is_service_available(GRAFANA_BASE_URL): |
| 204 | + pytest.skip(f"Grafana not available at {GRAFANA_BASE_URL}") |
| 205 | + |
| 206 | + def test_version_data_is_not_unknown(self, flask_session): |
| 207 | + """Test that version data is properly populated (not 'unknown').""" |
| 208 | + response = flask_session.get(urljoin(FLASK_BASE_URL, '/version')) |
| 209 | + data = response.json() |
| 210 | + |
| 211 | + # In production, version should not be 'unknown' |
| 212 | + # In development, it might be 'unknown' but that's expected |
| 213 | + version = data.get('version', '') |
| 214 | + build_ts = data.get('build_ts', '') |
| 215 | + |
| 216 | + # At minimum, the fields should exist and have some value |
| 217 | + assert version, "version field is empty" |
| 218 | + assert build_ts, "build_ts field is empty" |
| 219 | + |
| 220 | + def test_grafana_infinity_datasource_exists(self, grafana_session): |
| 221 | + """Test that Infinity datasource is configured in Grafana.""" |
| 222 | + url = urljoin(GRAFANA_BASE_URL, '/api/datasources') |
| 223 | + response = grafana_session.get(url) |
| 224 | + assert response.status_code == 200 |
| 225 | + |
| 226 | + datasources = response.json() |
| 227 | + infinity_ds = None |
| 228 | + for ds in datasources: |
| 229 | + if 'infinity' in ds.get('type', '').lower(): |
| 230 | + infinity_ds = ds |
| 231 | + break |
| 232 | + |
| 233 | + assert infinity_ds is not None, "Infinity datasource not found in Grafana" |
| 234 | + |
| 235 | + def test_version_format_is_valid(self, flask_session): |
| 236 | + """Test that version string has a valid format.""" |
| 237 | + response = flask_session.get(urljoin(FLASK_BASE_URL, '/version')) |
| 238 | + data = response.json() |
| 239 | + |
| 240 | + version = data.get('version', '') |
| 241 | + |
| 242 | + # Accept 'unknown' for development, but if it's a version, validate format |
| 243 | + if version != 'unknown': |
| 244 | + # Version should match patterns like: 1.0.0, 0.14.0-beta.9, 1.2.3-rc.1 |
| 245 | + import re |
| 246 | + version_pattern = r'^\d+\.\d+\.\d+(-[\w.]+)?$' |
| 247 | + assert re.match(version_pattern, version), f"Invalid version format: {version}" |
| 248 | + |
| 249 | + |
| 250 | +if __name__ == '__main__': |
| 251 | + pytest.main([__file__, '-v']) |
0 commit comments