#!/usr/bin/env python3
"""Offline send-queue cleanup tests; all paths and database responses are fixtures."""
import json
import os
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile
import unittest


BIN = Path(__file__).resolve().parents[1] / "Server/Unix/p4/common/bin"
BASH = os.environ.get("SDP_TEST_BASH", shutil.which("bash"))


class SendqFixture(unittest.TestCase):
    def setUp(self):
        temporary = tempfile.TemporaryDirectory(prefix="sdp-sendq-")
        self.addCleanup(temporary.cleanup)
        self.folder = Path(temporary.name).resolve()
        self.database = self.folder / "actual database"
        self.database.mkdir()
        (self.database / "db.config").touch()
        (self.database / "server.id").write_text("local-id\n")
        self.root = self.folder / "root"
        self.root.symlink_to(self.database)
        self.config = self.folder / "cshow.txt"
        self.config.write_text("")
        self.logfile = self.folder / "cleanup.log"
        self.calls = self.folder / "p4d-calls.jsonl"
        self.mockbin = self.folder / "bin"
        self.mockbin.mkdir()
        self.p4d = self.mockbin / "p4d"
        self.p4d.write_text("#!" + sys.executable + '''
import json, os, pathlib, sys
with open(os.environ["TEST_CALLS"], "a") as out:
    out.write(json.dumps(sys.argv[1:]) + "\\n")
if os.environ.get("TEST_CSHOW_FAILURE"):
    print("injected database read failure", file=sys.stderr)
    sys.exit(7)
sys.stdout.write(pathlib.Path(os.environ["TEST_CSHOW"]).read_text())
''')
        self.p4d.chmod(0o755)
        self.env = dict(os.environ, P4ROOT=str(self.root), P4DBIN=str(self.p4d),
                        P4CBIN=str(self.folder), LOGFILE=str(self.logfile),
                        TEST_CALLS=str(self.calls), TEST_CSHOW=str(self.config),
                        BACKUP_FUNCTIONS=str(BIN / "backup_functions.sh"),
                        P4NAME="", P4DEBUG="", P4D_FLAGS="", SERVERID="local-id",
                        PATH=str(self.mockbin) + ":" + os.environ["PATH"])
        self.prefix = '''source "$BACKUP_FUNCTIONS"
log () { printf '%s\\n' "$*" >> "$LOGFILE"; }
die () { log "$*"; exit 97; }
'''

    def run_bash(self, source, **environment):
        return subprocess.run([BASH, "-c", self.prefix + source], cwd=self.folder,
                              env=dict(self.env, **environment), capture_output=True,
                              timeout=15)

    def resolve(self, text, **environment):
        self.config.write_text(text)
        return self.run_bash('''sdp_read_local_client_dirs || exit "$?"
printf '%s\\0' "$ClientConfigServer" "$ClientSendqDir" "$ClientReadonlyDir"
''', **environment)

    def assert_resolved(self, result, server, sendq="", readonly=""):
        self.assertEqual(result.returncode, 0, result.stderr.decode() + self.log())
        self.assertEqual(result.stdout.decode().split("\0"), [server, sendq, readonly, ""])

    def log(self):
        return self.logfile.read_text() if self.logfile.exists() else ""


