169 lines
6.4 KiB
Python
169 lines
6.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Exercise destructive-import guards without connecting to any real database."""
|
|
import json
|
|
import os
|
|
import secrets
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
SCRIPT = Path(__file__).resolve().parents[1] / "docker/init-db.sh"
|
|
BASE = ["in_daper", "in_soal", "in_jawab"]
|
|
ALL = BASE + ["rec_admin", "rec_exam_attempt", "rec_exam_event", "rec_candidate_login_guard"]
|
|
|
|
FAKE_MYSQL = r'''#!/usr/bin/env python3
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
path = Path(os.environ["FAKE_MYSQL_STATE"])
|
|
state = json.loads(path.read_text())
|
|
args = sys.argv[1:]
|
|
query = args[args.index("-e") + 1] if "-e" in args else None
|
|
if query is None:
|
|
sys.stdin.read()
|
|
state["imports"] = state.get("imports", 0) + 1
|
|
state["tables"] += state.get("import_tables", [])
|
|
path.write_text(json.dumps(state))
|
|
if state.get("fail_import"):
|
|
print("Synthetic import failure", file=sys.stderr)
|
|
sys.exit(1)
|
|
elif "information_schema.tables" in query:
|
|
tables = state["tables"]
|
|
if "table_name = 'rec_bootstrap_state'" in query:
|
|
print(int("rec_bootstrap_state" in tables))
|
|
elif "table_name IN" in query:
|
|
names = re.findall(r"'([^']+)'", query.split("table_name IN", 1)[1])
|
|
print(sum(name in tables for name in names))
|
|
else:
|
|
print(len(tables))
|
|
elif query.startswith("SELECT status"):
|
|
print(state.get("status", ""))
|
|
elif query.startswith("CREATE TABLE rec_bootstrap_state"):
|
|
if "rec_bootstrap_state" in state["tables"]:
|
|
sys.exit(1)
|
|
state["tables"].append("rec_bootstrap_state")
|
|
state["status"] = "importing"
|
|
path.write_text(json.dumps(state))
|
|
elif query.startswith("UPDATE rec_bootstrap_state"):
|
|
state["status"] = "imported"
|
|
path.write_text(json.dumps(state))
|
|
else:
|
|
print("Unexpected query: " + query, file=sys.stderr)
|
|
sys.exit(1)
|
|
'''
|
|
|
|
|
|
class InitializationTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
client = self.root / "mysql"
|
|
client.write_text(FAKE_MYSQL)
|
|
client.chmod(0o700)
|
|
self.state_path = self.root / "state.json"
|
|
self.dump = self.root / "initial.sql"
|
|
self.dump.write_text("-- Synthetic SQL; never loads project personal data.\n")
|
|
self.env = dict(os.environ, PATH=f"{self.root}:{os.environ['PATH']}",
|
|
MYSQL_HOST="test-db", MYSQL_USER="test-user",
|
|
MYSQL_DATABASE="test-db", MYSQL_PASSWORD=secrets.token_hex(16),
|
|
FAKE_MYSQL_STATE=str(self.state_path), DB_INIT_SQL_FILE=str(self.dump))
|
|
|
|
def state(self):
|
|
try:
|
|
return json.loads(self.state_path.read_text())
|
|
except (OSError, ValueError) as error:
|
|
self.fail(f"Invalid mock database state: {error}")
|
|
|
|
def configure(self, tables=None, **options):
|
|
self.state_path.write_text(json.dumps(dict(tables=tables or [], import_tables=ALL, **options)))
|
|
|
|
def run_init(self):
|
|
return subprocess.run(["sh", str(SCRIPT)], env=self.env, capture_output=True, text=True)
|
|
|
|
def test_empty_database_imports_once(self):
|
|
self.configure()
|
|
first = self.run_init()
|
|
self.assertEqual(first.returncode, 0, first.stderr)
|
|
self.assertEqual(self.state()["status"], "imported")
|
|
second = self.run_init()
|
|
self.assertEqual(second.returncode, 0, second.stderr)
|
|
self.assertEqual(self.state()["imports"], 1)
|
|
|
|
def test_existing_complete_database_skips_import(self):
|
|
self.configure(ALL)
|
|
result = self.run_init()
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertNotIn("imports", self.state())
|
|
self.assertEqual(self.state()["tables"], ALL)
|
|
|
|
def test_partial_database_refuses_import(self):
|
|
self.configure(BASE)
|
|
self.assertNotEqual(self.run_init().returncode, 0)
|
|
self.assertNotIn("imports", self.state())
|
|
|
|
def test_unrelated_database_refuses_import(self):
|
|
self.configure(["unrelated_table"])
|
|
self.assertNotEqual(self.run_init().returncode, 0)
|
|
self.assertNotIn("imports", self.state())
|
|
|
|
def test_failed_import_blocks_automatic_retry(self):
|
|
self.configure(fail_import=True)
|
|
self.assertNotEqual(self.run_init().returncode, 0)
|
|
self.assertEqual(self.state()["status"], "importing")
|
|
self.assertNotEqual(self.run_init().returncode, 0)
|
|
self.assertEqual(self.state()["imports"], 1)
|
|
|
|
def test_missing_dump_leaves_database_untouched(self):
|
|
self.configure()
|
|
self.dump.unlink()
|
|
self.assertNotEqual(self.run_init().returncode, 0)
|
|
self.assertEqual(self.state()["tables"], [])
|
|
|
|
def test_empty_dump_leaves_database_untouched(self):
|
|
self.configure()
|
|
self.dump.write_text("")
|
|
self.assertNotEqual(self.run_init().returncode, 0)
|
|
self.assertEqual(self.state()["tables"], [])
|
|
|
|
def test_dump_cannot_replace_the_safety_marker(self):
|
|
self.configure()
|
|
self.dump.write_text("DROP TABLE IF EXISTS `REC_BOOTSTRAP_STATE`;\n")
|
|
self.assertNotEqual(self.run_init().returncode, 0)
|
|
self.assertEqual(self.state()["tables"], [])
|
|
self.assertNotIn("imports", self.state())
|
|
|
|
def test_dump_missing_base_tables_is_not_marked_imported(self):
|
|
self.configure()
|
|
state = self.state()
|
|
state["import_tables"] = ["in_daper"]
|
|
self.state_path.write_text(json.dumps(state))
|
|
self.assertNotEqual(self.run_init().returncode, 0)
|
|
self.assertEqual(self.state()["status"], "importing")
|
|
self.assertNotEqual(self.run_init().returncode, 0)
|
|
self.assertEqual(self.state()["imports"], 1)
|
|
|
|
def test_imported_marker_with_missing_tables_is_refused(self):
|
|
self.configure(["rec_bootstrap_state", "in_daper"], status="imported")
|
|
self.assertNotEqual(self.run_init().returncode, 0)
|
|
self.assertNotIn("imports", self.state())
|
|
|
|
def test_marker_with_missing_status_is_refused(self):
|
|
self.configure(["rec_bootstrap_state"])
|
|
self.assertNotEqual(self.run_init().returncode, 0)
|
|
self.assertNotIn("imports", self.state())
|
|
|
|
def test_missing_password_is_refused_before_access(self):
|
|
self.configure()
|
|
del self.env["MYSQL_PASSWORD"]
|
|
self.assertNotEqual(self.run_init().returncode, 0)
|
|
self.assertEqual(self.state()["tables"], [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|