#!/usr/bin/env python3
"""
Unit tests for manual_compaction.py.

All HTTP calls are mocked; no real PD or TiKV cluster is required.

Run:
    python3 -m pytest scripts/manual_compaction/test_manual_compaction.py -v
  or:
    python3 scripts/manual_compaction/test_manual_compaction.py
"""

import os
import sys
import tempfile
import unittest
from unittest.mock import MagicMock, patch

sys.path.insert(0, os.path.join(os.path.dirname(__file__)))
import manual_compaction as mc


def make_region(region_id, start_hex, end_hex, leader_store_id=1):
    return {
        "id": region_id,
        "start_key": start_hex,
        "end_key": end_hex,
        "leader": {"id": region_id * 10, "store_id": leader_store_id},
        "peers": [{"id": region_id * 10, "store_id": leader_store_id}],
    }


def make_store_response(store_id, status_address):
    return {
        "store": {
            "id": store_id,
            "address": f"127.0.0.1:{10000 + store_id}",
            "status_address": status_address,
            "state": 0,
            "state_name": "Up",
        },
        "status": {},
    }


class TestHexKeyToBytes(unittest.TestCase):
    def test_empty(self):
        self.assertEqual(mc.hex_key_to_bytes(""), b"")

    def test_ascii(self):
        self.assertEqual(mc.hex_key_to_bytes("4142"), b"AB")

    def test_binary(self):
        self.assertEqual(mc.hex_key_to_bytes("00ff"), b"\x00\xff")

    def test_invalid_returns_empty(self):
        self.assertEqual(mc.hex_key_to_bytes("gg"), b"")


class TestNormalizeStatusAddress(unittest.TestCase):
    def test_plain_host_port_http(self):
        self.assertEqual(mc.normalize_status_address("127.0.0.1:20292", False), "http://127.0.0.1:20292")

    def test_plain_host_port_https(self):
        self.assertEqual(mc.normalize_status_address("127.0.0.1:20292", True), "https://127.0.0.1:20292")

    def test_already_http(self):
        self.assertEqual(mc.normalize_status_address("http://127.0.0.1:20292", False), "http://127.0.0.1:20292")

    def test_already_https(self):
        self.assertEqual(mc.normalize_status_address("https://127.0.0.1:20292", True), "https://127.0.0.1:20292")


class TestGetRegionEngineStats(unittest.TestCase):
    @patch("manual_compaction.http_get")
    def test_returns_tombs_and_old_entries(self, mock_get):
        mock_get.return_value = {
            "active": True,
            "entries": 100,
            "old-entries": 20,
            "tombs": 5,
        }

        stats = mc.get_region_engine_stats("http://127.0.0.1:20292", 42, None, 10)

        self.assertEqual(stats, {"active": True, "entries": 100, "old_entries": 20, "tombs": 5})


class TestIsEligible(unittest.TestCase):
    def test_inactive(self):
        ok, reason = mc.is_eligible({"active": False, "entries": 100, "old_entries": 50}, 0.3)
        self.assertFalse(ok)
        self.assertEqual(reason, "inactive")

    def test_zero_entries(self):
        ok, reason = mc.is_eligible({"active": True, "entries": 0, "old_entries": 0}, 0.3)
        self.assertFalse(ok)
        self.assertEqual(reason, "zero-entries")

    def test_below_threshold(self):
        ok, reason = mc.is_eligible({"active": True, "entries": 100, "old_entries": 10}, 0.3)
        self.assertFalse(ok)
        self.assertIn("below-threshold", reason)

    def test_above_threshold(self):
        ok, reason = mc.is_eligible({"active": True, "entries": 50, "old_entries": 60}, 0.3)
        self.assertTrue(ok)
        self.assertIsNone(reason)

    def test_tombs_contribute_to_threshold(self):
        ok, reason = mc.is_eligible({"active": True, "entries": 100, "old_entries": 10, "tombs": 30}, 0.3)
        self.assertTrue(ok)
        self.assertIsNone(reason)

    def test_exactly_at_threshold_is_not_eligible(self):
        ok, reason = mc.is_eligible({"active": True, "entries": 70, "old_entries": 30}, 0.3)
        self.assertFalse(ok)

    def test_just_above_threshold(self):
        ok, reason = mc.is_eligible({"active": True, "entries": 69, "old_entries": 31}, 0.3)
        self.assertTrue(ok)