class LocalClientDirectoryResolution(SendqFixture):
    def test_database_read_uses_current_root_and_local_p4d_only(self):
        self.assert_resolved(self.resolve(""), "local-id")
        self.assertEqual([json.loads(line) for line in self.calls.read_text().splitlines()],
                         [["-r", str(self.root), "-cshow"]])

    def test_named_settings_override_global_independent_of_record_order(self):
        rows = ["any: client.sendq.dir = global", "local-id: client.sendq.dir = local",
                "any: client.readonly.dir = shared", "local-id: client.readonly.dir = persistent",
                "remote-id: client.sendq.dir = wrong"]
        for values in (rows, list(reversed(rows))):
            with self.subTest(values=values):
                self.assert_resolved(self.resolve("\n".join(values)), "local-id", "local", "persistent")

    def test_global_setting_is_used_when_local_override_is_absent(self):
        self.assert_resolved(self.resolve("any: client.sendq.dir = global\nremote: client.sendq.dir = wrong\n"),
                             "local-id", "global")

    def test_unset_has_no_invented_directory_default(self):
        self.assert_resolved(self.resolve("any: security = 4\n"), "local-id")

    def test_explicit_empty_named_values_disable_global_settings(self):
        self.assert_resolved(self.resolve("any: client.sendq.dir = global\nlocal-id: client.sendq.dir = \n"
                                          "any: client.readonly.dir = shared\nlocal-id: client.readonly.dir = \n"),
                             "local-id")

    def test_paths_preserve_spaces_equals_and_shell_characters(self):
        value = " path = 'quoted' $(touch marker) `touch second-marker` * \\\tend "
        self.assert_resolved(self.resolve("local-id: client.sendq.dir = " + value + "\n"), "local-id", value)
        self.assertFalse((self.folder / "marker").exists())
        self.assertFalse((self.folder / "second-marker").exists())

    def test_server_id_matching_is_literal_and_uses_disk_identity(self):
        (self.database / "server.id").write_text("edge.1\n")
        self.assert_resolved(self.resolve("edgeX1: client.sendq.dir = wrong\n"
                                          "other-edge.1: client.sendq.dir = also-wrong\n"
                                          "edge.1: client.sendq.dir = right\n"),
                             "edge.1", "right")

    def test_p4name_environment_selects_configuration_scope(self):
        self.assert_resolved(self.resolve("local-id: client.sendq.dir = wrong\n"
                                          "legacy: client.sendq.dir = right\n", P4NAME="legacy"),
                             "legacy", "right")

    def test_startup_In_takes_precedence_and_accepts_both_supported_forms(self):
        rows = "local-id: client.sendq.dir = wrong\nlegacy: client.sendq.dir = right\n"
        for flags in ("-p 1666 -In legacy -d", "-p 1666 -Inlegacy -d", "-p 1666\n-In\tlegacy -d"):
            with self.subTest(flags=flags):
                self.assert_resolved(self.resolve(rows, P4NAME="local-id", P4D_FLAGS=flags), "legacy", "right")

    def test_database_p4name_redirects_once_without_recursing(self):
        rows = ("local-id: P4NAME = legacy\nlegacy: P4NAME = third\n"
                "local-id: client.sendq.dir = wrong\nlegacy: client.sendq.dir = right\n"
                "third: client.sendq.dir = third-value\n")
        self.assert_resolved(self.resolve(rows), "legacy", "right")
        self.assert_resolved(self.resolve(rows, P4NAME="legacy"), "third", "third-value")
        self.assert_resolved(self.resolve(rows, P4D_FLAGS="-In local-id"), "local-id", "wrong")

    def test_missing_server_identity_uses_any_scope(self):
        (self.database / "server.id").unlink()
        self.assert_resolved(self.resolve("any: client.sendq.dir = global\n"), "any", "global")

    def test_legacy_global_p4name_redirect_is_overridden_by_named_or_In(self):
        rows = ("any: P4NAME = legacy\nany: client.sendq.dir = global\n"
                "legacy: client.sendq.dir = legacy-path\n"
                "local-id: client.sendq.dir = local-path\nthird: client.sendq.dir = third-path\n")
        self.assert_resolved(self.resolve(rows), "legacy", "legacy-path")
        self.assert_resolved(self.resolve(rows + "local-id: P4NAME = third\n"), "third", "third-path")
        self.assert_resolved(self.resolve(rows, P4D_FLAGS="-In local-id"), "local-id", "local-path")
        (self.database / "server.id").unlink()
        self.assert_resolved(self.resolve(rows), "legacy", "legacy-path")

    def test_read_failure_returns_nonzero_and_preserves_diagnostic(self):
        result = self.resolve("any: client.sendq.dir = wrong\n", TEST_CSHOW_FAILURE="1")
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("injected database read failure", self.log())

    def test_missing_database_is_not_treated_as_an_unset_configurable(self):
        (self.database / "db.config").unlink()
        result = self.resolve("")
        self.assertNotEqual(result.returncode, 0)
        self.assertFalse(self.calls.exists())

    def test_malformed_and_duplicate_records_fail_closed(self):
        for text in ("local-id: client.sendq.dir wrong\n", "not a config record\n",
                     "local-id: client.sendq.dir = one\nlocal-id: client.sendq.dir = two\n",
                     "local-id: P4NAME = \n", "any: P4NAME = \n"):
            with self.subTest(text=text):
                self.assertNotEqual(self.resolve(text).returncode, 0)

    def test_ambiguous_identity_and_unsupported_path_overrides_fail_closed(self):
        for flags in ("-In", "-In -d", "-In one -Intwo", "-v client.sendq.dir=override",
                      "-vclient.readonly.dir=override", "-v P4NAME=other",
                      "-v server=3,client.sendq.dir=override"):
            with self.subTest(flags=flags):
                self.assertNotEqual(self.resolve("", P4D_FLAGS=flags).returncode, 0)
        self.assertNotEqual(self.resolve("", P4DEBUG="client.sendq.dir=override").returncode, 0)

    def test_unrelated_startup_arguments_are_not_mistaken_for_overrides(self):
        flags = "-r /storage/client.sendq.dir/root -L /logs/client.readonly.dir -In P4NAME-node -v server=3"
        self.assert_resolved(self.resolve("P4NAME-node: client.sendq.dir = queues\n", P4D_FLAGS=flags),
                             "P4NAME-node", "queues")


