79 lines
3.0 KiB
Python
79 lines
3.0 KiB
Python
"""Check venue cleanup using a fake Docker; never remove real data."""
|
|
|
|
from pathlib import Path
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
|
|
|
|
workspace = Path(__file__).resolve().parents[2]
|
|
with tempfile.TemporaryDirectory(prefix="venue-clean-") as directory:
|
|
root = Path(directory)
|
|
venue = root / "venue"
|
|
venue.mkdir()
|
|
shutil.copy(workspace / "venue/clean.sh", venue / "clean.sh")
|
|
protected = [".env", "certificates/privkey.pem", "cudy/password", "backend/app",
|
|
"frontend/dist/index.html"]
|
|
for name in protected:
|
|
path = venue / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text("preserve me")
|
|
bin_dir = root / "bin"
|
|
bin_dir.mkdir()
|
|
docker = bin_dir / "docker"
|
|
docker.write_text('''#!/usr/bin/env python3
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
args = sys.argv[1:]
|
|
with open(os.environ["TEST_LOG"], "a") as log:
|
|
log.write(json.dumps(args) + "\\n")
|
|
if os.environ.get("TEST_FAIL") == args[0] or os.environ.get("TEST_FAIL") == " ".join(args[:2]):
|
|
sys.exit(1)
|
|
state = Path(os.environ["TEST_VOLUMES"])
|
|
names = state.read_text().splitlines()
|
|
if args[:2] == ["volume", "ls"]:
|
|
print("\\n".join(names))
|
|
elif args[:2] == ["volume", "rm"]:
|
|
assert args[2] in names
|
|
names.remove(args[2])
|
|
state.write_text("\\n".join(names))
|
|
''')
|
|
docker.chmod(0o755)
|
|
log = root / "commands"
|
|
state = root / "volumes"
|
|
targets = ["evening-detective-local_postgres_data", "evening-detective-local_rustfs_data"]
|
|
unrelated = ["dev_postgres_data", "evening-detective-local_postgres_data-backup"]
|
|
env = {**os.environ, "PATH": str(bin_dir) + os.pathsep + os.environ["PATH"],
|
|
"TEST_LOG": str(log), "TEST_VOLUMES": str(state)}
|
|
|
|
def check(fail=""):
|
|
log.write_text("")
|
|
result = subprocess.run(["bash", str(venue / "clean.sh")],
|
|
env={**env, "TEST_FAIL": fail}, capture_output=True, text=True)
|
|
assert (result.returncode == 0) == (not fail), result.stderr
|
|
for name in protected:
|
|
assert (venue / name).read_text() == "preserve me"
|
|
return [json.loads(line) for line in log.read_text().splitlines()]
|
|
|
|
state.write_text("\n".join(targets + unrelated))
|
|
commands = check()
|
|
assert commands == [
|
|
["compose", "-f", "docker-compose.yml", "-f", "docker-compose.https.yml",
|
|
"down", "--volumes", "--remove-orphans"],
|
|
["volume", "ls", "--format", "{{.Name}}"],
|
|
*[["volume", "rm", name] for name in targets],
|
|
]
|
|
assert state.read_text().splitlines() == unrelated
|
|
assert len(check()) == 2 # Repeated cleanup skips absent volumes.
|
|
state.write_text("\n".join(targets + unrelated))
|
|
assert len(check("compose")) == 1
|
|
assert len(check("volume ls")) == 2
|
|
assert len(check("volume rm")) == 3
|
|
assert state.read_text().splitlines() == targets + unrelated
|
|
|
|
print("PASS: cleanup order, repeat calls, Docker failures and preserved files/volumes")
|