70 lines
3.5 KiB
Python
70 lines
3.5 KiB
Python
import io
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import psycopg
|
|
|
|
from car_traker.database import database_error_hint, main
|
|
|
|
|
|
class DatabaseCliTests(unittest.TestCase):
|
|
def test_safe_error_classification(self):
|
|
cases = (
|
|
(psycopg.OperationalError('no pg_hba.conf entry for host "private-host"'), "pg_hba.conf"),
|
|
(psycopg.OperationalError("fe_sendauth: no password supplied"), "PGPASSFILE"),
|
|
(psycopg.OperationalError("password authentication failed"), "rejected authentication"),
|
|
(psycopg.OperationalError("could not translate host name"), "DNS/VPN"),
|
|
(psycopg.OperationalError("Connection refused"), "listen_addresses"),
|
|
(psycopg.OperationalError("timeout expired"), "route/VPN"),
|
|
(psycopg.OperationalError("certificate verify failed"), "PGSSLROOTCERT"),
|
|
(psycopg.OperationalError("server does not support SSL"), "server TLS"),
|
|
(psycopg.errors.InvalidCatalogName("private-db"), "PGDATABASE"),
|
|
(psycopg.errors.InsufficientPrivilege("private-table"), "privileges"),
|
|
(psycopg.errors.UndefinedTable("private-table"), "migrate"),
|
|
)
|
|
for exc, expected in cases:
|
|
with self.subTest(error=type(exc).__name__, expected=expected):
|
|
hint = database_error_hint(exc)
|
|
self.assertIn(expected, hint)
|
|
self.assertNotIn("private-", hint)
|
|
|
|
def test_unknown_error_does_not_expose_details(self):
|
|
hint = database_error_hint(psycopg.OperationalError("password=synthetic-secret source=private-text"))
|
|
self.assertNotIn("synthetic-secret", hint)
|
|
self.assertNotIn("private-text", hint)
|
|
self.assertIn("server logs", hint)
|
|
|
|
def test_check_only_runs_select(self):
|
|
connection = MagicMock()
|
|
with patch.dict("os.environ", {"PGHOST": "synthetic-host"}, clear=True), patch(
|
|
"sys.argv", ["car-traker-db", "check"]
|
|
), patch("car_traker.database.psycopg.connect") as connect, patch(
|
|
"sys.stdout", new_callable=io.StringIO
|
|
) as output, patch("car_traker.database.migrate") as migrate:
|
|
connect.return_value.__enter__.return_value = connection
|
|
self.assertEqual(main(), 0)
|
|
connection.execute.assert_called_once_with("SELECT 1")
|
|
migrate.assert_not_called()
|
|
self.assertIn("no migrations or imports", output.getvalue())
|
|
|
|
def test_hba_failure_is_actionable_without_echoing_credentials(self):
|
|
exc = psycopg.OperationalError('no pg_hba.conf entry for host "private-host"; password=synthetic-secret')
|
|
with patch.dict("os.environ", {"PGHOST": "synthetic-host"}, clear=True), patch(
|
|
"sys.argv", ["car-traker-db", "check"]
|
|
), patch("car_traker.database.psycopg.connect", side_effect=exc), patch(
|
|
"sys.stderr", new_callable=io.StringIO
|
|
) as output:
|
|
self.assertEqual(main(), 1)
|
|
self.assertIn("pg_hba.conf", output.getvalue())
|
|
self.assertNotIn("synthetic-secret", output.getvalue())
|
|
self.assertNotIn("private-host", output.getvalue())
|
|
|
|
def test_no_implicit_local_connection(self):
|
|
with patch.dict("os.environ", {}, clear=True), patch(
|
|
"sys.argv", ["car-traker-db", "check"]
|
|
), patch("sys.stderr", new_callable=io.StringIO), patch(
|
|
"car_traker.database.psycopg.connect"
|
|
) as connect, self.assertRaises(SystemExit):
|
|
main()
|
|
connect.assert_not_called()
|