241 lines
7.8 KiB
Python
241 lines
7.8 KiB
Python
"""Safety checks for the disposable migration rehearsal. No production database."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SCRIPT = ROOT / "deploy" / "rehearse-production-migrations.sh"
|
|
|
|
|
|
def resolve_bash() -> str:
|
|
found = shutil.which("bash")
|
|
git_bash = Path(r"D:\Program Files\Git\bin\bash.exe")
|
|
if os.name == "nt" and git_bash.exists():
|
|
lowered = (found or "").lower()
|
|
if not found or "system32" in lowered or "windowsapps" in lowered:
|
|
return str(git_bash)
|
|
if found:
|
|
return found
|
|
if git_bash.exists():
|
|
return str(git_bash)
|
|
return "bash"
|
|
|
|
|
|
BASH = resolve_bash()
|
|
|
|
|
|
def run(env: dict[str, str]) -> subprocess.CompletedProcess[str]:
|
|
full = os.environ.copy()
|
|
full.update(env)
|
|
return subprocess.run(
|
|
[BASH, str(SCRIPT)],
|
|
cwd=ROOT,
|
|
env=full,
|
|
stdin=subprocess.DEVNULL,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
)
|
|
|
|
|
|
def test_refuses_a_non_production_path() -> None:
|
|
completed = run({"DEPLOY_PATH": "/tmp/not-production"})
|
|
assert completed.returncode == 1
|
|
assert "unexpected production path" in completed.stderr
|
|
|
|
|
|
def test_refuses_the_production_database_name() -> None:
|
|
completed = run(
|
|
{
|
|
"JYOTISHA_REHEARSAL_LOCAL": "1",
|
|
"JYOTISHA_REHEARSAL_DATABASE_URL": "postgresql://postgres:secret-value@127.0.0.1:1/jyotisha",
|
|
}
|
|
)
|
|
assert completed.returncode == 1
|
|
assert "refusing to touch the production database" in completed.stderr
|
|
assert "secret-value" not in completed.stdout
|
|
assert "secret-value" not in completed.stderr
|
|
|
|
|
|
def test_plan_keeps_the_stop_file_out_of_the_first_batch() -> None:
|
|
completed = run(
|
|
{
|
|
"JYOTISHA_REHEARSAL_LOCAL": "1",
|
|
"JYOTISHA_REHEARSAL_PLAN": "1",
|
|
"JYOTISHA_REHEARSAL_DATABASE_URL": "postgresql://postgres:secret-value@127.0.0.1:1/rehearse_plan",
|
|
}
|
|
)
|
|
assert completed.returncode == 0, completed.stderr
|
|
text = completed.stdout
|
|
assert "rehearsal_plan=1" in text
|
|
assert "database_name=rehearse_plan" in text
|
|
assert "stop_before=20260925030000_chart_subject_trust_and_limit.sql" in text
|
|
assert "stop_in_before=no" in text
|
|
values = dict(line.split("=", 1) for line in text.splitlines() if "=" in line)
|
|
assert int(values["batch_before"]) < int(values["batch_all"])
|
|
assert "secret-value" not in completed.stdout
|
|
assert "secret-value" not in completed.stderr
|
|
|
|
|
|
def test_local_url_with_query_is_refused() -> None:
|
|
completed = run(
|
|
{
|
|
"JYOTISHA_REHEARSAL_LOCAL": "1",
|
|
"JYOTISHA_REHEARSAL_DATABASE_URL": "postgresql://postgres:secret-value@127.0.0.1:1/rehearse_plan?database=jyotisha",
|
|
}
|
|
)
|
|
assert completed.returncode == 1
|
|
assert "query parameters" in completed.stderr
|
|
assert "secret-value" not in completed.stdout
|
|
assert "secret-value" not in completed.stderr
|
|
|
|
|
|
def bash_path(path: Path) -> str:
|
|
if os.name != "nt":
|
|
return str(path)
|
|
completed = subprocess.run(
|
|
[BASH, "-lc", 'cygpath -u "$1"', "cygpath", str(path)],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
return completed.stdout.strip()
|
|
|
|
|
|
def test_host_mode_does_not_call_host_node_and_failed_cleanup_is_nonzero(tmp_path: Path) -> None:
|
|
bin_dir = tmp_path / "bin"
|
|
host = tmp_path / "host"
|
|
(host / ".state").mkdir(parents=True)
|
|
(host / "backups").mkdir()
|
|
dump = host / "backups" / "production-pre-migration-20261014T000000Z.dump.enc"
|
|
dump.write_bytes(b"not-a-real-dump")
|
|
(host / ".env.production.database").write_text(
|
|
"\n".join(
|
|
[
|
|
"POSTGRES_DB=jyotisha",
|
|
"POSTGRES_USER=postgres",
|
|
"POSTGRES_PASSWORD=secret-value",
|
|
"PRODUCTION_BACKUP_ENCRYPTION_KEY=not-a-real-key",
|
|
"",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
newline="\n",
|
|
)
|
|
docker_log = tmp_path / "docker.log"
|
|
node_log = tmp_path / "node.log"
|
|
bin_dir.mkdir()
|
|
(bin_dir / "docker").write_text(
|
|
"\n".join(
|
|
[
|
|
"#!/bin/bash",
|
|
'printf "%s\\n" "$*" >> "${DOCKER_LOG:?}"',
|
|
'case "$*" in',
|
|
" *exec\\ -i*|*-i\\ *) cat >/dev/null || true ;;",
|
|
"esac",
|
|
'joined="$*"',
|
|
'case "$joined" in',
|
|
' *ps\\ *) echo container1; exit 0 ;;',
|
|
"esac",
|
|
'case "$joined" in',
|
|
" *'{{.Image}}'*) echo sha256:0000000000000000000000000000000000000000000000000000000000000000; exit 0 ;;",
|
|
"esac",
|
|
'case "$joined" in',
|
|
" *inspect*) echo jyotisha-net; exit 0 ;;",
|
|
"esac",
|
|
'case "$joined" in',
|
|
" *DROP\\ DATABASE*) exit 1 ;;",
|
|
"esac",
|
|
'case "$joined" in',
|
|
" *has_table_privilege*)",
|
|
" printf '%s\\n' 'authenticated public.profiles true' 'authenticated public.chat_sessions true' 'service_role public.chart_profiles true'",
|
|
" exit 0",
|
|
" ;;",
|
|
"esac",
|
|
'case "$joined" in',
|
|
" *count*) echo 0; exit 0 ;;",
|
|
"esac",
|
|
"exit 0",
|
|
"",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
newline="\n",
|
|
)
|
|
(bin_dir / "node").write_text(
|
|
"\n".join(
|
|
[
|
|
"#!/bin/bash",
|
|
'printf "%s\\n" "$*" >> "${NODE_LOG:?}"',
|
|
"exit 9",
|
|
"",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
newline="\n",
|
|
)
|
|
(bin_dir / "openssl").write_text("#!/bin/bash\nexit 0\n", encoding="utf-8", newline="\n")
|
|
unix_bin = bash_path(bin_dir)
|
|
python_bin = bash_path(Path(sys.executable))
|
|
(bin_dir / "python3").write_text(
|
|
f"#!/bin/bash\nexec '{python_bin}' \"$@\"\n",
|
|
encoding="utf-8",
|
|
newline="\n",
|
|
)
|
|
subprocess.run(
|
|
[BASH, "-lc", f'chmod +x "{unix_bin}/docker" "{unix_bin}/node" "{unix_bin}/openssl" "{unix_bin}/python3"'],
|
|
check=True,
|
|
)
|
|
full = os.environ.copy()
|
|
full.update(
|
|
{
|
|
"JYOTISHA_REHEARSAL_FAKE_HOST": "1",
|
|
"DEPLOY_PATH": bash_path(host),
|
|
"RECOVERY_DUMP": bash_path(dump),
|
|
"DOCKER_LOG": bash_path(docker_log),
|
|
"NODE_LOG": bash_path(node_log),
|
|
}
|
|
)
|
|
completed = subprocess.run(
|
|
[
|
|
BASH,
|
|
"-lc",
|
|
'export PATH="$1:$PATH"; exec bash "$2"',
|
|
"bash",
|
|
unix_bin,
|
|
bash_path(SCRIPT),
|
|
],
|
|
cwd=ROOT,
|
|
env=full,
|
|
stdin=subprocess.DEVNULL,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
)
|
|
assert completed.returncode != 0, completed.stdout + completed.stderr
|
|
assert "演练库未删除，请手动 DROP DATABASE rehearse_" in completed.stderr
|
|
assert "secret-value" not in completed.stdout
|
|
assert "secret-value" not in completed.stderr
|
|
assert not node_log.exists()
|
|
docker_text = docker_log.read_text(encoding="utf-8")
|
|
assert "--entrypoint node" in docker_text
|
|
assert "DROP DATABASE" in docker_text
|
|
|
|
|
|
def test_restore_keeps_backup_owners_and_compares_grants() -> None:
|
|
text = SCRIPT.read_text(encoding="utf-8")
|
|
assert "--no-owner" not in text
|
|
assert "ALTER TYPE" not in text
|
|
assert "typtype" not in text
|
|
assert "revoked_grants" in text
|
|
assert "chart_profiles_service_role_select_missing" in text
|
|
assert "role%3Dschema_owner" in text
|
|
assert "grant connect, create on database" in text
|