97 lines
3.3 KiB
Python
97 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from sqlalchemy import text
|
|
from sqlalchemy.dialects import postgresql
|
|
|
|
from govoplan_reporting.backend.postgres_planner import compile_postgres_query
|
|
from govoplan_reporting.backend.schemas import (
|
|
DatasetDefinition,
|
|
ReportQuery,
|
|
SemanticModelDefinition,
|
|
)
|
|
|
|
|
|
class PostgresBindNameTests(unittest.TestCase):
|
|
def test_allowed_measure_key_punctuation_never_becomes_bind_parameter_syntax(
|
|
self,
|
|
) -> None:
|
|
dataset = DatasetDefinition(
|
|
source_kind="static",
|
|
source_ref="fixture",
|
|
static_rows=[{"value": 10}],
|
|
purpose="Bound parameter fixture",
|
|
)
|
|
semantic = SemanticModelDefinition.model_validate(
|
|
{
|
|
"dataset_id": "fixture",
|
|
"dataset_revision": 1,
|
|
"measures": [
|
|
{
|
|
"key": "base",
|
|
"label": "Base",
|
|
"aggregation": "sum",
|
|
"field": "value",
|
|
},
|
|
{
|
|
"key": "extra-cost",
|
|
"label": "Extra",
|
|
"aggregation": "calculated",
|
|
"expression": {
|
|
"op": "add",
|
|
"args": [
|
|
{"op": "measure", "ref": "base"},
|
|
{"op": "literal", "value": 5},
|
|
],
|
|
},
|
|
},
|
|
{
|
|
"key": "tax.factor",
|
|
"label": "Tax",
|
|
"aggregation": "calculated",
|
|
"expression": {
|
|
"op": "multiply",
|
|
"args": [
|
|
{"op": "measure", "ref": "base"},
|
|
{"op": "literal", "value": 9},
|
|
],
|
|
},
|
|
},
|
|
{
|
|
"key": "grand-total",
|
|
"label": "Total",
|
|
"aggregation": "calculated",
|
|
"expression": {
|
|
"op": "add",
|
|
"args": [
|
|
{"op": "measure", "ref": "extra-cost"},
|
|
{"op": "measure", "ref": "tax.factor"},
|
|
],
|
|
},
|
|
},
|
|
],
|
|
}
|
|
)
|
|
plan = compile_postgres_query(
|
|
dataset,
|
|
semantic,
|
|
ReportQuery(measures=["extra-cost", "tax.factor", "grand-total"]),
|
|
)
|
|
binds = text(plan.sql).compile(dialect=postgresql.dialect()).params
|
|
self.assertEqual(
|
|
set(plan.parameters) | {"rows_json", "result_limit", "result_offset"},
|
|
set(binds),
|
|
)
|
|
self.assertTrue(
|
|
all("-" not in name and "." not in name for name in plan.parameters)
|
|
)
|
|
self.assertIn('AS "extra-cost"', plan.sql)
|
|
self.assertIn('AS "tax.factor"', plan.sql)
|
|
self.assertEqual(2, list(plan.parameters.values()).count(5))
|
|
self.assertEqual(2, list(plan.parameters.values()).count(9))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|