|
| 1 | +# backend/tests/test_app.py |
| 2 | +# |
| 3 | +# Tests for the Flask routes in app.py. |
| 4 | +# |
| 5 | +# The Athena introspection layer (get_options) is mocked to return a fixed |
| 6 | +# schema, so these tests run anywhere without Docker. |
| 7 | +# |
| 8 | +# What is tested: |
| 9 | +# - /api/health returns the right shape |
| 10 | +# - /api/schema returns a list whose entries match BLOCK_TREE |
| 11 | +# - /api/export-yaml returns valid YAML containing the submitted config |
| 12 | + |
| 13 | +import json |
| 14 | +import pytest |
| 15 | +import yaml |
| 16 | +from unittest.mock import patch |
| 17 | + |
| 18 | +# Adjust path so we can import app from the backend directory |
| 19 | +import sys, os |
| 20 | +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) |
| 21 | + |
| 22 | + |
| 23 | +# Mock get_options to return a minimal option list — avoids needing Athena |
| 24 | +MOCK_OPTIONS = [ |
| 25 | + {"name": "containerName", "type": "str", "default": "AnaJets", |
| 26 | + "info": "The container name", "required": False, "noneAction": "ignore"}, |
| 27 | +] |
| 28 | + |
| 29 | +@pytest.fixture() |
| 30 | +def client(): |
| 31 | + """Return a Flask test client with a pre-built schema cache.""" |
| 32 | + with patch("introspect.get_options", return_value=MOCK_OPTIONS): |
| 33 | + import app as flask_app |
| 34 | + flask_app.app.config["TESTING"] = True |
| 35 | + # Force schema rebuild with the mock in place |
| 36 | + flask_app._schema_cache = None |
| 37 | + with flask_app.app.test_client() as c: |
| 38 | + # Trigger the schema build |
| 39 | + c.get("/api/schema") |
| 40 | + yield c |
| 41 | + |
| 42 | + |
| 43 | +# ───────────────────────────────────────────────────────────────────────────── |
| 44 | +# /api/health |
| 45 | +# ───────────────────────────────────────────────────────────────────────────── |
| 46 | + |
| 47 | +class TestHealth: |
| 48 | + |
| 49 | + def test_returns_200(self, client): |
| 50 | + r = client.get("/api/health") |
| 51 | + assert r.status_code == 200 |
| 52 | + |
| 53 | + def test_contains_status_ok(self, client): |
| 54 | + data = r = client.get("/api/health").get_json() |
| 55 | + assert data["status"] == "ok" |
| 56 | + |
| 57 | + def test_contains_app_version(self, client): |
| 58 | + data = client.get("/api/health").get_json() |
| 59 | + assert "app_version" in data |
| 60 | + assert data["app_version"] # non-empty |
| 61 | + |
| 62 | + def test_contains_athena_key(self, client): |
| 63 | + data = client.get("/api/health").get_json() |
| 64 | + assert "athena" in data |
| 65 | + |
| 66 | + |
| 67 | +# ───────────────────────────────────────────────────────────────────────────── |
| 68 | +# /api/schema |
| 69 | +# ───────────────────────────────────────────────────────────────────────────── |
| 70 | + |
| 71 | +class TestSchema: |
| 72 | + from block_schema import BLOCK_TREE as _BLOCK_TREE |
| 73 | + |
| 74 | + def test_returns_200(self, client): |
| 75 | + assert client.get("/api/schema").status_code == 200 |
| 76 | + |
| 77 | + def test_returns_list(self, client): |
| 78 | + data = client.get("/api/schema").get_json() |
| 79 | + assert isinstance(data, list) |
| 80 | + |
| 81 | + def test_length_matches_block_tree(self, client): |
| 82 | + from block_schema import BLOCK_TREE |
| 83 | + data = client.get("/api/schema").get_json() |
| 84 | + assert len(data) == len(BLOCK_TREE) |
| 85 | + |
| 86 | + def test_each_entry_has_required_fields(self, client): |
| 87 | + data = client.get("/api/schema").get_json() |
| 88 | + for block in data: |
| 89 | + assert "name" in block |
| 90 | + assert "label" in block |
| 91 | + assert "options" in block |
| 92 | + assert "sub_blocks" in block |
| 93 | + assert isinstance(block["options"], list) |
| 94 | + assert isinstance(block["sub_blocks"], list) |
| 95 | + |
| 96 | + def test_options_have_required_fields(self, client): |
| 97 | + data = client.get("/api/schema").get_json() |
| 98 | + for block in data: |
| 99 | + for opt in block["options"]: |
| 100 | + assert "name" in opt |
| 101 | + assert "type" in opt |
| 102 | + assert "default" in opt |
| 103 | + assert "info" in opt |
| 104 | + assert "required" in opt |
| 105 | + assert "noneAction" in opt |
| 106 | + |
| 107 | + def test_block_names_match_block_tree(self, client): |
| 108 | + from block_schema import BLOCK_TREE |
| 109 | + schema_names = {b["name"] for b in client.get("/api/schema").get_json()} |
| 110 | + tree_names = {b["name"] for b in BLOCK_TREE} |
| 111 | + assert schema_names == tree_names |
| 112 | + |
| 113 | + |
| 114 | +# ───────────────────────────────────────────────────────────────────────────── |
| 115 | +# /api/export-yaml |
| 116 | +# ───────────────────────────────────────────────────────────────────────────── |
| 117 | + |
| 118 | +class TestExportYaml: |
| 119 | + |
| 120 | + def _post(self, client, config, filename="out.yaml"): |
| 121 | + return client.post( |
| 122 | + "/api/export-yaml", |
| 123 | + data=json.dumps({"config": config, "filename": filename}), |
| 124 | + content_type="application/json", |
| 125 | + ) |
| 126 | + |
| 127 | + def test_returns_200(self, client): |
| 128 | + assert self._post(client, {"Jets": [{"containerName": "AnaJets"}]}).status_code == 200 |
| 129 | + |
| 130 | + def test_content_type_is_yaml(self, client): |
| 131 | + r = self._post(client, {"Jets": [{"containerName": "AnaJets"}]}) |
| 132 | + assert "yaml" in r.content_type |
| 133 | + |
| 134 | + def test_output_is_valid_yaml(self, client): |
| 135 | + r = self._post(client, {"Jets": [{"containerName": "AnaJets"}]}) |
| 136 | + parsed = yaml.safe_load(r.data) |
| 137 | + assert parsed is not None |
| 138 | + |
| 139 | + def test_output_contains_submitted_key(self, client): |
| 140 | + r = self._post(client, {"Jets": [{"containerName": "AnaJets"}]}) |
| 141 | + parsed = yaml.safe_load(r.data) |
| 142 | + assert "Jets" in parsed |
| 143 | + |
| 144 | + def test_empty_config_is_valid_yaml(self, client): |
| 145 | + r = self._post(client, {}) |
| 146 | + assert r.status_code == 200 |
| 147 | + # Should not raise |
| 148 | + yaml.safe_load(r.data) |
| 149 | + |
| 150 | + def test_content_disposition_uses_filename(self, client): |
| 151 | + r = self._post(client, {}, filename="my_config.yaml") |
| 152 | + assert "my_config.yaml" in r.headers.get("Content-Disposition", "") |
0 commit comments