class ReadonlyCopyResolution(SendqFixture):
    def setUp(self):
        super().setUp()
        self.offline = self.folder / "offline database"
        self.offline.mkdir()
        self.env["TEST_OFFLINE_DB"] = str(self.offline)
        self.prefix += '''
OFFLINE_DB=$TEST_OFFLINE_DB
rsync_with_preflight () {
   [[ -d "${2%/*}" ]] || { log 'Fixture preflight found a missing target parent'; return 73; }
   printf '%s\\0' "$1" "$2" "$3"
   mkdir -- "$2" && cp -R -- "$1/." "$2/"
}
'''

    def copy(self, rows, **environment):
        self.config.write_text(rows)
        return self.run_bash("copy_readonly_clients_dir", **environment)

    def persistent_client(self, relative):
        directory = self.database / relative
        directory.mkdir(parents=True)
        (directory / "db.sendq.persistent-client").write_bytes(b"persistent have table\0exact bytes")
        return directory

    def test_copy_uses_legacy_scope_and_preserves_equals_in_readonly_path(self):
        relative = "readonly clients = selected"
        self.persistent_client(relative)
        result = self.copy("local-id: client.readonly.dir = wrong\n"
                           "legacy: client.readonly.dir = " + relative + "\n", P4NAME="legacy")
        self.assertEqual(result.returncode, 0, self.log() + result.stderr.decode())
        self.assertEqual(result.stdout.decode().split("\0"),
                         [str(self.root / relative), str(self.offline / relative), "1048576", ""])
        self.assertEqual((self.offline / relative / "db.sendq.persistent-client").read_bytes(),
                         b"persistent have table\0exact bytes")
        self.assertFalse((self.offline / "wrong").exists())

    def test_copy_uses_database_p4name_redirect_and_literal_server_identity(self):
        (self.database / "server.id").write_text("edge.1\n")
        self.persistent_client("retained")
        result = self.copy("edgeX1: client.readonly.dir = wrong\n"
                           "edge.1: P4NAME = legacy\nlegacy: client.readonly.dir = retained\n")
        self.assertEqual(result.returncode, 0, self.log() + result.stderr.decode())
        self.assertTrue((self.offline / "retained" / "db.sendq.persistent-client").is_file())
        self.assertFalse((self.offline / "wrong").exists())

    def test_explicit_empty_readonly_setting_does_not_copy_global_directory(self):
        self.persistent_client("global")
        result = self.copy("any: client.readonly.dir = global\nlocal-id: client.readonly.dir = \n")
        self.assertEqual(result.returncode, 0, self.log() + result.stderr.decode())
        self.assertEqual(result.stdout, b"")
        self.assertEqual(list(self.offline.iterdir()), [])

    def test_ambiguous_config_or_read_failure_prevents_any_copy(self):
        self.persistent_client("retained")
        for environment in ({"TEST_CSHOW_FAILURE": "1"}, {"P4D_FLAGS": "-v client.readonly.dir=other"}):
            with self.subTest(environment=environment):
                result = self.copy("any: client.readonly.dir = retained\n", **environment)
                self.assertNotEqual(result.returncode, 0)
                self.assertEqual(result.stdout, b"")
                self.assertEqual(list(self.offline.iterdir()), [])

    def test_sendq_only_startup_override_does_not_block_readonly_copy(self):
        self.persistent_client("retained")
        for environment in ({"P4D_FLAGS": "-v client.sendq.dir=other"},
                            {"P4D_FLAGS": "-vserver=3,client.sendq.dir=other"},
                            {"P4DEBUG": "client.sendq.dir=other"}):
            with self.subTest(environment=environment):
                result = self.copy("any: client.readonly.dir = retained\n", **environment)
                self.assertEqual(result.returncode, 0, self.log() + result.stderr.decode())
                self.assertEqual((self.offline / "retained" / "db.sendq.persistent-client").read_bytes(),
                                 b"persistent have table\0exact bytes")
                shutil.rmtree(self.offline / "retained")

    def test_nested_destination_parent_exists_before_copy_preflight(self):
        relative = "partitioned client data/readonly = clients"
        self.persistent_client(relative)
        self.assertFalse((self.offline / "partitioned client data").exists())
        result = self.copy("local-id: client.readonly.dir = " + relative + "\n")
        self.assertEqual(result.returncode, 0, self.log() + result.stderr.decode())
        self.assertEqual((self.offline / relative / "db.sendq.persistent-client").read_bytes(),
                         b"persistent have table\0exact bytes")

    def test_parent_creation_failure_prevents_preflight_and_copy(self):
        relative = "blocked parent/readonly = clients"
        original = self.persistent_client(relative) / "db.sendq.persistent-client"
        blocker = self.offline / "blocked parent"
        blocker.write_bytes(b"existing file must survive")
        result = self.copy("local-id: client.readonly.dir = " + relative + "\n")
        self.assertNotEqual(result.returncode, 0)
        self.assertEqual(result.stdout, b"")
        self.assertIn("Cannot create the parent directory", self.log())
        self.assertEqual(blocker.read_bytes(), b"existing file must survive")
        self.assertEqual(original.read_bytes(), b"persistent have table\0exact bytes")
        self.assertEqual(list(self.offline.iterdir()), [blocker])


