53 lines
2.5 KiB
Python
53 lines
2.5 KiB
Python
import contextlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
MODULE_PATH = Path(__file__).resolve().parents[1] / "scripts/check-database.py"
|
|
SPEC = importlib.util.spec_from_file_location("database_check", MODULE_PATH)
|
|
CHECK = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(CHECK)
|
|
|
|
|
|
class DatabaseCheckLauncherTest(unittest.TestCase):
|
|
def test_delegates_to_java_with_credentials_only_in_environment(self):
|
|
environment = {"DB_URL": "jdbc:mariadb://test.invalid/db", "DB_USER": "fixture", "DB_PASSWORD": "test-secret"}
|
|
with patch.object(CHECK.config, "backend_jar", return_value=Path("/fixture/backend.jar")), \
|
|
patch.object(CHECK.config, "database_environment", return_value=environment), \
|
|
patch.object(CHECK.os, "chdir"), patch.object(CHECK.os, "execvpe") as execute:
|
|
self.assertEqual(CHECK.main(), 0)
|
|
|
|
command, arguments, actual_environment = execute.call_args.args
|
|
self.assertEqual(command, "java")
|
|
self.assertEqual(arguments[-1], "--database-check")
|
|
self.assertNotIn("test-secret", " ".join(arguments))
|
|
self.assertNotIn("test.invalid", " ".join(arguments))
|
|
self.assertEqual(actual_environment, environment)
|
|
|
|
def test_missing_jar_never_loads_credentials_or_starts_java(self):
|
|
output = io.StringIO()
|
|
with patch.object(CHECK.config, "backend_jar", side_effect=FileNotFoundError), \
|
|
patch.object(CHECK.config, "database_environment") as load, \
|
|
patch.object(CHECK.os, "execvpe") as execute, contextlib.redirect_stdout(output):
|
|
self.assertEqual(CHECK.main(), 2)
|
|
load.assert_not_called()
|
|
execute.assert_not_called()
|
|
self.assertFalse(json.loads(output.getvalue())["connected"])
|
|
|
|
def test_configuration_error_is_sanitized_json(self):
|
|
output = io.StringIO()
|
|
with patch.object(CHECK.config, "backend_jar", return_value=Path("/fixture/backend.jar")), \
|
|
patch.object(CHECK.config, "database_environment", side_effect=ValueError("secret=do-not-expose")), \
|
|
patch.object(CHECK.os, "execvpe") as execute, contextlib.redirect_stdout(output):
|
|
self.assertEqual(CHECK.main(), 2)
|
|
execute.assert_not_called()
|
|
self.assertNotIn("do-not-expose", output.getvalue())
|
|
self.assertEqual(json.loads(output.getvalue())["writes"], 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|