# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved.
# OpenOLC is licensed under Mulan PSL v2.
# You can use this software according to the terms and conditions of the Mulan PSL v2.
# You may obtain a copy of Mulan PSL v2 at:
#         `http://license.coscl.org.cn/MulanPSL2`
# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
# See the Mulan PSL v2 for more details.

import unittest
import time
from unittest.mock import Mock, patch, MagicMock

from olc.collector.local_resource.local_resource_collector import (
    LocalResourceMeasurable,
    LocalResourceMeter,
    LocalResourceCollector,
)


class TestLocalResourceMeasurable(unittest.TestCase):
    """LocalResourceMeasurable 测试类"""

    def test_init(self):
        """测试初始化"""
        measurable = LocalResourceMeasurable(0.5, 0.6, 0.1, 0.2, 0.3)
        self.assertEqual(measurable.cpu, 0.5)
        self.assertEqual(measurable.memory, 0.6)
        self.assertEqual(measurable.io, 0.1)
        self.assertEqual(measurable.network, 0.2)
        self.assertEqual(measurable.loadaverage, 0.3)

    def test_get_properties(self):
        """测试获取属性列表"""
        properties = LocalResourceMeasurable.get_properties()
        self.assertEqual(properties, ["cpu", "memory", "io", "network", "loadaverage"])

    def test_str_representation(self):
        """测试字符串表示"""
        measurable = LocalResourceMeasurable(0.5, 0.6, 0.1, 0.2, 0.3)
        str_repr = str(measurable)
        self.assertIn("cpu=0.5", str_repr)
        self.assertIn("memory=0.6", str_repr)
        self.assertIn("io=0.1", str_repr)
        self.assertIn("network=0.2", str_repr)
        self.assertIn("loadaverage=0.3", str_repr)

    def test_repr(self):
        """测试 repr"""
        measurable = LocalResourceMeasurable(0.5, 0.6, 0.1, 0.2, 0.3)
        self.assertEqual(repr(measurable), str(measurable))


