"""Exercise real Bash release/rollback file operations with all infrastructure faked.""" import json import os from pathlib import Path import shutil import subprocess import sys import tempfile import unittest from deploy.gray_config import GrayConfig, write_release FAKE_COMMAND = '''import json, os, pathlib, sys name = pathlib.Path(sys.argv[0]).name args = sys.argv[1:] log = pathlib.Path(os.environ['FAKE_LOG']) with log.open('a') as output: output.write(json.dumps([name, args]) + '\\n') scenario = os.environ['FAKE_SCENARIO'] if name == 'docker': if args[:2] == ['image', 'inspect']: print('sha256:' + 'b'*64 if args[3] == '{{.Id}}' else 'a'*40) elif args[:2] == ['container', 'inspect']: sys.exit(0) elif args[0] == 'inspect': print('baiyue-logistics-chat-gray') elif args[0] == 'port': print('127.0.0.1:29013') elif args[0] == 'compose' and 'up' in args and scenario == 'app-fail' and os.environ['FAKE_NEW'] + '/release.env' in args: sys.exit(17) elif name == 'nginx' and '-t' in args and scenario == 'proxy-fail': calls = [json.loads(line) for line in log.read_text().splitlines()] if sum(item[0] == 'nginx' and '-t' in item[1] for item in calls) == 1: sys.exit(19) elif name == 'curl' and '-w' in args: calls = [json.loads(line) for line in log.read_text().splitlines()] if scenario == 'reload-pending' and sum(item[0] == 'curl' and '-w' in item[1] for item in calls) == 1: sys.exit(60) if scenario == 'proxy-unsafe': print('200', end='') sys.exit(0) path = '/' + args[-1].split('://', 1)[-1].partition('/')[2] if '-c' in args: pathlib.Path(args[args.index('-c') + 1]).write_text('fixture-visitor-cookie') if '-b' in args: assert pathlib.Path(args[args.index('-b') + 1]).read_text() == 'fixture-visitor-cookie' status = '200' if path == '/' or path == '/api/visitor/session' or (path == '/api/get_history' and '-b' in args) else '401' if scenario == 'visitor-bootstrap-fail' and path == '/api/visitor/session': status = '503' if scenario == 'visitor-history-fail' and path == '/api/get_history' and '-b' in args: status = '503' print(status, end='') elif name == 'sleep': pass ''' @unittest.skipUnless(os.name == 'posix' and shutil.which('bash'), 'Gray activation requires Linux Bash') class GrayActivationTests(unittest.TestCase): def run_activation(self, scenario): temporary = tempfile.TemporaryDirectory() self.addCleanup(temporary.cleanup) directory = Path(temporary.name) root = directory / 'gray-logistics-chat' proxy = directory / 'proxy' auth = directory / 'baiyue-logistics-chat-gray-proxy' for path in (root / 'data', root / 'logs', root / 'releases', proxy, auth): path.mkdir(parents=True, exist_ok=True) (root / '.project').write_text('baiyue-logistics-chat-gray\n') (root / 'data/history.db').write_bytes(b'preserve-existing-data') certificate, private_key = directory / 'cert.pem', directory / 'key.pem' certificate.touch() private_key.touch() previous = root / 'releases/previous' previous.mkdir() (previous / 'release.env').write_text('CHAT_IMAGE=previous\n') (previous / 'compose.yml').write_text('services: {}\n') (root / 'current-release').write_text(str(previous) + '\n') old_site = '# Managed for baiyue-logistics-chat-gray\n# previous site\n' (proxy / 'baiyue-logistics-chat-gray.conf').write_text(old_site) (auth / '.project').write_text('another-service\n' if scenario == 'unowned-auth' else 'baiyue-logistics-chat-gray\n') (auth / 'access.htpasswd').write_text('previous-auth-hash\n') unrelated = proxy / 'unrelated.conf' unrelated.write_text('# unrelated site\n') config = GrayConfig('chat.example.com', 29013, str(root), str(root / 'releases/new'), str(proxy), str(certificate), str(private_key), 'a'*40, 'baiyue-logistics-chat-gray:' + 'a'*12 + '-b7', 'sha256:' + 'd'*64, 'sha256:' + 'c'*64, str(auth)) release = Path(config.release) write_release(release, config, 'fixture-model-key') # Docker 26 and 29 may assign different IDs after an archive is loaded. loaded_id = ('e' if scenario == 'image-mismatch' else 'b') * 64 (release / 'loaded-image-ids.env').write_text('CHAT_IMAGE_ID=sha256:' + loaded_id + '\n') (release / 'site.htpasswd').write_text('new-auth-hash\n') commands = directory / 'commands' commands.mkdir() for name in ('docker', 'nginx', 'openssl', 'ss', 'curl', 'sleep'): command = commands / name command.write_text('#!' + sys.executable + '\n' + FAKE_COMMAND) command.chmod(0o755) environment = os.environ.copy() log = directory / 'commands.jsonl' environment.update(PATH=str(commands) + os.pathsep + environment['PATH'], FAKE_LOG=str(log), FAKE_SCENARIO=scenario, FAKE_NEW=str(release)) result = subprocess.run(['bash', str(release / 'activate.sh'), str(root), str(release)], env=environment, capture_output=True, text=True, timeout=30) self.assertEqual((root / 'data/history.db').read_bytes(), b'preserve-existing-data') self.assertEqual(unrelated.read_text(), '# unrelated site\n') self.assertFalse(list(release.glob('.visitor-smoke.*')), 'Temporary visitor cookies must be removed on success and rollback') calls = [json.loads(line) for line in log.read_text().splitlines()] return result, root, proxy, previous, release, old_site, calls def test_success_publishes_only_the_owned_site_and_release_pointer(self): result, root, proxy, previous, release, old_site, calls = self.run_activation('success') self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual((root / 'current-release').read_text().strip(), str(release)) self.assertIn('server_name chat.example.com', (proxy / 'baiyue-logistics-chat-gray.conf').read_text()) auth = proxy.parent / 'baiyue-logistics-chat-gray-proxy' self.assertEqual((auth / 'access.htpasswd').read_text(), 'new-auth-hash\n') self.assertIn(str(auth / 'access.htpasswd'), (proxy / 'baiyue-logistics-chat-gray.conf').read_text()) self.assertEqual(auth.stat().st_mode & 0o777, 0o755) self.assertEqual((auth / 'access.htpasswd').stat().st_mode & 0o777, 0o644) def assert_rolled_back(self, scenario): result, root, proxy, previous, release, old_site, calls = self.run_activation(scenario) self.assertNotEqual(result.returncode, 0) self.assertEqual((root / 'current-release').read_text().strip(), str(previous)) self.assertEqual((proxy / 'baiyue-logistics-chat-gray.conf').read_text(), old_site) self.assertEqual((proxy.parent / 'baiyue-logistics-chat-gray-proxy/access.htpasswd').read_text(), 'previous-auth-hash\n') self.assertTrue(any(name == 'docker' and 'up' in args and str(previous / 'release.env') in args for name, args in calls), result.stderr) return result def test_failed_application_restores_previous_release(self): self.assert_rolled_back('app-fail') def test_async_nginx_reload_waits_for_the_new_tls_workers(self): result, root, proxy, previous, release, old_site, calls = self.run_activation('reload-pending') self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual((root / 'current-release').read_text().strip(), str(release)) self.assertTrue(any(name == 'sleep' for name, args in calls)) def test_unprotected_proxy_never_publishes_the_release_pointer(self): self.assert_rolled_back('proxy-unsafe') def test_failed_visitor_bootstrap_rolls_back_and_removes_cookie_file(self): result = self.assert_rolled_back('visitor-bootstrap-fail') self.assertIn('Guest session bootstrap verification failed.', result.stderr) def test_failed_cookie_history_rolls_back_and_removes_cookie_file(self): result = self.assert_rolled_back('visitor-history-fail') self.assertIn('Guest history verification failed.', result.stderr) def test_image_retagged_after_verified_load_is_rejected_before_activation(self): result, root, proxy, previous, release, old_site, calls = self.run_activation('image-mismatch') self.assertNotEqual(result.returncode, 0) self.assertEqual((root / 'current-release').read_text().strip(), str(previous)) self.assertEqual((proxy / 'baiyue-logistics-chat-gray.conf').read_text(), old_site) self.assertFalse(any(name == 'docker' and 'compose' in args for name, args in calls)) def test_password_directory_owned_by_another_service_is_not_modified(self): result, root, proxy, previous, release, old_site, calls = self.run_activation('unowned-auth') self.assertNotEqual(result.returncode, 0) auth = proxy.parent / 'baiyue-logistics-chat-gray-proxy' self.assertEqual((auth / '.project').read_text(), 'another-service\n') self.assertEqual((auth / 'access.htpasswd').read_text(), 'previous-auth-hash\n') self.assertEqual((root / 'current-release').read_text().strip(), str(previous)) self.assertFalse(any(name == 'docker' and 'compose' in args for name, args in calls)) def test_invalid_nginx_configuration_restores_previous_site_and_application(self): self.assert_rolled_back('proxy-fail') if __name__ == '__main__': unittest.main()