109 lines
3.3 KiB
Python
109 lines
3.3 KiB
Python
import json
|
|
import os
|
|
import subprocess
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
from unittest import TestCase
|
|
|
|
from helpers import DirectoryDescriptor, scan_dir
|
|
from helpers.datman_helpers import set_up_simple_datman
|
|
from helpers.yama_helpers import set_up_simple_yama
|
|
|
|
|
|
class TestMysqlHelper(TestCase):
|
|
def setUp(self):
|
|
if "TEST_MYSQL" not in os.environ:
|
|
self.skipTest(
|
|
"TEST_MYSQL environment variable not set. "
|
|
"Should be set to a MySQL host, database,"
|
|
" user combination, comma-separated."
|
|
)
|
|
|
|
def test_helper_fails_on_bad_connection(self):
|
|
proc = subprocess.Popen("datman-helper-mysql-backup", stdin=subprocess.PIPE)
|
|
proc.stdin.write(
|
|
json.dumps(
|
|
{"database": "mydatabase", "host": "notmyhost", "user": "bobjones"}
|
|
).encode()
|
|
)
|
|
proc.stdin.close()
|
|
self.assertNotEqual(proc.wait(), 0)
|
|
|
|
def test_helper_succeeds_on_correct_config(self):
|
|
my_host, my_database, my_user = os.environ["TEST_MYSQL"].split(",")
|
|
|
|
proc = subprocess.Popen(
|
|
"datman-helper-mysql-backup",
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
)
|
|
proc.stdin.write(
|
|
json.dumps(
|
|
{
|
|
"database": my_database or None,
|
|
"host": my_host or None,
|
|
"user": my_user or None,
|
|
}
|
|
).encode()
|
|
)
|
|
proc.stdin.close()
|
|
stdout = proc.stdout.read()
|
|
self.assertEqual(proc.wait(), 0)
|
|
self.assertIn(b"CREATE TABLE", stdout)
|
|
self.assertIn(b"INSERT INTO", stdout)
|
|
print(stdout) # TODO
|
|
|
|
def test_backup_and_extraction(self):
|
|
td = TemporaryDirectory("test_my_bae")
|
|
tdpath = Path(td.name)
|
|
|
|
my_host, my_database, my_user = os.environ["TEST_MYSQL"].split(",")
|
|
|
|
datman_path = tdpath.joinpath("datman")
|
|
yama_path = datman_path.joinpath("main")
|
|
|
|
set_up_simple_datman(
|
|
datman_path,
|
|
custom_extra_test=f"""
|
|
[sources.mysql123]
|
|
helper = "mysql"
|
|
label = "precious"
|
|
kind = {{ stdout = "mysql123.sql" }}
|
|
database = "{my_database}"
|
|
host = "{my_host}"
|
|
user = "{my_user}"
|
|
""",
|
|
)
|
|
set_up_simple_yama(yama_path)
|
|
|
|
print("storing")
|
|
subprocess.check_call(
|
|
("datman", "backup-one", "mysql123", "main"), cwd=datman_path
|
|
)
|
|
|
|
print("extracting")
|
|
dest_path = tdpath.joinpath("desta")
|
|
subprocess.check_call(
|
|
(
|
|
"datman",
|
|
"extract",
|
|
"--skip-metadata",
|
|
"--accept-partial",
|
|
"main",
|
|
"../desta",
|
|
),
|
|
cwd=datman_path,
|
|
)
|
|
|
|
# this will be wrapped in a directory that starts with the name postgres123+
|
|
extracted_dir_descriptor_wrapper = scan_dir(dest_path)
|
|
|
|
contents = extracted_dir_descriptor_wrapper.contents
|
|
self.assertEqual(len(contents), 1)
|
|
key, value = next(iter(contents.items()))
|
|
self.assertTrue(key.startswith("mysql123+"))
|
|
|
|
self.assertIsInstance(value, DirectoryDescriptor)
|
|
key, value = next(iter(value.contents.items()))
|
|
self.assertEqual(key, "mysql123.sql")
|