Files
redash/tests/query_runner/test_pg.py
2025-11-26 09:32:11 -05:00

54 lines
1.9 KiB
Python

from unittest import TestCase
from redash.query_runner.pg import _parse_dsn, build_schema
class TestParameters(TestCase):
def test_parse_dsn(self):
configuration = {"dsn": "application_name=redash connect_timeout=5"}
self.assertDictEqual(_parse_dsn(configuration), {"application_name": "redash", "connect_timeout": "5"})
def test_parse_dsn_not_permitted(self):
configuration = {"dsn": "password=xyz"}
self.assertRaises(ValueError, _parse_dsn, configuration)
class TestBuildSchema(TestCase):
def test_handles_dups_between_public_and_other_schemas(self):
results = {
"rows": [
{
"table_schema": "public",
"table_name": "main.users",
"column_name": "id",
},
{"table_schema": "main", "table_name": "users", "column_name": "id"},
{"table_schema": "main", "table_name": "users", "column_name": "name"},
]
}
schema = {}
build_schema(results, schema)
self.assertIn("main.users", schema.keys())
self.assertListEqual(schema["main.users"]["columns"], ["id", "name"])
self.assertIn('public."main.users"', schema.keys())
self.assertListEqual(schema['public."main.users"']["columns"], ["id"])
def test_build_schema_with_data_types(self):
results = {
"rows": [
{"table_schema": "main", "table_name": "users", "column_name": "id", "data_type": "integer"},
{"table_schema": "main", "table_name": "users", "column_name": "name", "data_type": "varchar"},
]
}
schema = {}
build_schema(results, schema)
self.assertListEqual(
schema["main.users"]["columns"], [{"name": "id", "type": "integer"}, {"name": "name", "type": "varchar"}]
)