"""Real MySQL integration in a fresh disposable Docker container, no production access.""" import hashlib import io from pathlib import Path import subprocess import tarfile import time import unittest import uuid ROOT = Path(__file__).resolve().parents[1] TARGET = '/workspace/wwwroot/zhct_zhangjiagang' STATE = '/workspace/zhct-upgrade-backups/zhangjiagang/220' RUN = '/root/.zhct-upgrade-220/run-' + 'a' * 32 INCOMING = TARGET + '/.zhct_220-' + 'a' * 32 + '.tgz' CNF = RUN + '/mysql.cnf' DEPLOY = '#!/bin/bash\nset -eu\n[[ $1 == 220 ]]\necho deployed >> /workspace/deploy-count\n' class RemoteTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.name = 'zhct220-test-' + uuid.uuid4().hex[:10] subprocess.run(['docker', 'run', '-d', '--rm', '--name', cls.name, '-e', 'MYSQL_ALLOW_EMPTY_PASSWORD=yes', '-v', f'{ROOT}:/src:ro', 'mysql:8.0'], check=True, capture_output=True) cls.addClassCleanup(lambda: subprocess.run(['docker', 'stop', cls.name], check=False, capture_output=True)) for _ in range(90): # The image briefly starts a socket-only initialization server. # Wait for TCP queries, so tests do not race its shutdown/restart. ready = cls.command(['mysql', '-h127.0.0.1', '-uroot', '-NBe', 'SELECT 1'], check=False) if ready.returncode == 0: break time.sleep(1) else: raise RuntimeError('MySQL container did not become ready') # Oracle's minimal MySQL image omits flock. Use Alpine's genuine kernel # flock via BusyBox + its matching loader, not a lock mock. if cls.command(['bash', '-c', 'command -v flock'], check=False).returncode: fixture = ROOT / 'runtime' fixture.mkdir(exist_ok=True) subprocess.run(['docker', 'run', '--rm', '-v', f'{fixture}:/out', 'alpine:latest', 'sh', '-ec', 'cp /bin/busybox /out/busybox; cp /lib/ld-musl-*.so.1 /out/test-musl'], check=True, capture_output=True, timeout=60) cls.command(['bash', '-c', 'cat > /usr/bin/flock; chmod 755 /usr/bin/flock'], input='#!/bin/sh\nexec /src/runtime/test-musl /src/runtime/busybox flock "$@"\n') cls.command(['bash', '-ec', 'for tool in curl flock timeout realpath gzip sha256sum tar; do command -v "$tool"; done']) @classmethod def command(cls, args, input=None, check=True): return subprocess.run(['docker', 'exec', '-i', cls.name, *args], input=input, text=True, capture_output=True, check=check, timeout=60) def put(self, path, content, mode='600'): self.command(['bash', '-c', 'cat > "$1"; chmod "$2" "$1"', '_', path, mode], input=content) def sql(self, query): return self.command(['mysql', '-uroot', '--default-character-set=utf8mb4', '-NBe', query]).stdout.strip() def setUp(self): self.command(['bash', '-c', 'rm -rf /workspace /root/.zhct-upgrade-220 /tmp/test-bin; ' f'mkdir -p {TARGET}/project {RUN} /tmp/test-bin; ' f'echo old-code > {TARGET}/project/index.php']) self.sql("DROP DATABASE IF EXISTS zhct_zhangjiagang; CREATE DATABASE zhct_zhangjiagang CHARACTER SET utf8mb4; " "CREATE TABLE zhct_zhangjiagang.ydy_config (config_key varchar(255), config_value varchar(255)) ENGINE=InnoDB; " "INSERT INTO zhct_zhangjiagang.ydy_config VALUES ('ossHost','http://old'); " "CREATE TABLE zhct_zhangjiagang.ydy_meal_order (id int, pay_method varchar(255), tail int) ENGINE=InnoDB;") self.put(TARGET + '/deploy_zhct.sh', DEPLOY, '700') self.put('/tmp/test-bin/curl', '#!/bin/bash\ncat /tmp/http-code\n', '700') self.put('/tmp/http-code', '200') self.make_credentials() package = io.BytesIO() with tarfile.open(fileobj=package, mode='w:gz') as archive: item = tarfile.TarInfo('index.php') content = b'new-code' item.size = len(content) archive.addfile(item, io.BytesIO(content)) self.package_bytes = package.getvalue() self.package_hash = hashlib.sha256(self.package_bytes).hexdigest() self.script_hash = hashlib.sha256(DEPLOY.encode()).hexdigest() self.upload() def make_credentials(self): # Test-only root credentials; localhost is inside this disposable container. self.put(CNF, '[client]\nuser=root\npassword=""\ndefault-character-set=utf8mb4\n') def upload(self): subprocess.run(['docker', 'exec', '-i', self.name, 'bash', '-c', 'cat > "$1"', '_', INCOMING], input=self.package_bytes, check=True, capture_output=True) def run_upgrade(self, package_hash=None, script_hash=None): return self.command(['env', 'PATH=/tmp/test-bin:/usr/bin:/bin', 'bash', '/src/remote-upgrade.sh', CNF, package_hash or self.package_hash, script_hash or self.script_hash, INCOMING], check=False) def assert_stopped(self, result, marker): self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr) self.assertIn(marker, result.stdout + result.stderr) self.assertFalse(self.command(['test', '-e', CNF], check=False).returncode == 0) def test_fresh_upgrade_and_repeat(self): result = self.run_upgrade() self.assertEqual(result.returncode, 0, result.stdout + result.stderr) self.assertIn('SQL_VERIFIED', result.stdout) self.assertIn('BUSINESS_SMOKE_PENDING', result.stdout) self.assertEqual(self.sql("SELECT config_value FROM zhct_zhangjiagang.ydy_config;"), 'http://10.28.45.75') self.assertEqual(self.sql("SELECT column_type FROM information_schema.columns WHERE table_schema='zhct_zhangjiagang' AND column_name='bank_user_name';"), 'varchar(255)') self.assertEqual(self.command(['stat', '-c', '%a', STATE]).stdout.strip(), '700') self.make_credentials() self.upload() repeat = self.run_upgrade() self.assertEqual(repeat.returncode, 0, repeat.stdout + repeat.stderr) self.assertIn('DEPLOY_ALREADY_EXECUTED', repeat.stdout) self.assertEqual(self.command(['wc', '-l', '/workspace/deploy-count']).stdout.split()[0], '1') def test_column_conflict_before_write(self): self.sql('ALTER TABLE zhct_zhangjiagang.ydy_meal_order ADD bank_user_name int;') self.assert_stopped(self.run_upgrade(), 'BANK_COLUMN_CONFLICT') self.assertEqual(self.sql('SELECT config_value FROM zhct_zhangjiagang.ydy_config;'), 'http://old') def test_missing_config(self): self.sql('DELETE FROM zhct_zhangjiagang.ydy_config;') self.assert_stopped(self.run_upgrade(), 'OSS_CONFIG_NOT_UNIQUE') def test_duplicate_config(self): self.sql("INSERT INTO zhct_zhangjiagang.ydy_config VALUES ('ossHost','duplicate');") self.assert_stopped(self.run_upgrade(), 'OSS_CONFIG_NOT_UNIQUE') def test_missing_table(self): self.sql('DROP TABLE zhct_zhangjiagang.ydy_meal_order;') self.assert_stopped(self.run_upgrade(), 'REQUIRED_TABLE_MISSING') def test_corrupt_package(self): self.assert_stopped(self.run_upgrade(package_hash='0' * 64), 'PACKAGE_HASH_MISMATCH') def test_changed_deploy_script(self): self.assert_stopped(self.run_upgrade(script_hash='0' * 64), 'DEPLOY_SCRIPT_HASH_MISMATCH') def test_failed_backup_prevents_sql(self): self.put('/tmp/test-bin/mysqldump', '#!/bin/bash\nexit 2\n', '700') self.assert_stopped(self.run_upgrade(), 'COMMAND_FAILED') self.assertEqual(self.sql('SELECT config_value FROM zhct_zhangjiagang.ydy_config;'), 'http://old') def test_non_innodb_requires_engineer(self): self.sql('ALTER TABLE zhct_zhangjiagang.ydy_config ENGINE=MyISAM;') self.assert_stopped(self.run_upgrade(), 'NON_INNODB_REQUIRES_MAINTENANCE_BACKUP') def test_existing_correct_column(self): self.sql("ALTER TABLE zhct_zhangjiagang.ydy_meal_order ADD bank_user_name varchar(255) NOT NULL DEFAULT '' COMMENT '张家港银行核销返回的消费人姓名' AFTER pay_method;") result = self.run_upgrade() self.assertEqual(result.returncode, 0, result.stdout + result.stderr) def test_partial_sql_can_resume_without_repeating_deploy(self): self.put('/tmp/test-bin/mysql', '#!/bin/bash\nif [[ "$*" == *"ALTER TABLE"* ]]; then exit 1; fi\nexec /usr/bin/mysql "$@"\n', '700') self.assert_stopped(self.run_upgrade(), 'COMMAND_FAILED') self.assertEqual(self.sql('SELECT config_value FROM zhct_zhangjiagang.ydy_config;'), 'http://10.28.45.75') self.assertNotEqual(self.command(['test', '-e', STATE + '/deploy.started'], check=False).returncode, 0) self.command(['rm', '/tmp/test-bin/mysql']) self.make_credentials() self.upload() resumed = self.run_upgrade() self.assertEqual(resumed.returncode, 0, resumed.stdout + resumed.stderr) def test_concurrent_run_does_not_publish_package(self): self.command(['mkdir', '-p', STATE]) subprocess.run(['docker', 'exec', '-d', self.name, 'bash', '-c', f'exec 8>{STATE}/upgrade.lock; flock -n 8; touch /tmp/lock-held; sleep 30'], check=True) for _ in range(50): if self.command(['test', '-e', '/tmp/lock-held'], check=False).returncode == 0: break time.sleep(.1) else: self.fail('Lock holder did not become ready') self.assert_stopped(self.run_upgrade(), 'UPGRADE_ALREADY_RUNNING') self.assertNotEqual(self.command(['test', '-e', TARGET + '/zhct_220.tgz'], check=False).returncode, 0) def test_application_account_and_password_option_escaping(self): password = 'p"a\\ss#; $()' self.sql("SET SESSION sql_mode='NO_BACKSLASH_ESCAPES'; DROP USER IF EXISTS 'wwwweb'@'%'; " f"CREATE USER 'wwwweb'@'%' IDENTIFIED BY '{password}'; " "GRANT ALL ON zhct_zhangjiagang.* TO 'wwwweb'@'%';") self.put(CNF, '[client]\nhost=127.0.0.1\nprotocol=tcp\nuser=wwwweb\ndefault-character-set=utf8mb4\npassword="p\\"a\\\\ss#; $()"\n') result = self.run_upgrade() self.assertEqual(result.returncode, 0, result.stdout + result.stderr) def test_failed_deploy_blocks_repeat(self): failed = DEPLOY + 'exit 9\n' self.put(TARGET + '/deploy_zhct.sh', failed, '700') self.script_hash = hashlib.sha256(failed.encode()).hexdigest() self.assert_stopped(self.run_upgrade(), 'DEPLOY_FAILED_ENGINEER_REQUIRED') self.make_credentials() self.upload() self.assert_stopped(self.run_upgrade(), 'DEPLOY_INTERRUPTED_ENGINEER_REQUIRED') def test_bad_http_after_deploy(self): self.put('/tmp/http-code', '503') self.assert_stopped(self.run_upgrade(), 'HTTP_STATUS_503') self.assertEqual(self.command(['test', '-e', STATE + '/deploy.ok'], check=False).returncode, 0) def test_backup_is_database_from_before_migration(self): result = self.run_upgrade() self.assertEqual(result.returncode, 0, result.stderr) self.sql('DROP DATABASE zhct_zhangjiagang; CREATE DATABASE zhct_zhangjiagang;') self.command(['bash', '-c', f'gzip -cd {STATE}/database.sql.gz | mysql -uroot zhct_zhangjiagang']) self.assertEqual(self.sql('SELECT config_value FROM zhct_zhangjiagang.ydy_config;'), 'http://old') self.assertEqual(self.sql("SELECT COUNT(*) FROM information_schema.columns WHERE table_schema='zhct_zhangjiagang' AND column_name='bank_user_name';"), '0') if __name__ == '__main__': unittest.main(verbosity=2)