#!/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",

"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)