class TestCheckpoint(unittest.TestCase):
    def test_write_and_load(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = os.path.join(tmp, "ckpt.json")
            data = mc.build_checkpoint(b"\x01\x02", 42, "0102", "0304", {"triggered": 5})
            mc.write_checkpoint(path, data)
            loaded = mc.load_checkpoint(path)
            self.assertEqual(loaded["scan_key"], "0102")
            self.assertEqual(loaded["last_region_id"], 42)
            self.assertEqual(loaded["counters"]["triggered"], 5)
            self.assertEqual(loaded["version"], mc.CHECKPOINT_VERSION)

    def test_load_nonexistent_returns_none(self):
        self.assertIsNone(mc.load_checkpoint("/tmp/__no_such_file__.json"))

    def test_atomic_write_no_partial_read(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = os.path.join(tmp, "ckpt.json")
            data1 = mc.build_checkpoint(b"\x01", 1, "01", "02", {"triggered": 1})
            mc.write_checkpoint(path, data1)
            data2 = mc.build_checkpoint(b"\x03", 2, "03", "04", {"triggered": 2})
            mc.write_checkpoint(path, data2)
            loaded = mc.load_checkpoint(path)
            self.assertEqual(loaded["scan_key"], "03")

    def test_empty_scan_key(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = os.path.join(tmp, "ckpt.json")
            data = mc.build_checkpoint(b"", None, "", "", {})
            mc.write_checkpoint(path, data)
            loaded = mc.load_checkpoint(path)
            self.assertEqual(loaded["scan_key"], "")


class TestScanRegionsBatch(unittest.TestCase):
    @patch("manual_compaction.pd_request")
    def test_basic_request(self, mock_pd):
        mock_pd.return_value = {
            "count": 1,
            "regions": [make_region(1, "4100", "4200", 1)],
        }
        regions = mc.scan_regions_batch(["http://pd:2379"], b"A", b"Z", 16, None, 10)
        self.assertEqual(len(regions), 1)
        self.assertEqual(regions[0]["id"], 1)
        mock_pd.assert_called_once()
        args = mock_pd.call_args
        params = args[0][4] if len(args[0]) > 4 else args[1].get("params", {})
        self.assertIn("key", params)
        self.assertIn("limit", params)
        self.assertEqual(params.get("format"), "hex")
        self.assertEqual(params["key"], b"A".hex())

    @patch("manual_compaction.pd_request")
    def test_empty_start_key(self, mock_pd):
        mock_pd.return_value = {"count": 0, "regions": []}
        mc.scan_regions_batch(["http://pd:2379"], b"", b"", 16, None, 10)
        params = mock_pd.call_args[1].get("params", mock_pd.call_args[0][4] if len(mock_pd.call_args[0]) > 4 else {})
        self.assertEqual(params.get("key", None), "")
        self.assertEqual(params.get("format"), "hex")

    @patch("manual_compaction.pd_request")
    def test_no_regions_returns_empty(self, mock_pd):
        mock_pd.return_value = {"count": 0, "regions": []}
        result = mc.scan_regions_batch(["http://pd:2379"], b"", b"", 16, None, 10)
        self.assertEqual(result, [])

    @patch("manual_compaction.pd_request")
    def test_missing_regions_key(self, mock_pd):
        mock_pd.return_value = {"count": 0}
        result = mc.scan_regions_batch(["http://pd:2379"], b"", b"", 16, None, 10)
        self.assertEqual(result, [])


class TestGetStoreStatusAddress(unittest.TestCase):
    @patch("manual_compaction.pd_request")
    def test_resolves_status_address(self, mock_pd):
        mock_pd.return_value = make_store_response(1, "127.0.0.1:20292")
        cache = {}
        url = mc.get_store_status_address(["http://pd:2379"], 1, None, 10, False, cache)
        self.assertEqual(url, "http://127.0.0.1:20292")
        self.assertEqual(cache[1], "http://127.0.0.1:20292")

    @patch("manual_compaction.pd_request")
    def test_caches_result(self, mock_pd):
        mock_pd.return_value = make_store_response(2, "127.0.0.1:20293")
        cache = {}
        mc.get_store_status_address(["http://pd:2379"], 2, None, 10, False, cache)
        mc.get_store_status_address(["http://pd:2379"], 2, None, 10, False, cache)
        self.assertEqual(mock_pd.call_count, 1)

    def test_uses_cache_without_pd_call(self):
        cache = {3: "http://127.0.0.1:30000"}
        with patch("manual_compaction.pd_request") as mock_pd:
            url = mc.get_store_status_address(["http://pd:2379"], 3, None, 10, False, cache)
            mock_pd.assert_not_called()
        self.assertEqual(url, "http://127.0.0.1:30000")

    @patch("manual_compaction.pd_request")
    def test_falls_back_to_address(self, mock_pd):
        mock_pd.return_value = {
            "store": {"id": 4, "address": "127.0.0.1:20294"},
            "status": {},
        }
        cache = {}
        url = mc.get_store_status_address(["http://pd:2379"], 4, None, 10, False, cache)
        self.assertEqual(url, "http://127.0.0.1:20294")

    @patch("manual_compaction.pd_request")
    def test_raises_on_no_address(self, mock_pd):
        mock_pd.return_value = {"store": {"id": 5}, "status": {}}
        cache = {}
        with self.assertRaises(RuntimeError):
            mc.get_store_status_address(["http://pd:2379"], 5, None, 10, False, cache)

    @patch("manual_compaction.pd_request")
    def test_raises_on_pd_failure(self, mock_pd):
        mock_pd.side_effect = RuntimeError("all pd down")
        cache = {}
        with self.assertRaises(RuntimeError):
            mc.get_store_status_address(["http://pd:2379"], 6, None, 10, False, cache)


class TestRunCompaction(unittest.TestCase):
    def _args(self, **kwargs):
        defaults = dict(
            pd="http://pd:2379",
            start_key="",
            end_key="",
            region_limit=128,
            old_ratio=0.3,
            interval=0,
            scan_interval=0,
            timeout=10,
            dry_run=False,
            checkpoint_file="",
            ignore_checkpoint=False,
            checkpoint_interval=0,
            ca_cert="",
            client_cert="",
            client_key="",
        )
        defaults.update(kwargs)
        return MagicMock(**defaults)

    @patch("manual_compaction.http_post")
    @patch("manual_compaction.http_get")
    @patch("manual_compaction.pd_request")
    def test_single_eligible_region_triggered(self, mock_pd, mock_get, mock_post):
        region = make_region(1, "4100", "4200", 1)
        mock_pd.side_effect = [
            {"count": 1, "regions": [region]},
            make_store_response(1, "127.0.0.1:20292"),
        ]
        mock_get.return_value = {"active": True, "entries": 50, "old-entries": 60}
        mock_post.return_value = (200, "ok")

        args = self._args()
        rc = mc.run_compaction(args, ["http://pd:2379"], None)
        self.assertEqual(rc, 0)
        mock_post.assert_called_once()
        call_url = mock_post.call_args[0][0]
        self.assertIn("/major-compact", call_url)

    @patch("manual_compaction.http_get")
    @patch("manual_compaction.pd_request")
    def test_dry_run_no_post(self, mock_pd, mock_get):
        region = make_region(1, "4100", "4200", 1)
        mock_pd.side_effect = [
            {"count": 1, "regions": [region]},
            make_store_response(1, "127.0.0.1:20292"),
        ]
        mock_get.return_value = {"active": True, "entries": 10, "old-entries": 90}

        args = self._args(dry_run=True)
        with patch("manual_compaction.http_post") as mock_post:
            rc = mc.run_compaction(args, ["http://pd:2379"], None)
            mock_post.assert_not_called()
        self.assertEqual(rc, 0)

    @patch("manual_compaction.pd_request")
    def test_stop_when_no_regions(self, mock_pd):
        mock_pd.return_value = {"count": 0, "regions": []}
        args = self._args()
        rc = mc.run_compaction(args, ["http://pd:2379"], None)
        self.assertEqual(rc, 0)

    @patch("manual_compaction.http_get")
    @patch("manual_compaction.pd_request")
    def test_stop_at_end_key(self, mock_pd, mock_get):
        region = make_region(1, "4100", "4200", 1)
        mock_pd.side_effect = [
            {"count": 1, "regions": [region]},
            make_store_response(1, "127.0.0.1:20292"),
            {"count": 0, "regions": []},
        ]
        mock_get.return_value = {"active": False, "entries": 0, "old-entries": 0}

        args = self._args(start_key="41", end_key="42")  # 'A' to 'B'
        rc = mc.run_compaction(args, ["http://pd:2379"], None)
        self.assertEqual(rc, 0)

    @patch("manual_compaction.pd_request")
    def test_missing_leader_counts_as_failure(self, mock_pd):
        region = {"id": 1, "start_key": "4100", "end_key": "4200"}
        mock_pd.return_value = {"count": 1, "regions": [region]}
        args = self._args()
        rc = mc.run_compaction(args, ["http://pd:2379"], None)
        self.assertEqual(rc, 1)

    @patch("manual_compaction.http_get")
    @patch("manual_compaction.pd_request")
    def test_stats_failure_increments_counter(self, mock_pd, mock_get):
        region = make_region(1, "4100", "4200", 1)
        mock_pd.side_effect = [
            {"count": 1, "regions": [region]},
            make_store_response(1, "127.0.0.1:20292"),
        ]
        mock_get.side_effect = RuntimeError("stats error")
        args = self._args()
        rc = mc.run_compaction(args, ["http://pd:2379"], None)
        self.assertEqual(rc, 1)

    @patch("manual_compaction.http_post")
    @patch("manual_compaction.http_get")
    @patch("manual_compaction.pd_request")
    def test_trigger_failure_returns_nonzero(self, mock_pd, mock_get, mock_post):
        region = make_region(1, "4100", "4200", 1)
        mock_pd.side_effect = [
            {"count": 1, "regions": [region]},
            make_store_response(1, "127.0.0.1:20292"),
        ]
        mock_get.return_value = {"active": True, "entries": 10, "old-entries": 90}
        mock_post.return_value = (500, "internal error")
        args = self._args()
        rc = mc.run_compaction(args, ["http://pd:2379"], None)
        self.assertEqual(rc, 1)

    @patch("manual_compaction.pd_request")
    def test_non_advancing_region_stops_and_errors(self, mock_pd):
        region = make_region(1, "4100", "4100", 1)
        mock_pd.side_effect = [
            {"count": 1, "regions": [region]},
            make_store_response(1, "127.0.0.1:20292"),
        ]
        args = self._args(start_key="41")
        rc = mc.run_compaction(args, ["http://pd:2379"], None)
        self.assertEqual(rc, 1)

    @patch("manual_compaction.http_post")
    @patch("manual_compaction.http_get")
    @patch("manual_compaction.pd_request")
    def test_pagination_advances_by_end_key(self, mock_pd, mock_get, mock_post):
        region1 = make_region(1, "4100", "4200", 1)
        region2 = make_region(2, "4200", "4300", 1)
        mock_pd.side_effect = [
            {"count": 1, "regions": [region1]},
            make_store_response(1, "127.0.0.1:20292"),
            {"count": 1, "regions": [region2]},
            {"count": 0, "regions": []},
        ]
        mock_get.return_value = {"active": False, "entries": 0, "old-entries": 0}
        mock_post.return_value = (200, "ok")

        args = self._args(region_limit=1)
        rc = mc.run_compaction(args, ["http://pd:2379"], None)
        self.assertEqual(rc, 0)
        pd_calls = mock_pd.call_args_list
        regions_key_calls = [c for c in pd_calls if mc.PD_REGIONS_KEY_PATH in str(c)]
        self.assertGreaterEqual(len(regions_key_calls), 2)

    @patch("manual_compaction.http_post")
    @patch("manual_compaction.http_get")
    @patch("manual_compaction.pd_request")
    def test_checkpoint_written_after_each_batch(self, mock_pd, mock_get, mock_post):
        region = make_region(1, "4100", "4200", 1)
        mock_pd.side_effect = [
            {"count": 1, "regions": [region]},
            make_store_response(1, "127.0.0.1:20292"),
            {"count": 0, "regions": []},
        ]
        mock_get.return_value = {"active": False, "entries": 0, "old-entries": 0}
        mock_post.return_value = (200, "ok")

        with tempfile.TemporaryDirectory() as tmp:
            ckpt_path = os.path.join(tmp, "ckpt.json")
            args = self._args(checkpoint_file=ckpt_path, region_limit=1)
            rc = mc.run_compaction(args, ["http://pd:2379"], None)
            self.assertEqual(rc, 0)
            self.assertTrue(os.path.exists(ckpt_path))
            ckpt = mc.load_checkpoint(ckpt_path)
            self.assertIsNotNone(ckpt)

    @patch("manual_compaction.http_post")
    @patch("manual_compaction.http_get")
    @patch("manual_compaction.pd_request")
    def test_checkpoint_resume(self, mock_pd, mock_get, mock_post):
        region2 = make_region(2, "4200", "4300", 1)
        mock_pd.side_effect = [
            {"count": 1, "regions": [region2]},
            make_store_response(1, "127.0.0.1:20292"),
            {"count": 0, "regions": []},
        ]
        mock_get.return_value = {"active": True, "entries": 10, "old-entries": 90}
        mock_post.return_value = (200, "ok")

        with tempfile.TemporaryDirectory() as tmp:
            ckpt_path = os.path.join(tmp, "ckpt.json")
            resume_key = bytes.fromhex("4200")
            ckpt_data = mc.build_checkpoint(
                resume_key, 1, "4100", "4200",
                {"triggered": 1, "regions_seen": 1, "stats_failures": 0,
                 "leader_store_failures": 0, "inactive": 0, "zero_entries": 0,
                 "below_threshold": 0, "dry_run_would_trigger": 0,
                 "trigger_failures": 0, "checkpoint_writes": 1, "checkpoint_loads": 0}
            )
            mc.write_checkpoint(ckpt_path, ckpt_data)

            args = self._args(
                start_key="41",
                end_key="",
                checkpoint_file=ckpt_path,
                ignore_checkpoint=False,
            )
            rc = mc.run_compaction(args, ["http://pd:2379"], None)
            self.assertEqual(rc, 0)
            pd_calls = mock_pd.call_args_list
            first_regions_call = next(
                c for c in pd_calls if mc.PD_REGIONS_KEY_PATH in str(c)
            )
            call_params = first_regions_call[1].get("params", {}) if first_regions_call[1] else {}
            if not call_params and len(first_regions_call[0]) > 4:
                call_params = first_regions_call[0][4]
            key_param = call_params.get("key", "")
            self.assertEqual(key_param, resume_key.hex())
            self.assertEqual(call_params.get("format"), "hex")

    @patch("manual_compaction.http_post")
    @patch("manual_compaction.http_get")
    @patch("manual_compaction.pd_request")
    def test_dry_run_does_not_write_checkpoint(self, mock_pd, mock_get, mock_post):
        region = make_region(1, "4100", "4200", 1)
        mock_pd.side_effect = [
            {"count": 1, "regions": [region]},
            make_store_response(1, "127.0.0.1:20292"),
            {"count": 0, "regions": []},
        ]
        mock_get.return_value = {"active": True, "entries": 10, "old-entries": 90}
        mock_post.return_value = (200, "ok")

        with tempfile.TemporaryDirectory() as tmp:
            ckpt_path = os.path.join(tmp, "ckpt.json")
            args = self._args(dry_run=True, checkpoint_file=ckpt_path, region_limit=1)
            mc.run_compaction(args, ["http://pd:2379"], None)
            self.assertFalse(os.path.exists(ckpt_path))

    @patch("manual_compaction.http_post")
    @patch("manual_compaction.http_get")
    @patch("manual_compaction.pd_request")
    def test_end_of_keyspace_empty_end_key(self, mock_pd, mock_get, mock_post):
        region = make_region(1, "4100", "", 1)
        mock_pd.side_effect = [
            {"count": 1, "regions": [region]},
            make_store_response(1, "127.0.0.1:20292"),
        ]
        mock_get.return_value = {"active": False, "entries": 0, "old-entries": 0}

        args = self._args()
        rc = mc.run_compaction(args, ["http://pd:2379"], None)
        self.assertEqual(rc, 0)
        pd_calls = [c for c in mock_pd.call_args_list if mc.PD_REGIONS_KEY_PATH in str(c)]
        self.assertEqual(len(pd_calls), 1)

    @patch("manual_compaction.http_post")
    @patch("manual_compaction.http_get")
    @patch("manual_compaction.pd_request")
    def test_keyboard_interrupt_writes_checkpoint(self, mock_pd, mock_get, mock_post):
        region = make_region(1, "4100", "4200", 1)
        mock_pd.side_effect = [
            {"count": 1, "regions": [region]},
            make_store_response(1, "127.0.0.1:20292"),
        ]
        mock_get.side_effect = KeyboardInterrupt()

        with tempfile.TemporaryDirectory() as tmp:
            ckpt_path = os.path.join(tmp, "ckpt.json")
            args = self._args(checkpoint_file=ckpt_path, checkpoint_interval=0)
            rc = mc.run_compaction(args, ["http://pd:2379"], None)
            self.assertEqual(rc, 130)
            ckpt = mc.load_checkpoint(ckpt_path)
            self.assertIsNotNone(ckpt)
            self.assertTrue(ckpt.get("interrupted"))

    @patch("manual_compaction.http_get")
    @patch("manual_compaction.pd_request")
    def test_keyboard_interrupt_dry_run_no_checkpoint(self, mock_pd, mock_get):
        region = make_region(1, "4100", "4200", 1)
        mock_pd.side_effect = [
            {"count": 1, "regions": [region]},
            make_store_response(1, "127.0.0.1:20292"),
        ]
        mock_get.side_effect = KeyboardInterrupt()

        with tempfile.TemporaryDirectory() as tmp:
            ckpt_path = os.path.join(tmp, "ckpt.json")
            args = self._args(dry_run=True, checkpoint_file=ckpt_path)
            rc = mc.run_compaction(args, ["http://pd:2379"], None)
            self.assertEqual(rc, 130)
            self.assertFalse(os.path.exists(ckpt_path))

    @patch("manual_compaction.http_post")
    @patch("manual_compaction.http_get")
    @patch("manual_compaction.pd_request")
    def test_checkpoint_interval_suppresses_early_write(self, mock_pd, mock_get, mock_post):
        region1 = make_region(1, "4100", "4200", 1)
        region2 = make_region(2, "4200", "4300", 1)
        mock_pd.side_effect = [
            {"count": 1, "regions": [region1]},
            make_store_response(1, "127.0.0.1:20292"),
            {"count": 1, "regions": [region2]},
            {"count": 0, "regions": []},
        ]
        mock_get.return_value = {"active": False, "entries": 0, "old-entries": 0}
        mock_post.return_value = (200, "ok")

        with tempfile.TemporaryDirectory() as tmp:
            ckpt_path = os.path.join(tmp, "ckpt.json")
            # Large interval: no mid-run checkpoint should be written
            args = self._args(checkpoint_file=ckpt_path, region_limit=1, checkpoint_interval=9999)
            with patch("manual_compaction.time.monotonic", side_effect=[0.0, 1.0, 2.0, 3.0, 4.0, 5.0]):
                mc.run_compaction(args, ["http://pd:2379"], None)
            # Only the final completion checkpoint is written
            ckpt = mc.load_checkpoint(ckpt_path)
            self.assertIsNotNone(ckpt)
            self.assertEqual(ckpt.get("counters", {}).get("checkpoint_writes", 0), 0)

    @patch("manual_compaction.http_post")
    @patch("manual_compaction.http_get")
    @patch("manual_compaction.pd_request")
    def test_checkpoint_interval_zero_writes_every_batch(self, mock_pd, mock_get, mock_post):
        region1 = make_region(1, "4100", "4200", 1)
        region2 = make_region(2, "4200", "4300", 1)
        mock_pd.side_effect = [
            {"count": 1, "regions": [region1]},
            make_store_response(1, "127.0.0.1:20292"),
            {"count": 1, "regions": [region2]},
            {"count": 0, "regions": []},
        ]
        mock_get.return_value = {"active": False, "entries": 0, "old-entries": 0}
        mock_post.return_value = (200, "ok")

        with tempfile.TemporaryDirectory() as tmp:
            ckpt_path = os.path.join(tmp, "ckpt.json")
            args = self._args(checkpoint_file=ckpt_path, region_limit=1, checkpoint_interval=0)
            mc.run_compaction(args, ["http://pd:2379"], None)
            ckpt = mc.load_checkpoint(ckpt_path)
            self.assertIsNotNone(ckpt)
            self.assertGreaterEqual(ckpt.get("counters", {}).get("checkpoint_writes", 0), 2)

    @patch("manual_compaction.http_post")
    @patch("manual_compaction.http_get")
    @patch("manual_compaction.pd_request")
    def test_fewer_regions_than_limit_stops(self, mock_pd, mock_get, mock_post):
        regions = [make_region(i, f"{i:04x}", f"{i+1:04x}", 1) for i in range(1, 4)]
        mock_pd.side_effect = [
            {"count": 3, "regions": regions},
            make_store_response(1, "127.0.0.1:20292"),
        ]
        mock_get.return_value = {"active": False, "entries": 0, "old-entries": 0}

        args = self._args(region_limit=128)
        rc = mc.run_compaction(args, ["http://pd:2379"], None)
        self.assertEqual(rc, 0)


class TestScanInterval(unittest.TestCase):
    def _args(self, **kwargs):
        defaults = dict(
            pd="http://pd:2379",
            start_key="",
            end_key="",
            region_limit=128,
            old_ratio=0.3,
            interval=0,
            scan_interval=0,
            timeout=10,
            dry_run=False,
            checkpoint_file="",
            ignore_checkpoint=False,
            checkpoint_interval=0,
            ca_cert="",
            client_cert="",
            client_key="",
        )
        defaults.update(kwargs)
        return MagicMock(**defaults)

    @patch("manual_compaction.http_post")
    @patch("manual_compaction.http_get")
    @patch("manual_compaction.pd_request")
    def test_scan_interval_sleep_between_batches(self, mock_pd, mock_get, mock_post):
        region1 = make_region(1, "4100", "4200", 1)
        region2 = make_region(2, "4200", "4300", 1)
        mock_pd.side_effect = [
            {"count": 1, "regions": [region1]},
            make_store_response(1, "127.0.0.1:20292"),
            {"count": 1, "regions": [region2]},
            {"count": 0, "regions": []},
        ]
        mock_get.return_value = {"active": False, "entries": 0, "old-entries": 0}
        mock_post.return_value = (200, "ok")

        with patch("manual_compaction.time") as mock_time:
            mock_time.monotonic.return_value = 0.0
            mock_time.sleep = MagicMock()
            args = self._args(region_limit=1, scan_interval=1.0)
            mc.run_compaction(args, ["http://pd:2379"], None)

        sleep_calls = [c for c in mock_time.sleep.call_args_list if c[0][0] == 1.0]
        self.assertEqual(len(sleep_calls), 3)

    @patch("manual_compaction.pd_request")
    def test_scan_interval_zero_no_sleep(self, mock_pd):
        mock_pd.return_value = {"count": 0, "regions": []}
        with patch("manual_compaction.time") as mock_time:
            mock_time.monotonic.return_value = 0.0
            mock_time.sleep = MagicMock()
            args = self._args(scan_interval=0)
            mc.run_compaction(args, ["http://pd:2379"], None)
        mock_time.sleep.assert_not_called()

    @patch("manual_compaction.pd_request")
    def test_scan_interval_default_is_one_second(self, mock_pd):
        mock_pd.return_value = {"count": 0, "regions": []}
        args = mc.parse_args([])
        self.assertAlmostEqual(args.scan_interval, mc.DEFAULT_SCAN_INTERVAL)
        self.assertAlmostEqual(mc.DEFAULT_SCAN_INTERVAL, 0.5)


    def test_defaults(self):
        args = mc.parse_args([])
        self.assertEqual(args.start_key, "")
        self.assertEqual(args.end_key, "")
        self.assertEqual(args.region_limit, mc.DEFAULT_REGION_LIMIT)
        self.assertAlmostEqual(args.old_ratio, mc.DEFAULT_OLD_RATIO)
        self.assertFalse(args.dry_run)
        self.assertEqual(args.checkpoint_file, "")
        self.assertAlmostEqual(args.checkpoint_interval, mc.DEFAULT_CHECKPOINT_INTERVAL)

    def test_key_range(self):
        args = mc.parse_args(["--start-key", "41", "--end-key", "5a"])
        self.assertEqual(args.start_key, "41")
        self.assertEqual(args.end_key, "5a")

    def test_dry_run(self):
        args = mc.parse_args(["--dry-run"])
        self.assertTrue(args.dry_run)

    def test_checkpoint(self):
        args = mc.parse_args(["--checkpoint-file", "/tmp/ckpt.json"])
        self.assertEqual(args.checkpoint_file, "/tmp/ckpt.json")

    def test_region_limit(self):
        args = mc.parse_args(["--region-limit", "64"])
        self.assertEqual(args.region_limit, 64)


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