class TestLocalResourceMeter(unittest.TestCase):
    """LocalResourceMeter 测试类"""

    def setUp(self):
        """每个测试前重置单例，避免状态污染"""
        LocalResourceMeter._instance = None
        self.meter = LocalResourceMeter()

    def tearDown(self):
        """每个测试后清理单例"""
        LocalResourceMeter._instance = None

    def test_singleton(self):
        """测试单例模式"""
        instance1 = LocalResourceMeter.get_instance()
        instance2 = LocalResourceMeter.get_instance()
        self.assertIs(instance1, instance2)

    @patch("olc.collector.local_resource.local_resource_collector.psutil")
    def test_get_cpu_usage(self, mock_psutil):
        """测试获取 CPU 使用率"""
        mock_psutil.cpu_percent.return_value = 50.0
        result = self.meter._get_cpu_usage()
        self.assertEqual(result, 0.5)
        mock_psutil.cpu_percent.assert_called_once_with(interval=0)

    @patch("olc.collector.local_resource.local_resource_collector.psutil")
    def test_get_memory_usage(self, mock_psutil):
        """测试获取内存使用率"""
        mock_memory = Mock()
        mock_memory.percent = 75.0
        mock_psutil.virtual_memory.return_value = mock_memory
        result = self.meter._get_memory_usage()
        self.assertEqual(result, 0.75)
        mock_psutil.virtual_memory.assert_called_once()

    @patch("olc.collector.local_resource.local_resource_collector.psutil")
    def test_get_io_usage_first_call(self, mock_psutil):
        """测试首次获取 IO 使用率（应返回 0 并记录快照）"""
        mock_counters = Mock()
        mock_counters.read_bytes = 1024 * 1024
        mock_counters.write_bytes = 512 * 1024
        mock_psutil.disk_io_counters.return_value = mock_counters

        result = self.meter._get_io_usage()
        self.assertEqual(result, 0.0)
        self.assertIsNotNone(self.meter._prev_disk_io)
        self.assertEqual(self.meter._prev_disk_io, (1024 * 1024, 512 * 1024))
        self.assertGreater(self.meter._prev_disk_time, 0)

    @patch("olc.collector.local_resource.local_resource_collector.psutil")
    def test_get_io_usage_subsequent_call(self, mock_psutil):
        """测试第二次获取 IO 使用率（应计算速率）"""
        mock_counters = Mock()
        # 首次调用：1MB 读，0.5MB 写
        mock_counters.read_bytes = 1024 * 1024
        mock_counters.write_bytes = 512 * 1024
        mock_psutil.disk_io_counters.return_value = mock_counters

        self.meter._get_io_usage()

        # 模拟 1 秒后，增加了 500MB 读 + 500MB 写 = 1000MB
        mock_counters.read_bytes = 1024 * 1024 + 500 * 1024 * 1024
        mock_counters.write_bytes = 512 * 1024 + 500 * 1024 * 1024

        # 直接修改时间，模拟过了 1 秒
        prev_time = self.meter._prev_disk_time
        self.meter._prev_disk_time = prev_time - 1.0

        result = self.meter._get_io_usage()
        # 1000MB / 1s = 1000MB/s，上限 500MB/s -> 1.0
        self.assertEqual(result, 1.0)

    @patch("olc.collector.local_resource.local_resource_collector.psutil")
    def test_get_io_usage_none_counters(self, mock_psutil):
        """测试无法获取 IO 计数器时返回 0"""
        mock_psutil.disk_io_counters.return_value = None
        result = self.meter._get_io_usage()
        self.assertEqual(result, 0.0)

    @patch("olc.collector.local_resource.local_resource_collector.psutil")
    def test_get_network_usage_first_call(self, mock_psutil):
        """测试首次获取网络使用率（应返回 0 并记录快照）"""
        mock_counters = Mock()
        mock_counters.bytes_sent = 1024 * 1024
        mock_counters.bytes_recv = 2048 * 1024
        mock_psutil.net_io_counters.return_value = mock_counters

        result = self.meter._get_network_usage()
        self.assertEqual(result, 0.0)
        self.assertIsNotNone(self.meter._prev_net_io)
        self.assertEqual(self.meter._prev_net_io, (1024 * 1024, 2048 * 1024))
        self.assertGreater(self.meter._prev_net_time, 0)

    @patch("olc.collector.local_resource.local_resource_collector.psutil")
    def test_get_network_usage_subsequent_call(self, mock_psutil):
        """测试第二次获取网络使用率（应计算速率）"""
        mock_counters = Mock()
        mock_counters.bytes_sent = 1024 * 1024
        mock_counters.bytes_recv = 1024 * 1024
        mock_psutil.net_io_counters.return_value = mock_counters

        self.meter._get_network_usage()

        # 模拟 1 秒后，增加了 1000MB 发送 + 1000MB 接收 = 2000MB
        mock_counters.bytes_sent = 1024 * 1024 + 1000 * 1024 * 1024
        mock_counters.bytes_recv = 1024 * 1024 + 1000 * 1024 * 1024

        # 修改时间，模拟过了 1 秒
        prev_time = self.meter._prev_net_time
        self.meter._prev_net_time = prev_time - 1.0

        result = self.meter._get_network_usage()
        # 2000MB / 1s = 2000MB/s，上限 1000MB/s -> 1.0
        self.assertEqual(result, 1.0)

    @patch("olc.collector.local_resource.local_resource_collector.psutil")
    def test_get_load_usage_linux(self, mock_psutil):
        """测试 Linux 环境下获取负载使用率"""
        mock_psutil.getloadavg.return_value = (2.0, 1.5, 1.0)
        mock_psutil.cpu_count.return_value = 4
        result = self.meter._get_load_usage()
        # load1 / cpu_count = 2.0 / 4 = 0.5
        self.assertEqual(result, 0.5)

    @patch("olc.collector.local_resource.local_resource_collector.psutil")
    def test_get_load_usage_high_load(self, mock_psutil):
        """测试高负载情况下截断到 1.0"""
        mock_psutil.getloadavg.return_value = (10.0, 5.0, 2.0)
        mock_psutil.cpu_count.return_value = 4
        result = self.meter._get_load_usage()
        # 10.0 / 4 = 2.5 -> min(2.5, 1.0) = 1.0
        self.assertEqual(result, 1.0)

    @patch("olc.collector.local_resource.local_resource_collector.psutil")
    def test_get_load_usage_windows_fallback(self, mock_psutil):
        """测试 Windows 环境下回退到 CPU 使用率"""
        mock_psutil.getloadavg.side_effect = AttributeError("getloadavg not available")
        mock_psutil.cpu_percent.return_value = 30.0
        result = self.meter._get_load_usage()
        self.assertEqual(result, 0.3)
        mock_psutil.cpu_percent.assert_called_once_with(interval=0)

    @patch("olc.collector.local_resource.local_resource_collector.psutil")
    def test_get_load_usage_oserror_fallback(self, mock_psutil):
        """测试 OSError 时回退到 CPU 使用率"""
        mock_psutil.getloadavg.side_effect = OSError("not supported")
        mock_psutil.cpu_percent.return_value = 45.0
        result = self.meter._get_load_usage()
        self.assertEqual(result, 0.45)

    @patch("olc.collector.local_resource.local_resource_collector.psutil")
    def test_get_load_usage_no_cpu_count(self, mock_psutil):
        """测试无法获取 CPU 核心数时直接使用负载值"""
        mock_psutil.getloadavg.return_value = (0.8, 0.5, 0.3)
        mock_psutil.cpu_count.return_value = None
        result = self.meter._get_load_usage()
        self.assertEqual(result, 0.8)

    @patch("olc.collector.local_resource.local_resource_collector.psutil")
    def test_measure_success(self, mock_psutil):
        """测试成功测量所有指标"""
        mock_psutil.cpu_percent.return_value = 50.0
        mock_memory = Mock()
        mock_memory.percent = 60.0
        mock_psutil.virtual_memory.return_value = mock_memory
        mock_psutil.disk_io_counters.return_value = None
        mock_psutil.net_io_counters.return_value = Mock(bytes_sent=0, bytes_recv=0)
        mock_psutil.getloadavg.return_value = (2.0, 1.5, 1.0)
        mock_psutil.cpu_count.return_value = 4

        result = self.meter.measure()
        self.assertIsNotNone(result)
        self.assertIsInstance(result, LocalResourceMeasurable)
        self.assertEqual(result.cpu, 0.5)
        self.assertEqual(result.memory, 0.6)
        self.assertEqual(result.io, 0.0)
        self.assertEqual(result.network, 0.0)
        self.assertEqual(result.loadaverage, 0.5)

    @patch("olc.collector.local_resource.local_resource_collector.psutil")
    def test_measure_partial_exception(self, mock_psutil):
        """测试部分指标异常时返回默认值 -1.0"""
        mock_psutil.cpu_percent.side_effect = Exception("cpu error")
        mock_psutil.virtual_memory.side_effect = Exception("memory error")
        mock_psutil.disk_io_counters.return_value = None
        mock_psutil.net_io_counters.return_value = Mock(bytes_sent=0, bytes_recv=0)
        mock_psutil.getloadavg.return_value = (2.0, 1.5, 1.0)
        mock_psutil.cpu_count.return_value = 4

        result = self.meter.measure()
        self.assertIsNotNone(result)
        self.assertEqual(result.cpu, -1.0)
        self.assertEqual(result.memory, -1.0)
        self.assertEqual(result.io, 0.0)
        self.assertEqual(result.network, 0.0)
        self.assertEqual(result.loadaverage, 0.5)


