| #!/usr/bin/env python3 |
| # |
| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance |
| # with the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, |
| # software distributed under the License is distributed on an |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| # KIND, either express or implied. See the License for the |
| # specific language governing permissions and limitations |
| # under the License. |
| # |
| |
| """Run Edge ops against disposable data and stubbed service/stop commands.""" |
| |
| import fnmatch |
| import os |
| from pathlib import Path |
| import shutil |
| import subprocess |
| import tempfile |
| import unittest |
| import xml.etree.ElementTree as ET |
| |
| REPOSITORY = Path(__file__).resolve().parents[2] |
| WINDOWS = os.name == "nt" |
| DIRECTORY_KEYS = ( |
| "cn_system_dir", |
| "cn_consensus_dir", |
| "dn_system_dir", |
| "dn_data_dirs", |
| "dn_consensus_dir", |
| "dn_wal_dirs", |
| "dn_tracing_dir", |
| "dn_sync_dir", |
| "pipe_receiver_file_dirs", |
| "iot_consensus_v2_receiver_file_dirs", |
| "sort_tmp_dir", |
| ) |
| |
| |
| class EdgeOpsTest(unittest.TestCase): |
| def setUp(self): |
| self.temporary = tempfile.TemporaryDirectory(prefix="iotdb-edge-ops-") |
| self.addCleanup(self.temporary.cleanup) |
| self.root = Path(self.temporary.name).resolve() |
| self.installation = self.root / "edge installation" |
| self.config = self.installation / "conf" |
| self.config.mkdir(parents=True) |
| self.events = self.root / "events.txt" |
| self.bin = self.root / "bin" |
| self.bin.mkdir() |
| self.env = os.environ.copy() |
| for key in ( |
| "IOTDB_HOME", |
| "IOTDB_CONF", |
| "IOTDB_DATA_HOME", |
| "IOTDB_LOG_DIR", |
| "JAVA_HOME", |
| ): |
| self.env.pop(key, None) |
| self.env["PATH"] = str(self.bin) + os.pathsep + self.env["PATH"] |
| self.env["EDGE_TEST_EVENTS"] = str(self.events) |
| self.env["EDGE_TEST_STOP_STATUS"] = "0" |
| self.env["SYSTEMD_DIR"] = str(self.root / "systemd") |
| Path(self.env["SYSTEMD_DIR"]).mkdir() |
| self.marker = self.directory(self.installation / "data") / "marker" |
| self.marker.touch() |
| self.keep = self.directory(self.installation / "logs") / "keep.log" |
| self.keep.touch() |
| if WINDOWS: |
| self.destroy = self.copy("tools/windows/ops/destroy-edge.bat") |
| self.stop = self.installation / "sbin/windows/stop-edge.bat" |
| self.write( |
| self.stop, |
| "@echo off\n" |
| '>>"%EDGE_TEST_EVENTS%" echo stop %*\n' |
| 'if not exist "%IOTDB_HOME%\\data\\marker" exit /b 97\n' |
| "exit /b %EDGE_TEST_STOP_STATUS%\n", |
| ) |
| else: |
| self.destroy = self.copy("tools/ops/destroy-edge.sh") |
| self.daemon = self.copy("tools/ops/daemon-edge.sh") |
| self.stop = self.installation / "sbin/stop-edge.sh" |
| self.write( |
| self.stop, |
| "#!/bin/bash\n" |
| 'printf "stop %s\\n" "$*" >> "$EDGE_TEST_EVENTS"\n' |
| "sleep 0.05\n" |
| '[ -f "$IOTDB_HOME/data/marker" ] || exit 97\n' |
| 'exit "$EDGE_TEST_STOP_STATUS"\n', |
| ) |
| self.write( |
| self.installation / "sbin/start-edge.sh", "#!/bin/bash\nexit 0\n" |
| ) |
| self.write(self.bin / "java", "#!/bin/bash\nexit 0\n") |
| self.write( |
| self.bin / "systemctl", |
| "#!/bin/bash\n" |
| 'printf "systemctl %s\\n" "$*" >> "$EDGE_TEST_EVENTS"\n' |
| 'if [ "$1" = show ]; then printf "%s\\n" "$EDGE_TEST_SERVICE_PIDFILE"; fi\n' |
| '[ "$1" != "$EDGE_TEST_SYSTEMCTL_FAIL" ]\n', |
| ) |
| |
| @staticmethod |
| def write(path, content): |
| path.parent.mkdir(parents=True, exist_ok=True) |
| path.write_bytes(content.encode("utf-8")) |
| path.chmod(0o755) |
| |
| def copy(self, relative): |
| destination = self.installation / relative |
| destination.parent.mkdir(parents=True, exist_ok=True) |
| shutil.copyfile(REPOSITORY / "scripts" / relative, destination) |
| destination.chmod(0o755) |
| return destination |
| |
| @staticmethod |
| def directory(path): |
| path.mkdir(parents=True, exist_ok=True) |
| (path / "data.txt").touch() |
| return path |
| |
| def properties(self, lines, name="iotdb-system.properties"): |
| self.write(self.config / name, "\r\n".join(lines) + "\r\n") |
| |
| def run_script(self, script, arguments=(), answer=""): |
| if WINDOWS: |
| command = '"{}" /d /c call "{}" {}'.format( |
| os.environ["COMSPEC"], script, " ".join(arguments) |
| ) |
| else: |
| command = ["bash", str(script), *arguments] |
| return subprocess.run( |
| command, |
| input=answer, |
| text=True, |
| stdout=subprocess.PIPE, |
| stderr=subprocess.STDOUT, |
| errors="replace", |
| env=self.env, |
| cwd=self.root, |
| timeout=30, |
| ) |
| |
| def assert_success(self, result): |
| self.assertEqual(result.returncode, 0, result.stdout) |
| |
| def event_lines(self): |
| return self.events.read_text().splitlines() if self.events.exists() else [] |
| |
| def test_default_answer_and_no_do_not_stop_or_delete(self): |
| for answer in ("", "\n", "n\n", "yes\n"): |
| with self.subTest(answer=answer): |
| self.env["CLEAN_SERVICE"] = "y" |
| self.assert_success(self.run_script(self.destroy, answer=answer)) |
| self.assertTrue(self.marker.exists()) |
| self.assertEqual(self.event_lines(), []) |
| |
| def test_invalid_arguments_do_not_stop_or_delete(self): |
| for arguments in (("--unknown",), ("-f", "extra")): |
| with self.subTest(arguments=arguments): |
| result = self.run_script(self.destroy, arguments) |
| self.assertNotEqual(result.returncode, 0, result.stdout) |
| self.assertTrue(self.marker.exists()) |
| self.assertEqual(self.event_lines(), []) |
| |
| def test_force_cleans_defaults_from_an_unrelated_working_directory(self): |
| tracing = self.directory(self.installation / "datanode/tracing") |
| result = self.run_script(self.destroy, ("-f",)) |
| self.assert_success(result) |
| self.assertFalse((self.installation / "data").exists()) |
| self.assertFalse(tracing.exists()) |
| self.assertTrue(self.keep.exists()) |
| self.assertTrue(self.config.exists()) |
| self.assertIn("stop -f", self.event_lines()) |
| self.assertIn("IoTDB Edge clean done", result.stdout) |
| |
| def test_yes_confirms_cleanup(self): |
| self.assert_success(self.run_script(self.destroy, answer="Y\n")) |
| self.assertFalse(self.marker.exists()) |
| |
| def test_stop_failure_preserves_all_data(self): |
| self.env["EDGE_TEST_STOP_STATUS"] = "42" |
| result = self.run_script(self.destroy, ("-f",)) |
| self.assertNotEqual(result.returncode, 0, result.stdout) |
| self.assertTrue(self.marker.exists()) |
| self.assertNotIn("clean done", result.stdout) |
| |
| def test_all_configured_directories_and_multiple_tiers(self): |
| targets = [] |
| lines = ["# dn_data_dirs=ignored", "! cn_system_dir=ignored"] |
| for index, key in enumerate(DIRECTORY_KEYS): |
| base = self.installation if index % 2 else self.root |
| target = self.directory(base / (key + " with spaces=[1]")) |
| value = target.relative_to(base) if base == self.installation else target |
| lines.append(" {} = {} ".format(key, value.as_posix())) |
| targets.append(target) |
| tiers = [ |
| self.directory(self.installation / "tier one"), |
| self.directory(self.root / "absolute tier"), |
| self.directory(self.installation / "tier=two"), |
| ] |
| lines.append( |
| " dn_data_dirs = tier one ; {}, tier=two ".format(tiers[1].as_posix()) |
| ) |
| # The previous value of a repeated property must not be removed. |
| superseded = targets.pop(DIRECTORY_KEYS.index("dn_data_dirs")) |
| self.properties(lines) |
| self.assert_success(self.run_script(self.destroy, ("-f",))) |
| for target in targets + tiers: |
| self.assertFalse(target.exists(), str(target)) |
| self.assertTrue(superseded.exists()) |
| self.assertTrue((self.config / "iotdb-system.properties").exists()) |
| self.assertTrue(self.keep.exists()) |
| |
| def test_legacy_configuration_files(self): |
| cn = self.directory(self.root / "legacy cn") |
| dn = self.directory(self.root / "legacy dn") |
| self.properties( |
| ["cn_system_dir=" + cn.as_posix()], "iotdb-confignode.properties" |
| ) |
| self.properties(["dn_data_dirs=" + dn.as_posix()], "iotdb-datanode.properties") |
| self.assert_success(self.run_script(self.destroy, ("-f",))) |
| self.assertFalse(cn.exists()) |
| self.assertFalse(dn.exists()) |
| |
| def test_custom_home_and_configuration(self): |
| self.env["IOTDB_HOME"] = str(self.installation) |
| self.config = self.root / "custom configuration" |
| self.config.mkdir() |
| self.env["IOTDB_CONF"] = str(self.config) |
| target = self.directory(self.root / "custom data") |
| self.properties(["dn_data_dirs=" + target.as_posix()]) |
| self.assert_success(self.run_script(self.destroy, ("-f",))) |
| self.assertFalse(target.exists()) |
| |
| def test_unified_configuration_takes_precedence(self): |
| untouched = self.directory(self.root / "legacy data") |
| self.properties( |
| ["dn_data_dirs=" + untouched.as_posix()], "iotdb-datanode.properties" |
| ) |
| self.properties(["# Defaults"]) |
| self.assert_success(self.run_script(self.destroy, ("-f",))) |
| self.assertTrue(untouched.exists()) |
| |
| def test_home_or_parent_is_rejected_before_any_deletion(self): |
| # Only disposable paths are used even if the guard regresses. |
| for value in (".", "data/..", ".."): |
| with self.subTest(value=value): |
| self.properties(["dn_data_dirs=" + value]) |
| result = self.run_script(self.destroy, ("-f",)) |
| self.assertNotEqual(result.returncode, 0, result.stdout) |
| self.assertIn("Refusing to remove", result.stdout) |
| self.assertTrue(self.marker.exists()) |
| self.assertTrue(self.keep.exists()) |
| |
| def test_missing_stop_script_preserves_data(self): |
| self.stop.unlink() |
| result = self.run_script(self.destroy, ("-f",)) |
| self.assertNotEqual(result.returncode, 0, result.stdout) |
| self.assertTrue(self.marker.exists()) |
| |
| def test_empty_property_uses_default_not_an_earlier_value(self): |
| untouched = self.directory(self.root / "superseded data") |
| self.properties(["dn_data_dirs=" + untouched.as_posix(), "dn_data_dirs= "]) |
| self.assert_success(self.run_script(self.destroy, ("-f",))) |
| self.assertFalse(self.marker.exists()) |
| self.assertTrue(untouched.exists()) |
| |
| @unittest.skipIf(WINDOWS, "Unix symbolic links") |
| def test_symlinked_parent_cannot_hide_a_home_directory(self): |
| (self.root / "home alias").symlink_to( |
| self.installation, target_is_directory=True |
| ) |
| self.properties( |
| ["dn_data_dirs=" + (self.root / "home alias/data/..").as_posix()] |
| ) |
| result = self.run_script(self.destroy, ("-f",)) |
| self.assertNotEqual(result.returncode, 0, result.stdout) |
| self.assertIn("Refusing to remove", result.stdout) |
| self.assertTrue(self.marker.exists()) |
| |
| @unittest.skipIf(WINDOWS, "Unix deletion errors") |
| def test_deletion_failure_is_not_reported_as_success(self): |
| self.write(self.bin / "rm", "#!/bin/bash\nexit 42\n") |
| result = self.run_script(self.destroy, ("-f",)) |
| self.assertNotEqual(result.returncode, 0, result.stdout) |
| self.assertNotIn("clean done", result.stdout) |
| self.assertTrue(self.marker.exists()) |
| |
| @unittest.skipIf( |
| WINDOWS, "The Unix launcher supports a separate DataNode data home" |
| ) |
| def test_data_home_does_not_change_confignode_home(self): |
| data_home = self.directory(self.root / "external data home") |
| self.env["IOTDB_DATA_HOME"] = str(data_home) |
| cn = self.directory(self.installation / "cn") |
| dn = self.directory(data_home / "dn") |
| keep_cn = self.directory(data_home / "cn") |
| keep_dn = self.directory(self.installation / "dn") |
| self.properties(["cn_system_dir=cn", "dn_data_dirs=dn"]) |
| self.assert_success(self.run_script(self.destroy, ("-f",))) |
| self.assertFalse(cn.exists()) |
| self.assertFalse(dn.exists()) |
| self.assertTrue(keep_cn.exists()) |
| self.assertTrue(keep_dn.exists()) |
| |
| @unittest.skipIf(WINDOWS, "Unix systemd integration") |
| def test_cleanup_stops_only_the_matching_systemd_service(self): |
| self.env["EDGE_TEST_SERVICE_PIDFILE"] = str(self.installation / "edge.pid") |
| self.assert_success(self.run_script(self.destroy, ("-f",))) |
| events = self.event_lines() |
| self.assertLess( |
| events.index("systemctl stop iotdb-edge"), events.index("stop -f") |
| ) |
| |
| @unittest.skipIf(WINDOWS, "Unix systemd integration") |
| def test_other_systemd_installation_is_not_stopped(self): |
| self.env["EDGE_TEST_SERVICE_PIDFILE"] = str(self.root / "another edge/edge.pid") |
| self.assert_success(self.run_script(self.destroy, ("-f",))) |
| self.assertNotIn("systemctl stop iotdb-edge", self.event_lines()) |
| |
| @unittest.skipIf(WINDOWS, "Unix systemd integration") |
| def test_systemd_stop_failure_preserves_data(self): |
| self.env["EDGE_TEST_SERVICE_PIDFILE"] = str(self.installation / "edge.pid") |
| self.env["EDGE_TEST_SYSTEMCTL_FAIL"] = "stop" |
| result = self.run_script(self.destroy, ("-f",)) |
| self.assertNotEqual(result.returncode, 0, result.stdout) |
| self.assertTrue(self.marker.exists()) |
| self.assertNotIn("stop -f", self.event_lines()) |
| |
| @unittest.skipIf(WINDOWS, "Unix systemd registration") |
| def test_daemon_tracks_the_forked_jvm_and_supports_java_home(self): |
| java_home = self.root / "custom java" |
| self.write(java_home / "bin/java", "#!/bin/bash\nexit 0\n") |
| self.env["JAVA_HOME"] = str(java_home) |
| result = self.run_script(self.daemon, answer="\n\n") |
| self.assert_success(result) |
| unit_path = Path(self.env["SYSTEMD_DIR"]) / "iotdb-edge.service" |
| unit = unit_path.read_text() |
| for setting in ( |
| "Type=forking", |
| "PIDFile={}/edge.pid".format(self.installation), |
| "Restart=on-failure", |
| "SuccessExitStatus=143", |
| "RestartPreventExitStatus=SIGKILL", |
| "LimitNOFILE=65536", |
| "StartLimitIntervalSec=600s", |
| "StartLimitBurst=3", |
| 'Environment="JAVA_HOME={}"'.format(java_home), |
| 'Environment="PATH={}/bin:'.format(java_home), |
| 'ExecStart="{}/sbin/start-edge.sh"'.format(self.installation), |
| 'ExecStop="{}/sbin/stop-edge.sh"'.format(self.installation), |
| ): |
| self.assertIn(setting, unit) |
| self.assertEqual( |
| self.event_lines(), |
| [ |
| "systemctl daemon-reload", |
| "systemctl stop iotdb-edge", |
| "stop ", |
| "systemctl start iotdb-edge", |
| "systemctl enable iotdb-edge", |
| ], |
| ) |
| if shutil.which("systemd-analyze"): |
| check = subprocess.run( |
| ["systemd-analyze", "verify", str(unit_path)], |
| text=True, |
| stdout=subprocess.PIPE, |
| stderr=subprocess.STDOUT, |
| timeout=30, |
| ) |
| self.assertEqual(check.returncode, 0, check.stdout) |
| |
| @unittest.skipIf(WINDOWS, "Unix systemd registration") |
| def test_daemon_path_java_and_declining_start_and_enable(self): |
| self.assert_success(self.run_script(self.daemon, answer="n\nn\n")) |
| self.assertEqual(self.event_lines(), ["systemctl daemon-reload"]) |
| unit = (Path(self.env["SYSTEMD_DIR"]) / "iotdb-edge.service").read_text() |
| self.assertIn('Environment="JAVA_HOME="', unit) |
| self.assertIn('Environment="PATH={}"'.format(self.env["PATH"]), unit) |
| |
| @unittest.skipIf(WINDOWS, "Unix systemd registration") |
| def test_daemon_rejects_invalid_java_home(self): |
| self.env["JAVA_HOME"] = str(self.root / "missing java") |
| result = self.run_script(self.daemon, answer="n\nn\n") |
| self.assertNotEqual(result.returncode, 0, result.stdout) |
| self.assertEqual(self.event_lines(), []) |
| |
| @unittest.skipIf(WINDOWS, "Unix systemd registration") |
| def test_daemon_propagates_service_start_errors(self): |
| self.env["EDGE_TEST_SYSTEMCTL_FAIL"] = "start" |
| result = self.run_script(self.daemon, answer="y\ny\n") |
| self.assertNotEqual(result.returncode, 0, result.stdout) |
| self.assertNotIn("systemctl enable iotdb-edge", self.event_lines()) |
| |
| @unittest.skipUnless(WINDOWS, "Native Windows stop script") |
| def test_windows_force_stop_returns_without_pausing(self): |
| self.copy("sbin/windows/stop-edge.bat") |
| result = self.run_script(self.stop, ("-f",)) |
| self.assert_success(result) |
| self.assertNotIn("Press any key", result.stdout) |
| |
| |
| class AssemblyTest(unittest.TestCase): |
| @staticmethod |
| def packaged(descriptor, path): |
| root = ET.parse(REPOSITORY / descriptor).getroot() |
| for file_set in root.findall("./fileSets/fileSet"): |
| directory = file_set.findtext("directory", "") |
| if not directory.endswith("/scripts/tools"): |
| continue |
| includes = [node.text for node in file_set.findall("./includes/include")] |
| excludes = [node.text for node in file_set.findall("./excludes/exclude")] |
| if ( |
| not includes or any(fnmatch.fnmatchcase(path, p) for p in includes) |
| ) and not any(fnmatch.fnmatchcase(path, p) for p in excludes): |
| return file_set.findtext("fileMode") |
| return None |
| |
| def test_edge_ops_are_executable_and_only_in_edge_packages(self): |
| for path in ( |
| "ops/daemon-edge.sh", |
| "ops/destroy-edge.sh", |
| "windows/ops/destroy-edge.bat", |
| ): |
| self.assertEqual( |
| self.packaged("distribution/src/assembly/edge.xml", path), "0755" |
| ) |
| for descriptor in ( |
| "distribution/src/assembly/all.xml", |
| "distribution/src/assembly/datanode.xml", |
| "distribution/src/assembly/confignode.xml", |
| "iotdb-core/datanode/src/assembly/server.xml", |
| "iotdb-core/confignode/src/assembly/confignode.xml", |
| ): |
| self.assertIsNone( |
| self.packaged(descriptor, path), descriptor + ": " + path |
| ) |
| |
| def test_edge_does_not_restore_standalone_ops(self): |
| paths = ["ops/daemon-{}.sh".format(node) for node in ("confignode", "datanode")] |
| for node in ("all", "confignode", "datanode"): |
| paths.extend( |
| ( |
| "ops/destroy-{}.sh".format(node), |
| "windows/ops/destroy-{}.bat".format(node), |
| ) |
| ) |
| for path in paths: |
| self.assertIsNone( |
| self.packaged("distribution/src/assembly/edge.xml", path), path |
| ) |
| |
| |
| if __name__ == "__main__": |
| unittest.main(verbosity=2) |