class SendqCleanup(SendqFixture):
    def setUp(self):
        super().setUp()
        for name in ("offline_db", "depots", "checkpoints", "journals", "logs"):
            (self.folder / name).mkdir()
        self.env.update(TEST_INIT=str(self.p4d), TEST_SERVER_STATUS="1", P4HOME=str(self.folder),
                        DEPOTS=str(self.folder / "depots"), CHECKPOINTS=str(self.folder / "checkpoints"),
                        JOURNALS=str(self.folder / "journals"), LOGS=str(self.folder / "logs"),
                        TEST_OFFLINE_DB=str(self.folder / "offline_db"))
        self.prefix += '''
P4DInitScript=$TEST_INIT
OFFLINE_DB=$TEST_OFFLINE_DB
is_server_up () { [[ "$1" == p4d ]] || return 2; return "$TEST_SERVER_STATUS"; }
'''

    def clean(self, path, readonly=None, **environment):
        rows = [] if path is None else ["local-id: client.sendq.dir = " + str(path)]
        if readonly is not None:
            rows.append("local-id: client.readonly.dir = " + str(readonly))
        self.config.write_text("\n".join(rows) + "\n")
        return self.run_bash("clean_client_sendq_dir", **environment)

    def make_queue(self, name="queue"):
        queue = self.database / name
        queue.mkdir()
        (queue / "db.sendq.fixture").write_text("ephemeral fixture")
        return queue

    def test_dedicated_directory_cleans_all_files_but_preserves_directories(self):
        queue = self.make_queue()
        (queue / ".hidden").write_text("hidden")
        (queue / "nested").mkdir()
        (queue / "nested" / "file with spaces").write_text("nested")
        result = self.clean("queue")
        self.assertEqual(result.returncode, 0, self.log() + result.stderr.decode())
        self.assertEqual(sorted(p.relative_to(queue).as_posix() for p in queue.rglob("*")), ["nested"])
        self.assertTrue((self.database / "db.config").exists())
        self.assertTrue((self.database / "server.id").exists())

    def test_absolute_directory_and_configured_root_symlink_are_supported(self):
        for use_link in (False, True):
            with self.subTest(use_link=use_link):
                queue = self.make_queue("queue-" + str(use_link))
                configured = queue
                if use_link:
                    configured = self.folder / "queue-alias"
                    configured.symlink_to(queue)
                result = self.clean(configured)
                self.assertEqual(result.returncode, 0, self.log() + result.stderr.decode())
                self.assertEqual(list(queue.iterdir()), [])
                self.assertTrue(configured.exists())

    def test_physical_queue_trailing_newline_does_not_select_sibling(self):
        queue = self.make_queue("queue\n")
        sibling = self.make_queue("queue")
        (sibling / "keep").write_bytes(b"unrelated sibling exact bytes\0")
        configured = self.folder / "configured-queue"
        configured.symlink_to(queue)
        result = self.clean(configured)
        self.assertEqual(result.returncode, 0, self.log() + result.stderr.decode())
        self.assertTrue(configured.is_dir())
        self.assertEqual(list(queue.iterdir()), [])
        self.assertEqual((sibling / "db.sendq.fixture").read_text(), "ephemeral fixture")
        self.assertEqual((sibling / "keep").read_bytes(), b"unrelated sibling exact bytes\0")

    def test_nested_symlinks_are_unlinked_without_touching_their_targets(self):
        queue = self.make_queue()
        target = self.folder / "outside"
        target.mkdir()
        (target / "keep").write_text("persistent")
        (queue / "linked-dir").symlink_to(target)
        (queue / "linked-file").symlink_to(target / "keep")
        (queue / "broken-link").symlink_to(target / "absent")
        result = self.clean(queue)
        self.assertEqual(result.returncode, 0, self.log() + result.stderr.decode())
        self.assertEqual((target / "keep").read_text(), "persistent")
        self.assertEqual(list(queue.iterdir()), [])

    def test_unset_empty_and_missing_directories_are_successful_noops(self):
        queue = self.make_queue()
        for setting in (None, "", "not-created"):
            with self.subTest(setting=setting):
                result = self.clean(setting)
                self.assertEqual(result.returncode, 0, self.log() + result.stderr.decode())
                self.assertTrue((queue / "db.sendq.fixture").exists())

    def test_service_running_unknown_status_or_missing_init_refuses_cleanup(self):
        queue = self.make_queue()
        for overrides in ({"TEST_SERVER_STATUS": "0"}, {"TEST_SERVER_STATUS": "2"},
                          {"TEST_INIT": str(self.folder / "absent-init")}):
            with self.subTest(overrides=overrides):
                self.assertNotEqual(self.clean(queue, **overrides).returncode, 0)
                self.assertTrue((queue / "db.sendq.fixture").exists())

    def test_live_or_malformed_pid_is_not_mistaken_for_a_stopped_server(self):
        queue = self.make_queue()
        for value in (str(os.getpid()), "not-a-pid", "0", "12\n34"):
            with self.subTest(value=value):
                (self.database / "server.pid").write_text(value)
                self.assertNotEqual(self.clean(queue).returncode, 0)
                self.assertTrue((queue / "db.sendq.fixture").exists())

    def test_config_failure_prevents_any_deletion(self):
        queue = self.make_queue()
        result = self.clean(queue, TEST_CSHOW_FAILURE="1")
        self.assertNotEqual(result.returncode, 0)
        self.assertTrue((queue / "db.sendq.fixture").exists())

    def test_protected_roots_and_ancestors_are_not_cleaned(self):
        targets = ["/", self.root, self.database, self.folder,
                   *(self.folder / name for name in ("offline_db", "depots", "checkpoints", "journals", "logs"))]
        for target in targets:
            with self.subTest(target=target):
                self.assertNotEqual(self.clean(target).returncode, 0)
                self.assertTrue((self.database / "db.config").exists())
        alias = self.folder / "unsafe-alias"
        alias.symlink_to(self.database)
        self.assertNotEqual(self.clean(alias).returncode, 0)

    def test_file_or_dangling_configured_symlink_is_an_error(self):
        target = self.folder / "ordinary-file"
        target.write_text("keep")
        alias = self.folder / "dangling"
        alias.symlink_to(self.folder / "absent")
        for path in (target, alias):
            with self.subTest(path=path):
                self.assertNotEqual(self.clean(path).returncode, 0)
        self.assertEqual(target.read_text(), "keep")

    def test_partial_deletion_failure_is_propagated(self):
        queue = self.make_queue()
        (queue / "remaining").write_text("keep on failure")
        mock_find = self.mockbin / "find"
        mock_find.write_text("#!" + sys.executable + '''
import pathlib,sys
(pathlib.Path(sys.argv[1]) / "db.sendq.fixture").unlink()
print("injected find failure", file=sys.stderr)
sys.exit(9)
''')
        mock_find.chmod(0o755)
        result = self.clean(queue)
        self.assertNotEqual(result.returncode, 0)
        self.assertFalse((queue / "db.sendq.fixture").exists())
        self.assertTrue((queue / "remaining").exists())
        self.assertIn("injected find failure", self.log())

    def test_shared_or_nested_readonly_directory_is_warned_and_left_untouched(self):
        for position in ("same", "readonly-inside", "queue-inside", "symlink-alias"):
            with self.subTest(position=position):
                parent = self.database / ("shared-" + position)
                parent.mkdir()
                queue = parent
                readonly = parent
                if position == "readonly-inside":
                    readonly = parent / "client-data"
                    readonly.mkdir()
                elif position == "queue-inside":
                    queue = parent / "queue-data"
                    queue.mkdir()
                elif position == "symlink-alias":
                    readonly = self.database / "readonly-alias"
                    readonly.symlink_to(parent)
                (queue / "db.sendq.ephemeral").write_text("queue")
                (readonly / "db.sendq.named-persistent-client").write_text("have table")
                result = self.clean(queue, readonly)
                self.assertEqual(result.returncode, 0, self.log() + result.stderr.decode())
                self.assertEqual((queue / "db.sendq.ephemeral").read_text(), "queue")
                self.assertEqual((readonly / "db.sendq.named-persistent-client").read_text(), "have table")
                self.assertIn("warn", self.log().lower())

    def assert_readonly_alias_preserves_all_queue_entries(self, queue, readonly):
        (queue / ".hidden").write_bytes(b"shared hidden file")
        (queue / "extra-nested").mkdir()
        (queue / "extra-nested" / "keep").write_bytes(b"shared nested file")
        (queue / "unrelated-broken-link").symlink_to(self.folder / "absent target")
        persistent = readonly / "db.sendq.named-persistent-client"
        persistent.write_bytes(b"persistent have table\0exact bytes")

        def inventory():
            return {str(path.relative_to(queue)):
                    ("link", os.readlink(path)) if path.is_symlink() else
                    ("file", path.read_bytes()) if path.is_file() else ("directory",)
                    for path in queue.rglob("*")}

        before = inventory()
        result = self.clean(queue, readonly)
        self.assertEqual(result.returncode, 0, self.log() + result.stderr.decode())
        self.assertEqual(inventory(), before)
        self.assertTrue(readonly.is_dir(), "cleanup broke the configured readonly path")
        self.assertEqual(persistent.read_bytes(), b"persistent have table\0exact bytes")
        self.assertIn("warn", self.log().lower())
        self.assertIn("preserving all", self.log().lower())

    def test_readonly_symlink_inside_queue_preserves_all_shared_entries(self):
        queue = self.make_queue()
        persistent = self.folder / "persistent outside"
        persistent.mkdir()
        readonly = queue / "readonly-link"
        readonly.symlink_to(persistent)
        self.assert_readonly_alias_preserves_all_queue_entries(queue, readonly)

    def test_readonly_parent_symlink_inside_queue_preserves_all_shared_entries(self):
        queue = self.make_queue()
        persistent = self.folder / "persistent outside"
        (persistent / "client-data").mkdir(parents=True)
        alias = queue / "parent-link"
        alias.symlink_to(persistent)
        self.assert_readonly_alias_preserves_all_queue_entries(queue, alias / "client-data")

    def test_outside_readonly_alias_chained_through_queue_preserves_all_shared_entries(self):
        queue = self.make_queue()
        persistent = self.folder / "persistent outside"
        persistent.mkdir()
        queue_alias = queue / "readonly-link"
        queue_alias.symlink_to(persistent)
        readonly = self.folder / "outside-readonly-alias"
        readonly.symlink_to(queue_alias)
        self.assert_readonly_alias_preserves_all_queue_entries(queue, readonly)

    def test_relative_alias_targets_and_parent_components_preserve_shared_entries(self):
        queue = self.make_queue()
        persistent = self.folder / "persistent outside"
        (persistent / "nested").mkdir(parents=True)
        (persistent / "client-data").mkdir()
        queue_alias = queue / "readonly-link"
        queue_alias.symlink_to("../../persistent outside")
        outer = self.folder / "outside aliases"
        outer.mkdir()
        readonly = outer / "relative-link"
        readonly.symlink_to("../actual database/queue/readonly-link/nested/../client-data")
        self.assert_readonly_alias_preserves_all_queue_entries(queue, readonly)

    def test_trailing_newline_in_intermediate_link_target_is_preserved(self):
        queue = self.make_queue()
        persistent = self.folder / "persistent outside"
        persistent.mkdir()
        queue_alias = queue / "readonly-link\n"
        queue_alias.symlink_to(persistent)
        readonly = self.folder / "outside-readonly-alias"
        readonly.symlink_to(queue_alias)
        self.assert_readonly_alias_preserves_all_queue_entries(queue, readonly)

    def test_protected_configured_paths_depending_on_queue_alias_are_rejected(self):
        for variable, target, indirect in (("P4ROOT", self.database, False),
                                           ("P4HOME", self.folder, True),
                                           ("LOGS", self.folder / "logs", True)):
            with self.subTest(variable=variable):
                queue = self.make_queue("queue-" + variable.lower())
                alias = queue / "required-storage-alias"
                alias.symlink_to(target)
                configured = alias
                if indirect:
                    configured = self.folder / ("outside-" + variable.lower() + "-alias")
                    configured.symlink_to(alias)
                before = (self.database / "db.config").read_bytes()
                result = self.clean(queue, **{variable: str(configured)})
                self.assertNotEqual(result.returncode, 0, self.log() + result.stderr.decode())
                self.assertTrue(alias.is_symlink())
                self.assertTrue(configured.is_dir())
                self.assertEqual((queue / "db.sendq.fixture").read_text(), "ephemeral fixture")
                self.assertEqual((self.database / "db.config").read_bytes(), before)

    def test_sendq_path_depending_on_its_own_nested_alias_is_rejected(self):
        queue = self.make_queue()
        inner_alias = queue / "back-to-queue"
        inner_alias.symlink_to(queue)
        configured = self.folder / "configured-sendq"
        configured.symlink_to(inner_alias)
        result = self.clean(configured)
        self.assertNotEqual(result.returncode, 0, self.log() + result.stderr.decode())
        self.assertTrue(inner_alias.is_symlink())
        self.assertTrue(configured.is_dir())
        self.assertEqual((queue / "db.sendq.fixture").read_text(), "ephemeral fixture")

    def test_configured_journal_alias_chain_is_rejected_but_external_file_is_safe(self):
        queue = self.make_queue()
        journal = self.folder / "journals" / "journal"
        journal.write_bytes(b"live journal exact bytes\0")
        inner_alias = queue / "live-journal-link"
        inner_alias.symlink_to(journal)
        configured = self.folder / "configured-journal"
        configured.symlink_to(inner_alias)
        result = self.clean(queue, P4JOURNAL=str(configured), P4JOURNALDIR=str(configured.parent))
        self.assertNotEqual(result.returncode, 0, self.log() + result.stderr.decode())
        self.assertTrue(inner_alias.is_symlink())
        self.assertEqual(configured.read_bytes(), b"live journal exact bytes\0")
        self.assertEqual((queue / "db.sendq.fixture").read_text(), "ephemeral fixture")

        result = self.clean(queue, P4JOURNAL=str(journal), P4JOURNALDIR=str(journal.parent))
        self.assertEqual(result.returncode, 0, self.log() + result.stderr.decode())
        self.assertEqual(list(queue.iterdir()), [])
        self.assertEqual(journal.read_bytes(), b"live journal exact bytes\0")

    def test_actual_journal_file_in_queue_is_protected_with_relative_setting(self):
        queue = self.make_queue()
        journal = queue / "journal"
        journal.write_bytes(b"live journal exact bytes\0")
        # Keep the separately configured volume outside the queue so this
        # exercises the actual P4JOURNAL file guard and its P4ROOT-relative base.
        result = self.clean(queue, P4JOURNAL="queue/journal", P4JOURNALDIR=str(self.folder / "journals"))
        self.assertNotEqual(result.returncode, 0, self.log() + result.stderr.decode())
        self.assertEqual(journal.read_bytes(), b"live journal exact bytes\0")
        self.assertEqual((queue / "db.sendq.fixture").read_text(), "ephemeral fixture")

    def test_unrelated_nested_alias_is_removed_with_separate_readonly_alias(self):
        queue = self.make_queue()
        persistent = self.folder / "persistent outside"
        persistent.mkdir()
        (persistent / "keep").write_bytes(b"persistent exact bytes")
        readonly = self.folder / "readonly-alias"
        readonly.symlink_to(persistent)
        (queue / "unrelated-link").symlink_to(persistent)
        result = self.clean(queue, readonly)
        self.assertEqual(result.returncode, 0, self.log() + result.stderr.decode())
        self.assertEqual(list(queue.iterdir()), [])
        self.assertTrue(readonly.is_dir())
        self.assertEqual((readonly / "keep").read_bytes(), b"persistent exact bytes")

    def test_filesystem_root_readonly_setting_is_treated_as_queue_ancestor(self):
        queue = self.make_queue()
        result = self.clean(queue, "/")
        self.assertEqual(result.returncode, 0, self.log() + result.stderr.decode())
        self.assertEqual((queue / "db.sendq.fixture").read_text(), "ephemeral fixture")
        self.assertIn("preserving all", self.log().lower())

    def test_readonly_symlink_loop_stops_before_deleting_queue_files(self):
        queue = self.make_queue()
        readonly = self.folder / "readonly-loop"
        readonly.symlink_to("readonly-loop")
        result = self.clean(queue, readonly)
        self.assertNotEqual(result.returncode, 0)
        self.assertEqual((queue / "db.sendq.fixture").read_text(), "ephemeral fixture")
        self.assertTrue(readonly.is_symlink())

    def test_readonly_alias_read_failure_stops_before_deleting_queue_files(self):
        queue = self.make_queue()
        persistent = self.folder / "persistent outside"
        persistent.mkdir()
        (persistent / "keep").write_bytes(b"persistent exact bytes")
        readonly = self.folder / "readonly-alias"
        readonly.symlink_to(persistent)
        mock_readlink = self.mockbin / "readlink"
        mock_readlink.write_text("#!/bin/sh\nprintf 'injected readlink failure\\n' >&2\nexit 9\n")
        mock_readlink.chmod(0o755)
        result = self.clean(queue, readonly)
        self.assertNotEqual(result.returncode, 0)
        self.assertEqual((queue / "db.sendq.fixture").read_text(), "ephemeral fixture")
        self.assertTrue(readonly.is_dir())
        self.assertEqual((readonly / "keep").read_bytes(), b"persistent exact bytes")


if __name__ == "__main__":
    unittest.main()