class TestLocalResourceCollector(unittest.TestCase):
    """LocalResourceCollector 测试类"""

    def setUp(self):
        self.mock_meter = MagicMock(spec=LocalResourceMeter)
        self.collector = LocalResourceCollector(meter=self.mock_meter)

    def test_get_name(self):
        """测试获取收集器名称"""
        self.assertEqual(self.collector.get_name(), "localresource")

    def test_get_storage(self):
        """测试获取存储"""
        storage = self.collector.get_storage()
        self.assertIsInstance(storage, list)
        self.assertEqual(len(storage), 0)

    def test_get_properties(self):
        """测试获取属性列表"""
        properties = self.collector.get_properties()
        self.assertEqual(properties, ["cpu", "memory", "io", "network", "loadaverage"])

    def test_do_measure(self):
        """测试执行测量并存储结果"""
        measurable = LocalResourceMeasurable(0.5, 0.6, 0.1, 0.2, 0.3)
        self.mock_meter.measure.return_value = measurable

        self.collector.do_measure()

        storage = self.collector.get_storage()
        self.assertEqual(len(storage), 1)
        self.assertEqual(storage[0].cpu, 0.5)
        self.assertEqual(storage[0].memory, 0.6)
        self.assertEqual(storage[0].io, 0.1)
        self.assertEqual(storage[0].network, 0.2)
        self.assertEqual(storage[0].loadaverage, 0.3)

    def test_do_measure_none_result(self):
        """测试测量返回 None 时不存储"""
        self.mock_meter.measure.return_value = None

        self.collector.do_measure()

        storage = self.collector.get_storage()
        self.assertEqual(len(storage), 0)

    def test_do_measure_exception(self):
        """测试测量异常时不存储"""
        self.mock_meter.measure.side_effect = Exception("measure error")

        self.collector.do_measure()

        storage = self.collector.get_storage()
        self.assertEqual(len(storage), 0)

    def test_storage_truncation(self):
        """测试存储截断"""
        self.collector._sample = 3

        for i in range(5):
            measurable = LocalResourceMeasurable(
                float(i) / 10,
                float(i) / 10,
                float(i) / 10,
                float(i) / 10,
                float(i) / 10,
            )
            self.collector._measurements.append(measurable)

        new_measurable = LocalResourceMeasurable(0.9, 0.9, 0.9, 0.9, 0.9)
        self.mock_meter.measure.return_value = new_measurable

        self.collector.do_measure()

        storage = self.collector.get_storage()
        self.assertEqual(len(storage), 6)

    def test_init_default_meter(self):
        """测试默认使用单例 Meter"""
        with patch.object(LocalResourceMeter, "get_instance") as mock_get_instance:
            mock_meter = Mock(spec=LocalResourceMeter)
            mock_get_instance.return_value = mock_meter

            collector = LocalResourceCollector()
            self.assertEqual(collector._meter, mock_meter)


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