Files
telecom-cdr-pipeline-ai/tests/test_bootstrap.py

20 lines
833 B
Python

import unittest
from unittest.mock import Mock
from app.common.bootstrap import identifier, verify_table
class TestBootstrap(unittest.TestCase):
def test_identifiers_cannot_inject_sql(self):
self.assertEqual(identifier("cdr_records"), "cdr_records")
for value in ("a; DROP TABLE b", "a.b", "", "1table"):
with self.subTest(value=value), self.assertRaises(ValueError):
identifier(value)
def test_old_sorting_key_fails_without_mutating_data(self):
client = Mock()
client.query.return_value.result_rows = [("ReplacingMergeTree", "service_type, radio_access_type, cell_id, session_start_time, msisdn")]
with self.assertRaisesRegex(RuntimeError, "reset"):
verify_table(client, "telecom", "cdr_records")
client.command.assert_not_called()