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()