test_sendq_switch.py #1

"""Opt-in sendq cleanup around real temporary database/symlink switches.

The cleanup helper and service commands are stubs; helper path safety and live
P4 configuration are exercised by their separate suites. Filesystem switching,
metadata preservation, rollback, and retained recovery copies are real here.
"""
import unittest

import test_runtime_improvements as runtime_fixtures


class SendqSwitch(unittest.TestCase):
    def setUp(self):
        self.fixture = runtime_fixtures.RuntimeImprovements(methodName="runTest")
        self.fixture.setUp()
        self.addCleanup(self.fixture.doCleanups)
        f = self.fixture
        f.prepare_switch()
        self.old_queue = f.old / "old sendq"
        self.new_queue = f.new / "candidate sendq"
        for root, queue in ((f.old, self.old_queue), (f.new, self.new_queue)):
            queue.mkdir()
            (queue / "queue.tmp").write_text("temporary queue at " + str(root))
            (root / ".fixture-sendq-config").write_text(queue.name)
        self.state = f.root / "service-state"
        self.state.write_text("stopped")
        self.trace_file = f.root / "switch-trace"
        f.prefix += '''
compare_journal_numbers () {
  printf 'compare|%s\\n' "$(cd "$P4ROOT" && pwd -P)" >> "$TEST_ROOT/switch-trace"
}
copy_readonly_clients_dir () {
  printf 'copy|%s|%s\\n' "$(cd "$P4ROOT" && pwd -P)" "$(cd "$OFFLINE_DB" && pwd -P)" >> "$TEST_ROOT/switch-trace"
}
clean_client_sendq_dir () {
  local physical= configured= state=
  physical=$(cd "$P4ROOT" && pwd -P) || return 41
  configured=$(cat "$P4ROOT/.fixture-sendq-config") || return 42
  state=$(cat "$TEST_ROOT/service-state") || return 43
  printf 'clean|%s|%s|%s\\n' "$physical" "$configured" "$state" >> "$TEST_ROOT/switch-trace"
  [[ "$state" == stopped ]] || { log 'ERROR: attempted cleanup with service running'; return 44; }
  if [[ "$physical" == "${FAIL_CLEAN_ROOT:-}" ]]; then
    log "Injected sendq cleanup failure for $physical"
    return 45
  fi
  rm -- "$P4ROOT/$configured/queue.tmp" || return 46
}
start_p4d () {
  printf 'start|%s\\n' "$(cd "$P4ROOT" && pwd -P)" >> "$TEST_ROOT/switch-trace"
  printf started > "$TEST_ROOT/service-state"
}
'''

    def trace(self):
        return [line.split("|") for line in self.trace_file.read_text().splitlines()]

    def assert_original_database(self):
        f = self.fixture
        self.assertEqual((f.root / "root").resolve(), f.old)
        self.assertEqual((f.root / "offline_db").resolve(), f.new)
        self.assertEqual((f.old / "db.counters").read_text(), "old counters")
        self.assertEqual((f.old / "db.rev").read_text(), "old revisions")
        self.assertEqual((f.old / "server.id").read_text(), "original server")
        self.assertEqual((f.new / "server.id").read_text(), "offline server")
        self.assertEqual((f.old / "save/db.counters").read_text(), "prior saved counters")

    def test_each_physical_root_uses_its_local_config_while_stopped(self):
        f = self.fixture
        result = f.run_bash("switch_db_files clean_sendq")
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(self.trace(), [
            ["compare", str(f.old)],
            ["clean", str(f.old), self.old_queue.name, "stopped"],
            ["copy", str(f.old), str(f.new)],
            ["clean", str(f.new), self.new_queue.name, "stopped"],
        ])
        self.assertEqual(self.state.read_text(), "stopped")
        self.assertFalse((self.old_queue / "queue.tmp").exists())
        self.assertFalse((self.new_queue / "queue.tmp").exists())
        self.assertEqual((f.root / "root").resolve(), f.new)
        self.assertEqual((f.root / "offline_db").resolve(), f.old)

    def test_first_cleanup_failure_preserves_database_links_and_stays_down(self):
        f = self.fixture
        result = f.run_bash("switch_db_files clean_sendq", {"FAIL_CLEAN_ROOT": str(f.old)})
        self.assertEqual(result.returncode, 97)
        self.assert_original_database()
        self.assertEqual(self.trace(), [
            ["compare", str(f.old)], ["clean", str(f.old), self.old_queue.name, "stopped"],
        ])
        self.assertEqual(self.state.read_text(), "stopped")
        self.assertTrue((self.old_queue / "queue.tmp").exists())
        self.assertTrue((self.new_queue / "queue.tmp").exists())
        self.assertFalse(list(f.old.glob(".sdp-switch.*")))
        self.assertIn("Injected sendq cleanup failure", (f.logs / "checkpoint.log").read_text())

    def test_second_cleanup_failure_rolls_back_and_restarts_original(self):
        f = self.fixture
        result = f.run_bash("switch_db_files clean_sendq", {"FAIL_CLEAN_ROOT": str(f.new)})
        self.assertEqual(result.returncode, 97)
        self.assert_original_database()
        self.assertEqual(self.trace(), [
            ["compare", str(f.old)],
            ["clean", str(f.old), self.old_queue.name, "stopped"],
            ["copy", str(f.old), str(f.new)],
            ["clean", str(f.new), self.new_queue.name, "stopped"],
            ["start", str(f.old)],
        ])
        self.assertEqual(self.state.read_text(), "started")
        self.assertFalse((self.old_queue / "queue.tmp").exists())
        self.assertTrue((self.new_queue / "queue.tmp").exists())
        self.assertFalse((f.old / "P4ROOT_not_usable.txt").exists())
        self.assertFalse((f.new / "P4ROOT_not_usable.txt").exists())
        self.assertIn("rolled back", result.stderr.lower())

    def test_success_retains_previous_database_until_completion(self):
        f = self.fixture
        result = f.run_bash("switch_db_files clean_sendq")
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual((f.old / "save/db.counters").read_text(), "old counters")
        self.assertEqual((f.old / "save/db.rev").read_text(), "old revisions")
        self.assertEqual((f.new / "db.counters").read_text(), "new counters")
        stages = list(f.old.glob(".sdp-switch.*"))
        self.assertEqual(len(stages), 1)
        self.assertEqual((stages[0] / "previous-save/db.counters").read_text(), "prior saved counters")
        self.assertEqual(self.state.read_text(), "stopped")

    def test_recovery_copy_is_removed_only_after_replacement_start_and_completion(self):
        f = self.fixture
        result = f.run_bash('''switch_db_files clean_sendq
[[ -f "$SWITCH_DB_OLDROOT/save/db.counters" ]] || exit 70
start_p4d
complete_db_switch
''')
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(self.trace()[-1], ["start", str(f.new)])
        self.assertFalse((f.old / "save/db.counters").exists())
        self.assertFalse(list(f.old.glob(".sdp-switch.*")))
        self.assertEqual((f.new / "db.counters").read_text(), "new counters")

    def test_callers_without_opt_in_do_not_clean_either_root(self):
        f = self.fixture
        result = f.run_bash("switch_db_files")
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(self.trace(), [["compare", str(f.old)], ["copy", str(f.old), str(f.new)]])
        self.assertTrue((self.old_queue / "queue.tmp").exists())
        self.assertTrue((self.new_queue / "queue.tmp").exists())
        self.assertEqual((f.old / "save/db.counters").read_text(), "old counters")


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