diff --git a/tests/api/test_proxy_api.py b/tests/api/test_proxy_api.py index de672814..012a11e8 100644 --- a/tests/api/test_proxy_api.py +++ b/tests/api/test_proxy_api.py @@ -13,7 +13,9 @@ __author__ = 'JHao' import pytest +from unittest.mock import patch, MagicMock from helper.proxy import Proxy +from api.proxyApi import JsonResponse @pytest.fixture @@ -151,4 +153,39 @@ def test_count_empty(self, client, mocks): data = resp.get_json() assert data["count"] == 0 assert data["http_type"] == {} - assert data["source"] == {} \ No newline at end of file + assert data["source"] == {} + + +class TestRefresh: + + def test_refresh_returns_success(self, client): + resp = client.get("/refresh/") + assert resp.status_code == 200 + assert b"success" in resp.data + + +class TestJsonResponse: + + def test_force_type_with_dict(self, app): + """dict -> JSON Response""" + with app.app_context(): + resp = JsonResponse.force_type({"key": "val"}) + assert resp.content_type == "application/json" + + def test_force_type_with_list(self, app): + """list -> JSON Response""" + with app.app_context(): + resp = JsonResponse.force_type([1, 2, 3]) + assert resp.content_type == "application/json" + + +class TestRunFlask: + + @patch("api.proxyApi.platform") + @patch("api.proxyApi.app") + def test_runflask_windows_path(self, mock_app, mock_platform): + """Windows 下调用 app.run()""" + mock_platform.system.return_value = "Windows" + from api.proxyApi import runFlask + runFlask() + mock_app.run.assert_called_once() \ No newline at end of file diff --git a/tests/unit/test_check.py b/tests/unit/test_check.py new file mode 100644 index 00000000..eb1daf8f --- /dev/null +++ b/tests/unit/test_check.py @@ -0,0 +1,276 @@ +# -*- coding: utf-8 -*- +""" +------------------------------------------------- + File Name： test_check.py + Description : helper/check.py 单元测试 + Author : JHao + date： 2026/6/15 +------------------------------------------------- + Change Activity: + 2026/06/15: +------------------------------------------------- +""" +__author__ = 'JHao' + +import pytest +from unittest.mock import patch, MagicMock, PropertyMock +from datetime import datetime + +from helper.proxy import Proxy +from helper.check import DoValidator, _ThreadChecker + + +class TestDoValidator: + """DoValidator.validator 测试""" + + @patch("helper.check.ConfigHandler") + @patch("helper.check.ProxyValidator") + def test_validator_http_pass_https_pass(self, mock_pv_cls, mock_conf_cls): + """HTTP 通过 + HTTPS 通过 -> https=True, fail_count 不变""" + mock_pv = MagicMock() + mock_pv.http_validator = [MagicMock(return_value=True)] + mock_pv.https_validator = [MagicMock(return_value=True)] + mock_pv_cls.http_validator = mock_pv.http_validator + mock_pv_cls.https_validator = mock_pv.https_validator + + mock_conf = MagicMock() + mock_conf.proxyRegion = False + mock_conf_cls.return_value = mock_conf + + proxy = Proxy("1.2.3.4:8080", source="test") + proxy.fail_count = 0 + + # Patch DoValidator.conf at class level + with patch.object(DoValidator, "conf", mock_conf): + result = DoValidator.validator(proxy, "use") + + assert result.https is True + assert result.last_status is True + assert result.check_count == 1 + assert result.fail_count == 0 + + @patch("helper.check.ConfigHandler") + @patch("helper.check.ProxyValidator") + def test_validator_http_pass_https_fail(self, mock_pv_cls, mock_conf_cls): + """HTTP 通过 + HTTPS 失败 -> https=False""" + mock_pv_cls.http_validator = [MagicMock(return_value=True)] + mock_pv_cls.https_validator = [MagicMock(return_value=False)] + + mock_conf = MagicMock() + mock_conf.proxyRegion = False + + proxy = Proxy("1.2.3.4:8080", source="test") + + with patch.object(DoValidator, "conf", mock_conf): + result = DoValidator.validator(proxy, "use") + + assert result.https is False + assert result.last_status is True + + @patch("helper.check.ConfigHandler") + @patch("helper.check.ProxyValidator") + def test_validator_http_fail(self, mock_pv_cls, mock_conf_cls): + """HTTP 失败 -> fail_count += 1, last_status=False""" + mock_pv_cls.http_validator = [MagicMock(return_value=False)] + + mock_conf = MagicMock() + mock_conf.proxyRegion = False + + proxy = Proxy("1.2.3.4:8080", source="test") + proxy.fail_count = 0 + + with patch.object(DoValidator, "conf", mock_conf): + result = DoValidator.validator(proxy, "use") + + assert result.last_status is False + assert result.fail_count == 1 + + @patch("helper.check.ConfigHandler") + @patch("helper.check.ProxyValidator") + def test_validator_fail_count_decrement(self, mock_pv_cls, mock_conf_cls): + """HTTP 通过 + fail_count > 0 -> fail_count -= 1""" + mock_pv_cls.http_validator = [MagicMock(return_value=True)] + mock_pv_cls.https_validator = [MagicMock(return_value=True)] + + mock_conf = MagicMock() + mock_conf.proxyRegion = False + + proxy = Proxy("1.2.3.4:8080", source="test") + proxy.fail_count = 3 + + with patch.object(DoValidator, "conf", mock_conf): + result = DoValidator.validator(proxy, "use") + + assert result.fail_count == 2 + + @patch("helper.check.DoValidator.regionGetter", return_value="US") + @patch("helper.check.ConfigHandler") + @patch("helper.check.ProxyValidator") + def test_validator_raw_sets_region(self, mock_pv_cls, mock_conf_cls, mock_region): + """work_type='raw' + proxyRegion=True -> regionGetter 被调用""" + mock_pv_cls.http_validator = [MagicMock(return_value=True)] + mock_pv_cls.https_validator = [MagicMock(return_value=True)] + + mock_conf = MagicMock() + mock_conf.proxyRegion = True + + proxy = Proxy("1.2.3.4:8080", source="test") + + with patch.object(DoValidator, "conf", mock_conf): + result = DoValidator.validator(proxy, "raw") + + assert result.region == "US" + mock_region.assert_called_once_with(proxy) + + @patch("helper.check.DoValidator.regionGetter") + @patch("helper.check.ConfigHandler") + @patch("helper.check.ProxyValidator") + def test_validator_use_skips_region(self, mock_pv_cls, mock_conf_cls, mock_region): + """work_type='use' -> 不调用 regionGetter""" + mock_pv_cls.http_validator = [MagicMock(return_value=True)] + mock_pv_cls.https_validator = [MagicMock(return_value=True)] + + mock_conf = MagicMock() + mock_conf.proxyRegion = True + + proxy = Proxy("1.2.3.4:8080", source="test") + + with patch.object(DoValidator, "conf", mock_conf): + DoValidator.validator(proxy, "use") + + mock_region.assert_not_called() + + +class TestRegionGetter: + """DoValidator.regionGetter 测试""" + + @patch("helper.check.WebRequest") + def test_success_returns_country_code(self, mock_wr_cls): + """正常返回 -> country_code""" + mock_wr = MagicMock() + mock_wr.get.return_value.json = {"country_code": "CN"} + mock_wr_cls.return_value = mock_wr + + proxy = Proxy("1.2.3.4:8080") + result = DoValidator.regionGetter(proxy) + assert result == "CN" + + @patch("helper.check.WebRequest") + def test_exception_returns_error(self, mock_wr_cls): + """异常 -> 'error'""" + mock_wr = MagicMock() + mock_wr.get.side_effect = Exception("timeout") + mock_wr_cls.return_value = mock_wr + + proxy = Proxy("1.2.3.4:8080") + result = DoValidator.regionGetter(proxy) + assert result == "error" + + +def _make_checker(work_type, proxy_handler, conf=None): + """构造手动注入依赖的 _ThreadChecker（绕过 Thread.__init__）""" + checker = _ThreadChecker.__new__(_ThreadChecker) + # 手动初始化 Thread 所需的状态 + checker._initialized = True + checker._name = "test_thread" + checker._target = None + checker._args = () + checker._kwargs = {} + checker._daemonic = False + checker._ident = None + checker._tstate_lock = None + checker._started = MagicMock() + checker._is_stopped = False + checker._block = MagicMock() + checker._waiters = [] + checker._stderr = None + # 注入依赖 + checker.proxy_handler = proxy_handler + checker.log = MagicMock() + checker.work_type = work_type + checker.conf = conf or MagicMock() + return checker + + +class TestThreadCheckerIfRaw: + """_ThreadChecker.__ifRaw 测试""" + + def test_ifraw_new_proxy_gets_put(self): + """last_status=True, exists=False -> put""" + mock_ph = MagicMock() + mock_ph.exists.return_value = False + + proxy = Proxy("1.2.3.4:8080", source="test") + proxy.last_status = True + + checker = _make_checker("raw", mock_ph) + checker._ThreadChecker__ifRaw(proxy) + mock_ph.put.assert_called_once_with(proxy) + + def test_ifraw_existing_proxy_skipped(self): + """last_status=True, exists=True -> 不 put""" + mock_ph = MagicMock() + mock_ph.exists.return_value = True + + proxy = Proxy("1.2.3.4:8080", source="test") + proxy.last_status = True + + checker = _make_checker("raw", mock_ph) + checker._ThreadChecker__ifRaw(proxy) + mock_ph.put.assert_not_called() + + def test_ifraw_failed_proxy_not_put(self): + """last_status=False -> 不 put""" + mock_ph = MagicMock() + + proxy = Proxy("1.2.3.4:8080", source="test") + proxy.last_status = False + + checker = _make_checker("raw", mock_ph) + checker._ThreadChecker__ifRaw(proxy) + mock_ph.put.assert_not_called() + + +class TestThreadCheckerIfUse: + """_ThreadChecker.__ifUse 测试""" + + def test_ifuse_pass_gets_put(self): + """last_status=True -> put""" + mock_ph = MagicMock() + + proxy = Proxy("1.2.3.4:8080", source="test") + proxy.last_status = True + + checker = _make_checker("use", mock_ph) + checker._ThreadChecker__ifUse(proxy) + mock_ph.put.assert_called_once_with(proxy) + + def test_ifuse_fail_exceeds_max_deleted(self): + """fail_count > maxFailCount -> delete""" + mock_ph = MagicMock() + mock_conf = MagicMock() + mock_conf.maxFailCount = 3 + + proxy = Proxy("1.2.3.4:8080", source="test") + proxy.last_status = False + proxy.fail_count = 5 + + checker = _make_checker("use", mock_ph, mock_conf) + checker._ThreadChecker__ifUse(proxy) + mock_ph.delete.assert_called_once_with(proxy) + mock_ph.put.assert_not_called() + + def test_ifuse_fail_below_max_kept(self): + """fail_count <= maxFailCount -> put""" + mock_ph = MagicMock() + mock_conf = MagicMock() + mock_conf.maxFailCount = 3 + + proxy = Proxy("1.2.3.4:8080", source="test") + proxy.last_status = False + proxy.fail_count = 2 + + checker = _make_checker("use", mock_ph, mock_conf) + checker._ThreadChecker__ifUse(proxy) + mock_ph.put.assert_called_once_with(proxy) + mock_ph.delete.assert_not_called() diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py new file mode 100644 index 00000000..8c8fe662 --- /dev/null +++ b/tests/unit/test_cli.py @@ -0,0 +1,66 @@ +# -*- coding: utf-8 -*- +""" +------------------------------------------------- + File Name： test_cli.py + Description : proxyPool CLI 单元测试 + Author : JHao + date： 2026/6/15 +------------------------------------------------- + Change Activity: + 2026/06/15: +------------------------------------------------- +""" +__author__ = 'JHao' + +import pytest +from unittest.mock import patch, MagicMock +from click.testing import CliRunner + +from proxyPool import cli +from setting import VERSION + + +@pytest.fixture +def runner(): + return CliRunner() + + +class TestCli: + + def test_version_flag(self, runner): + """--version 显示版本号""" + result = runner.invoke(cli, ["--version"]) + assert result.exit_code == 0 + assert VERSION in result.output + + @patch("proxyPool.startScheduler") + def test_schedule_command(self, mock_scheduler, runner): + """schedule 命令调用 startScheduler""" + result = runner.invoke(cli, ["schedule"]) + assert result.exit_code == 0 + mock_scheduler.assert_called_once() + + @patch("proxyPool.startServer") + def test_server_command(self, mock_server, runner): + """server 命令调用 startServer""" + result = runner.invoke(cli, ["server"]) + assert result.exit_code == 0 + mock_server.assert_called_once() + + @patch("handler.configHandler.ConfigHandler") + @patch("helper.fetch._discover_fetchers") + def test_fetcher_command(self, mock_discover, mock_conf_cls, runner): + """fetcher 命令输出启用的代理源列表""" + mock_cls1 = MagicMock() + mock_cls1.name = "freeProxy01" + mock_cls2 = MagicMock() + mock_cls2.name = "freeProxy02" + mock_discover.return_value = [mock_cls1, mock_cls2] + mock_conf = MagicMock() + mock_conf.fetcherExclude = [] + mock_conf_cls.return_value = mock_conf + + result = runner.invoke(cli, ["fetcher"]) + assert result.exit_code == 0 + assert "freeProxy01" in result.output + assert "freeProxy02" in result.output \ No newline at end of file diff --git a/tests/unit/test_db_client.py b/tests/unit/test_db_client.py index 31becb5c..49ae429c 100644 --- a/tests/unit/test_db_client.py +++ b/tests/unit/test_db_client.py @@ -13,6 +13,8 @@ __author__ = 'JHao' import pytest +from unittest.mock import MagicMock, patch + from db.dbClient import DbClient @@ -59,4 +61,102 @@ def test_parse_returns_cls(self, uri, expected_type): """parseDbConn 返回 cls 以支持链式调用""" result = DbClient.parseDbConn(uri) assert result is DbClient - assert DbClient.db_type == expected_type \ No newline at end of file + assert DbClient.db_type == expected_type + + +class TestDbClientInit: + + @patch("db.dbClient.DbClient.parseDbConn") + def test_redis_init(self, mock_parse): + """Redis URI -> RedisClient 实例""" + with patch.object(DbClient, "_DbClient__initDbClient") as mock_init: + db = DbClient.__new__(DbClient) + DbClient.__init__(db, "redis://:pwd@127.0.0.1:6379/0") + mock_parse.assert_called_once_with("redis://:pwd@127.0.0.1:6379/0") + mock_init.assert_called_once() + + @patch("db.dbClient.DbClient.parseDbConn") + def test_ssdb_init(self, mock_parse): + """SSDB URI -> SsdbClient 实例""" + with patch.object(DbClient, "_DbClient__initDbClient") as mock_init: + db = DbClient.__new__(DbClient) + DbClient.__init__(db, "ssdb://:pwd@127.0.0.1:8888") + mock_parse.assert_called_once_with("ssdb://:pwd@127.0.0.1:8888") + mock_init.assert_called_once() + + +class TestDbClientDelegation: + """所有委托方法测试""" + + def _make_client(self): + """构造注入 mock client 的 DbClient""" + db = DbClient.__new__(DbClient) + db.client = MagicMock() + return db + + def test_get(self): + db = self._make_client() + db.client.get.return_value = '{"proxy": "1.2.3.4:8080"}' + result = db.get(True) + db.client.get.assert_called_once_with(True) + assert result == '{"proxy": "1.2.3.4:8080"}' + + def test_put(self): + db = self._make_client() + db.put("1.2.3.4:8080") + db.client.put.assert_called_once_with("1.2.3.4:8080") + + def test_update(self): + db = self._make_client() + db.update("key", "value") + db.client.update.assert_called_once_with("key", "value") + + def test_delete(self): + db = self._make_client() + db.delete("1.2.3.4:8080") + db.client.delete.assert_called_once_with("1.2.3.4:8080") + + def test_exists(self): + db = self._make_client() + db.client.exists.return_value = True + result = db.exists("1.2.3.4:8080") + db.client.exists.assert_called_once_with("1.2.3.4:8080") + assert result is True + + def test_pop(self): + db = self._make_client() + db.client.pop.return_value = '{"proxy": "1.2.3.4:8080"}' + result = db.pop(True) + db.client.pop.assert_called_once_with(True) + assert result == '{"proxy": "1.2.3.4:8080"}' + + def test_getAll(self): + db = self._make_client() + db.client.getAll.return_value = [] + result = db.getAll(False) + db.client.getAll.assert_called_once_with(False) + assert result == [] + + def test_clear(self): + db = self._make_client() + db.clear() + db.client.clear.assert_called_once() + + def test_changeTable(self): + db = self._make_client() + db.changeTable("use_proxy") + db.client.changeTable.assert_called_once_with("use_proxy") + + def test_getCount(self): + db = self._make_client() + db.client.getCount.return_value = 42 + result = db.getCount() + db.client.getCount.assert_called_once() + assert result == 42 + + def test_test(self): + db = self._make_client() + db.client.test.return_value = True + result = db.test() + db.client.test.assert_called_once() + assert result is True \ No newline at end of file diff --git a/tests/unit/test_fetch.py b/tests/unit/test_fetch.py new file mode 100644 index 00000000..73e3b29e --- /dev/null +++ b/tests/unit/test_fetch.py @@ -0,0 +1,144 @@ +# -*- coding: utf-8 -*- +""" +------------------------------------------------- + File Name： test_fetch.py + Description : helper/fetch.py 单元测试 + Author : JHao + date： 2026/6/15 +------------------------------------------------- + Change Activity: + 2026/06/15: +------------------------------------------------- +""" +__author__ = 'JHao' + +import os +import sys +import pytest +from unittest.mock import patch, MagicMock + +import helper.fetch as fetch_mod +from helper.fetch import _get_sources_dir, _load_module, _discover_fetchers, _ThreadFetcher +from helper.proxy import Proxy +from fetcher.baseFetcher import BaseFetcher + + +class TestGetSourcesDir: + + def test_returns_correct_path(self): + """返回 fetcher/sources/ 目录路径""" + path = _get_sources_dir() + assert path.endswith(os.path.join("fetcher", "sources")) + assert os.path.isdir(path) + + +class TestLoadModule: + + def setup_method(self): + """每个测试前清空缓存""" + fetch_mod._module_cache.clear() + + def test_fresh_load(self): + """缓存为空 -> importlib.import_module""" + # 使用一个已知存在的模块 + filepath = os.path.join(_get_sources_dir(), "kuaidaili.py") + result = _load_module("fetcher.sources.kuaidaili", filepath) + assert result is not None + assert "fetcher.sources.kuaidaili" in fetch_mod._module_cache + + def test_cache_hit(self): + """mtime 不变 -> 返回缓存""" + filepath = os.path.join(_get_sources_dir(), "kuaidaili.py") + first = _load_module("fetcher.sources.kuaidaili", filepath) + second = _load_module("fetcher.sources.kuaidaili", filepath) + assert first is second + + def test_cache_miss_reload(self): + """mtime 变化 -> importlib.reload""" + filepath = os.path.join(_get_sources_dir(), "kuaidaili.py") + first = _load_module("fetcher.sources.kuaidaili", filepath) + # 模拟 mtime 变化 + fetch_mod._module_cache["fetcher.sources.kuaidaili"] = (0, first) + second = _load_module("fetcher.sources.kuaidaili", filepath) + assert second is not None + + @patch("helper.fetch.os.path.getmtime", return_value=0) + @patch("helper.fetch.importlib") + def test_import_exception_returns_none(self, mock_importlib, mock_mtime): + """import 失败 -> 返回 None""" + mock_importlib.import_module.side_effect = ImportError("not found") + # 确保模块不在 sys.modules 中，避免走 reload 分支 + mock_importlib.reload.side_effect = ImportError("not found") + saved = sys.modules.pop("fetcher.sources.nonexistent", None) + try: + result = _load_module("fetcher.sources.nonexistent", "/fake/path.py") + assert result is None + finally: + if saved is not None: + sys.modules["fetcher.sources.nonexistent"] = saved + + +class TestDiscoverFetchers: + + def setup_method(self): + fetch_mod._module_cache.clear() + + def test_filters_enabled_only(self): + """enabled=False 的 fetcher 被排除""" + # 使用真实扫描，检查结果中所有 fetcher 都是 enabled=True + fetchers = _discover_fetchers([]) + for f in fetchers: + assert f.enabled is True + + def test_filters_exclude_list(self): + """exclude_list 中的被排除""" + all_fetchers = _discover_fetchers([]) + if not all_fetchers: + pytest.skip("No fetchers available") + first_name = all_fetchers[0].__name__ + filtered = _discover_fetchers([first_name]) + filtered_names = [f.__name__ for f in filtered] + assert first_name not in filtered_names + + def test_returns_sorted_by_name(self): + """返回结果按 name 排序""" + fetchers = _discover_fetchers([]) + names = [f.name for f in fetchers] + assert names == sorted(names) + + def test_prunes_stale_cache(self): + """已删除文件的缓存被清理""" + fetch_mod._module_cache["fetcher.sources.deleted_module"] = (0, MagicMock()) + _discover_fetchers([]) + assert "fetcher.sources.deleted_module" not in fetch_mod._module_cache + + +class TestThreadFetcher: + + def test_collects_proxies(self): + """fetcher.fetch() yield 代理 -> proxy_dict 有值""" + mock_cls = MagicMock() + mock_cls.name = "test_fetcher" + mock_cls.return_value.fetch.return_value = ["1.2.3.4:8080", "5.6.7.8:443"] + + proxy_dict = {} + thread = _ThreadFetcher(mock_cls, proxy_dict) + thread.run() + + assert "1.2.3.4:8080" in proxy_dict + assert "5.6.7.8:443" in proxy_dict + assert isinstance(proxy_dict["1.2.3.4:8080"], Proxy) + + def test_merges_duplicate_sources(self): + """同一代理出现两次 -> add_source 被调用""" + mock_cls = MagicMock() + mock_cls.name = "test_fetcher" + mock_cls.return_value.fetch.return_value = ["1.2.3.4:8080", "1.2.3.4:8080"] + + proxy_dict = {} + thread = _ThreadFetcher(mock_cls, proxy_dict) + thread.run() + + assert "1.2.3.4:8080" in proxy_dict + # source 应该包含两次 "test_fetcher"（add_source 去重，但只出现一次） + assert "test_fetcher" in proxy_dict["1.2.3.4:8080"].source diff --git a/tests/unit/test_launcher.py b/tests/unit/test_launcher.py new file mode 100644 index 00000000..20ceb9ea --- /dev/null +++ b/tests/unit/test_launcher.py @@ -0,0 +1,79 @@ +# -*- coding: utf-8 -*- +""" +------------------------------------------------- + File Name： test_launcher.py + Description : helper/launcher.py 单元测试 + Author : JHao + date： 2026/6/15 +------------------------------------------------- + Change Activity: + 2026/06/15: +------------------------------------------------- +""" +__author__ = 'JHao' + +import pytest +from unittest.mock import patch, MagicMock + +import helper.launcher as launcher_mod + + +class TestStartServer: + + def test_calls_before_start_then_flask(self): + """startServer 先调用 __beforeStart 再调用 runFlask""" + with patch.object(launcher_mod, "__beforeStart") as mock_before, \ + patch("api.proxyApi.runFlask") as mock_flask: + launcher_mod.startServer() + mock_before.assert_called_once() + + +class TestStartScheduler: + + def test_calls_before_start_then_scheduler(self): + """startScheduler 先调用 __beforeStart 再调用 runScheduler""" + with patch.object(launcher_mod, "__beforeStart") as mock_before, \ + patch("helper.scheduler.runScheduler") as mock_sched: + launcher_mod.startScheduler() + mock_before.assert_called_once() + + +class TestBeforeStart: + + def test_exits_when_db_check_fails(self): + """DB 检查失败 -> sys.exit()""" + with patch.object(launcher_mod, "__showVersion"), \ + patch.object(launcher_mod, "__showConfigure"), \ + patch.object(launcher_mod, "__checkDBConfig", return_value=True), \ + patch("helper.launcher.sys") as mock_sys: + getattr(launcher_mod, "__beforeStart")() + mock_sys.exit.assert_called_once() + + def test_continues_when_db_check_passes(self): + """DB 检查通过 -> 不调用 sys.exit""" + with patch.object(launcher_mod, "__showVersion"), \ + patch.object(launcher_mod, "__showConfigure"), \ + patch.object(launcher_mod, "__checkDBConfig", return_value=False), \ + patch("helper.launcher.sys") as mock_sys: + getattr(launcher_mod, "__beforeStart")() + mock_sys.exit.assert_not_called() + + +class TestCheckDBConfig: + + def test_returns_db_test_result(self): + """返回 db.test() 的结果""" + with patch.object(launcher_mod, "DbClient") as mock_db_cls, \ + patch.object(launcher_mod, "ConfigHandler") as mock_conf_cls: + mock_conf = MagicMock() + mock_conf.dbConn = "redis://:@127.0.0.1:6379/0" + mock_conf_cls.return_value = mock_conf + + mock_db = MagicMock() + mock_db.test.return_value = False + mock_db_cls.return_value = mock_db + + result = getattr(launcher_mod, "__checkDBConfig")() + + assert result is False + mock_db.test.assert_called_once() \ No newline at end of file diff --git a/tests/unit/test_log_handler.py b/tests/unit/test_log_handler.py new file mode 100644 index 00000000..10f09946 --- /dev/null +++ b/tests/unit/test_log_handler.py @@ -0,0 +1,115 @@ +# -*- coding: utf-8 -*- +""" +------------------------------------------------- + File Name： test_log_handler.py + Description : LogHandler 单元测试 + Author : JHao + date： 2026/6/15 +------------------------------------------------- + Change Activity: + 2026/06/15: +------------------------------------------------- +""" +__author__ = 'JHao' + +import logging +import pytest +from unittest.mock import patch, MagicMock + +from handler.logHandler import LogHandler, DEBUG, INFO, ERROR + + +class TestLogHandlerInit: + """__init__ 测试""" + + def test_default_creates_stream_handler(self): + """默认参数创建 stream handler""" + log = LogHandler("test_default_stream", stream=True, file=False) + handler_types = [type(h) for h in log.handlers] + assert logging.StreamHandler in handler_types + + @patch("handler.logHandler.platform") + def test_file_handler_on_linux(self, mock_platform): + """Linux 下创建 file handler""" + mock_platform.system.return_value = "Linux" + log = LogHandler("test_linux_file", stream=False, file=True) + has_file = any(isinstance(h, logging.handlers.TimedRotatingFileHandler) for h in log.handlers) + assert has_file + + @patch("handler.logHandler.platform") + def test_no_file_handler_on_windows(self, mock_platform): + """Windows 下不创建 file handler""" + mock_platform.system.return_value = "Windows" + log = LogHandler("test_windows_no_file", stream=False, file=True) + has_file = any(isinstance(h, logging.handlers.TimedRotatingFileHandler) for h in log.handlers) + assert not has_file + + def test_no_stream_handler_when_disabled(self): + """stream=False 时不创建 stream handler""" + log = LogHandler("test_no_stream", stream=False, file=False) + assert len(log.handlers) == 0 + + +class TestLogHandlerStreamLevel: + """stream handler level 测试""" + + def test_default_level_used_when_no_override(self): + """未指定 level 时使用 self.level""" + log = LogHandler("test_stream_level", level=ERROR, stream=True, file=False) + stream_handlers = [h for h in log.handlers if isinstance(h, logging.StreamHandler) + and not isinstance(h, logging.handlers.TimedRotatingFileHandler)] + assert len(stream_handlers) > 0 + assert stream_handlers[0].level == ERROR + + def test_explicit_level_overrides_default(self): + """显式指定 level 覆盖默认值""" + log = LogHandler("test_stream_override", level=DEBUG, stream=True, file=False) + log.__setStreamHandler__(level=ERROR) + # 最后添加的 handler 应该是 ERROR 级别 + last_handler = log.handlers[-1] + assert last_handler.level == ERROR + + +class TestLogHandlerFileLevel: + """file handler level 测试""" + + @patch("handler.logHandler.platform") + def test_file_handler_default_level(self, mock_platform): + """file handler 未指定 level 时使用 self.level""" + mock_platform.system.return_value = "Linux" + log = LogHandler("test_file_level", level=INFO, stream=False, file=True) + file_handlers = [h for h in log.handlers + if isinstance(h, logging.handlers.TimedRotatingFileHandler)] + assert len(file_handlers) > 0 + assert file_handlers[0].level == INFO + + @patch("handler.logHandler.platform") + def test_file_handler_explicit_level(self, mock_platform): + """file handler 显式指定 level""" + mock_platform.system.return_value = "Linux" + log = LogHandler("test_file_override", level=DEBUG, stream=False, file=True) + log.__setFileHandler__(level=ERROR) + last_handler = log.handlers[-1] + assert last_handler.level == ERROR + + +class TestLogHandlerDirCreation: + """log 目录创建测试""" + + @patch("os.path.exists", return_value=False) + @patch("os.mkdir") + def test_creates_log_dir_when_missing(self, mock_mkdir, mock_exists): + """log 目录不存在时创建""" + # 重新 import 触发模块级代码（无法直接测试，验证模块级逻辑） + # 这里测试的是 FileExistsError 处理 + import handler.logHandler as lh + # 模块加载时已执行，此处验证 LOG_PATH 存在 + assert lh.LOG_PATH is not None + + @patch("os.path.exists", return_value=False) + @patch("os.mkdir", side_effect=FileExistsError) + def test_handles_file_exists_race_condition(self, mock_mkdir, mock_exists): + """处理 mkdir 时的 FileExistsError 竞态条件""" + # 验证模块级代码不会因 FileExistsError 崩溃 + import handler.logHandler as lh + assert lh.LOG_PATH is not None \ No newline at end of file diff --git a/tests/unit/test_proxy_handler.py b/tests/unit/test_proxy_handler.py new file mode 100644 index 00000000..441b89a7 --- /dev/null +++ b/tests/unit/test_proxy_handler.py @@ -0,0 +1,178 @@ +# -*- coding: utf-8 -*- +""" +------------------------------------------------- + File Name： test_proxy_handler.py + Description : ProxyHandler 单元测试 + Author : JHao + date： 2026/6/15 +------------------------------------------------- + Change Activity: + 2026/06/15: +------------------------------------------------- +""" +__author__ = 'JHao' + +import pytest +from unittest.mock import MagicMock, patch + +from handler.proxyHandler import ProxyHandler +from helper.proxy import Proxy + + +def _make_handler(): + """构造注入 mock DbClient 的 ProxyHandler 实例""" + with patch("handler.proxyHandler.DbClient") as mock_db_cls, \ + patch("handler.proxyHandler.ConfigHandler") as mock_conf_cls: + mock_db = MagicMock() + mock_db_cls.return_value = mock_db + mock_conf = MagicMock() + mock_conf.dbConn = "redis://:test@127.0.0.1:6379/0" + mock_conf.tableName = "use_proxy" + mock_conf_cls.return_value = mock_conf + handler = ProxyHandler() + handler._mock_db = mock_db + return handler + + +class TestProxyHandlerGet: + """get() 测试""" + + def test_get_returns_proxy(self): + """DbClient 返回 JSON -> Proxy 对象""" + handler = _make_handler() + proxy = Proxy("1.2.3.4:8080", source="test", https=False) + handler._mock_db.get.return_value = proxy.to_json + + result = handler.get(https=False) + + assert result is not None + assert result.proxy == "1.2.3.4:8080" + assert result.https is False + handler._mock_db.get.assert_called_once_with(False) + + def test_get_returns_none_when_empty(self): + """DbClient 返回 None -> None""" + handler = _make_handler() + handler._mock_db.get.return_value = None + + result = handler.get(https=False) + + assert result is None + + def test_get_https_forwarded(self): + """https=True 转发给 DbClient""" + handler = _make_handler() + proxy = Proxy("5.6.7.8:443", source="test", https=True) + handler._mock_db.get.return_value = proxy.to_json + + result = handler.get(https=True) + + assert result is not None + assert result.https is True + handler._mock_db.get.assert_called_once_with(True) + + +class TestProxyHandlerPop: + """pop() 测试""" + + def test_pop_returns_proxy(self): + """pop 正常返回 Proxy 对象""" + handler = _make_handler() + proxy = Proxy("1.2.3.4:8080", source="test") + handler._mock_db.pop.return_value = proxy.to_json + + result = handler.pop(https=False) + + assert result is not None + assert result.proxy == "1.2.3.4:8080" + handler._mock_db.pop.assert_called_once_with(False) + + def test_pop_returns_none_when_empty(self): + """pop 无数据时返回 None""" + handler = _make_handler() + handler._mock_db.pop.return_value = None + + result = handler.pop(https=False) + + assert result is None + + +class TestProxyHandlerPut: + """put() 测试""" + + def test_put_delegates_to_db(self): + """put 调用 DbClient.put""" + handler = _make_handler() + proxy = Proxy("1.2.3.4:8080", source="test") + + handler.put(proxy) + + handler._mock_db.put.assert_called_once_with(proxy) + + +class TestProxyHandlerDelete: + """delete() 测试""" + + def test_delete_delegates_to_db(self): + """delete 传入 proxy.proxy 字符串给 DbClient""" + handler = _make_handler() + proxy = Proxy("1.2.3.4:8080", source="test") + + handler.delete(proxy) + + handler._mock_db.delete.assert_called_once_with("1.2.3.4:8080") + + +class TestProxyHandlerGetAll: + """getAll() 测试""" + + def test_getAll_returns_proxy_list(self): + """getAll 返回 Proxy 对象列表""" + handler = _make_handler() + proxy1 = Proxy("1.2.3.4:8080", source="test").to_json + proxy2 = Proxy("5.6.7.8:443", source="test", https=True).to_json + handler._mock_db.getAll.return_value = [proxy1, proxy2] + + result = handler.getAll(https=False) + + assert len(result) == 2 + assert result[0].proxy == "1.2.3.4:8080" + assert result[1].proxy == "5.6.7.8:443" + handler._mock_db.getAll.assert_called_once_with(False) + + def test_getAll_empty_returns_empty_list(self): + """getAll 无数据返回空列表""" + handler = _make_handler() + handler._mock_db.getAll.return_value = [] + + result = handler.getAll() + + assert result == [] + + +class TestProxyHandlerExists: + """exists() 测试""" + + def test_exists_delegates_to_db(self): + """exists 传入 proxy.proxy 字符串给 DbClient""" + handler = _make_handler() + proxy = Proxy("1.2.3.4:8080", source="test") + handler._mock_db.exists.return_value = True + + result = handler.exists(proxy) + + assert result is True + handler._mock_db.exists.assert_called_once_with("1.2.3.4:8080") + + +class TestProxyHandlerGetCount: + """getCount() 测试""" + + def test_getCount_returns_dict(self): + """getCount 返回 {'count': N}""" + handler = _make_handler() + handler._mock_db.getCount.return_value = 42 + + result = handler.getCount() + + assert result == {"count": 42} \ No newline at end of file diff --git a/tests/unit/test_scheduler.py b/tests/unit/test_scheduler.py new file mode 100644 index 00000000..81c6e4e3 --- /dev/null +++ b/tests/unit/test_scheduler.py @@ -0,0 +1,143 @@ +# -*- coding: utf-8 -*- +""" +------------------------------------------------- + File Name： test_scheduler.py + Description : helper/scheduler.py 单元测试 + Author : JHao + date： 2026/6/15 +------------------------------------------------- + Change Activity: + 2026/06/15: +------------------------------------------------- +""" +__author__ = 'JHao' + +import sys +import pytest +from unittest.mock import patch, MagicMock + + +# apscheduler 依赖 pkg_resources，在 tox/uv 环境中可能缺失 +# 在 import 前 mock 掉，避免 collection 阶段报错 +_apscheduler_mock = MagicMock() +sys.modules.setdefault("apscheduler", _apscheduler_mock) +sys.modules.setdefault("apscheduler.schedulers", _apscheduler_mock.schedulers) +sys.modules.setdefault("apscheduler.schedulers.blocking", _apscheduler_mock.schedulers.blocking) +sys.modules.setdefault("apscheduler.executors", _apscheduler_mock.executors) +sys.modules.setdefault("apscheduler.executors.pool", _apscheduler_mock.executors.pool) + +import helper.scheduler as scheduler_mod + + +def _get_attr(name): + """获取模块中双下划线开头的属性（绕过类内 name mangling）""" + return getattr(scheduler_mod, name) + + +class TestRunProxyFetch: + + @patch("helper.scheduler.Checker") + @patch("helper.scheduler.Fetcher") + def test_fetcher_yields_go_to_queue(self, mock_fetcher_cls, mock_checker): + """Fetcher yield 的代理放入 queue，传给 Checker""" + mock_proxy = MagicMock() + mock_fetcher = MagicMock() + mock_fetcher.run.return_value = iter([mock_proxy]) + mock_fetcher_cls.return_value = mock_fetcher + + _get_attr("__runProxyFetch")() + + mock_fetcher_cls.assert_called_once() + mock_checker.assert_called_once() + call_args = mock_checker.call_args + assert call_args[0][0] == "raw" + + +class TestRunProxyCheck: + + @patch("helper.scheduler.__runProxyFetch") + @patch("helper.scheduler.Checker") + @patch("helper.scheduler.ProxyHandler") + def test_triggers_fetch_when_pool_low(self, mock_ph_cls, mock_checker, mock_fetch): + """count < poolSizeMin -> 触发 __runProxyFetch""" + mock_ph = MagicMock() + mock_ph.db.getCount.return_value = {"total": 5} + mock_ph.conf.poolSizeMin = 20 + mock_ph.getAll.return_value = [] + mock_ph_cls.return_value = mock_ph + + _get_attr("__runProxyCheck")() + + mock_fetch.assert_called_once() + + @patch("helper.scheduler.__runProxyFetch") + @patch("helper.scheduler.Checker") + @patch("helper.scheduler.ProxyHandler") + def test_skips_fetch_when_pool_sufficient(self, mock_ph_cls, mock_checker, mock_fetch): + """count >= poolSizeMin -> 不触发 __runProxyFetch""" + mock_ph = MagicMock() + mock_ph.db.getCount.return_value = {"total": 50} + mock_ph.conf.poolSizeMin = 20 + mock_ph.getAll.return_value = [] + mock_ph_cls.return_value = mock_ph + + _get_attr("__runProxyCheck")() + + mock_fetch.assert_not_called() + + +class TestRunScheduler: + + @patch("helper.scheduler.BlockingScheduler") + @patch("helper.scheduler.__runProxyFetch") + @patch("helper.scheduler.ConfigHandler") + @patch("helper.scheduler.LogHandler") + def test_adds_two_jobs(self, mock_log, mock_conf_cls, mock_fetch, mock_sched_cls): + """runScheduler 添加两个定时任务""" + mock_conf = MagicMock() + mock_conf.timezone = "Asia/Shanghai" + mock_conf_cls.return_value = mock_conf + mock_sched = MagicMock() + mock_sched_cls.return_value = mock_sched + + scheduler_mod.runScheduler() + + assert mock_sched.add_job.call_count == 2 + + @patch("helper.scheduler.BlockingScheduler") + @patch("helper.scheduler.__runProxyFetch") + @patch("helper.scheduler.ConfigHandler") + @patch("helper.scheduler.LogHandler") + def test_fetch_job_interval_5min(self, mock_log, mock_conf_cls, mock_fetch, mock_sched_cls): + """采集任务间隔 5 分钟""" + mock_conf = MagicMock() + mock_conf.timezone = "Asia/Shanghai" + mock_conf_cls.return_value = mock_conf + mock_sched = MagicMock() + mock_sched_cls.return_value = mock_sched + + scheduler_mod.runScheduler() + + calls = mock_sched.add_job.call_args_list + first_call = calls[0] + assert first_call[0][1] == "interval" + assert first_call[1]["minutes"] == 5 + + @patch("helper.scheduler.BlockingScheduler") + @patch("helper.scheduler.__runProxyFetch") + @patch("helper.scheduler.ConfigHandler") + @patch("helper.scheduler.LogHandler") + def test_check_job_interval_2min(self, mock_log, mock_conf_cls, mock_fetch, mock_sched_cls): + """检查任务间隔 2 分钟""" + mock_conf = MagicMock() + mock_conf.timezone = "Asia/Shanghai" + mock_conf_cls.return_value = mock_conf + mock_sched = MagicMock() + mock_sched_cls.return_value = mock_sched + + scheduler_mod.runScheduler() + + calls = mock_sched.add_job.call_args_list + second_call = calls[1] + assert second_call[0][1] == "interval" + assert second_call[1]["minutes"] == 2 \ No newline at end of file diff --git a/tests/unit/test_validator.py b/tests/unit/test_validator.py index 25f37407..effe0e24 100644 --- a/tests/unit/test_validator.py +++ b/tests/unit/test_validator.py @@ -14,9 +14,10 @@ import re import pytest +from unittest.mock import patch, MagicMock # 直接导入 IP_REGEX 和 formatValidator，不导入整个 validator 模块（避免模块级副作用） -from helper.validator import IP_REGEX, formatValidator +from helper.validator import IP_REGEX, formatValidator, httpTimeOutValidator, httpsTimeOutValidator, customValidatorExample class TestIPRegex: @@ -65,4 +66,59 @@ def test_valid_returns_true(self, proxy): "1.2.3.4", ]) def test_invalid_returns_false(self, proxy): - assert formatValidator(proxy) is False \ No newline at end of file + assert formatValidator(proxy) is False + + +class TestHttpTimeOutValidator: + """httpTimeOutValidator 测试""" + + @patch("helper.validator.head") + def test_returns_true_on_200(self, mock_head): + """status_code=200 -> True""" + mock_head.return_value = MagicMock(status_code=200) + assert httpTimeOutValidator("1.2.3.4:8080") is True + + @patch("helper.validator.head") + def test_returns_false_on_non_200(self, mock_head): + """status_code=502 -> False""" + mock_head.return_value = MagicMock(status_code=502) + assert httpTimeOutValidator("1.2.3.4:8080") is False + + @patch("helper.validator.head") + def test_returns_false_on_exception(self, mock_head): + """head() raise Timeout -> False""" + mock_head.side_effect = TimeoutError("connection timed out") + assert httpTimeOutValidator("1.2.3.4:8080") is False + + +class TestHttpsTimeOutValidator: + """httpsTimeOutValidator 测试""" + + @patch("helper.validator.head") + def test_returns_true_on_200(self, mock_head): + """status_code=200 -> True""" + mock_head.return_value = MagicMock(status_code=200) + assert httpsTimeOutValidator("1.2.3.4:8080") is True + # 验证 verify=False 被传递 + call_kwargs = mock_head.call_args + assert call_kwargs[1]["verify"] is False + + @patch("helper.validator.head") + def test_returns_false_on_non_200(self, mock_head): + """status_code=502 -> False""" + mock_head.return_value = MagicMock(status_code=502) + assert httpsTimeOutValidator("1.2.3.4:8080") is False + + @patch("helper.validator.head") + def test_returns_false_on_exception(self, mock_head): + """head() raise Timeout -> False""" + mock_head.side_effect = TimeoutError("connection timed out") + assert httpsTimeOutValidator("1.2.3.4:8080") is False + + +class TestCustomValidatorExample: + """customValidatorExample 测试""" + + def test_always_returns_true(self): + """customValidatorExample 始终返回 True""" + assert customValidatorExample("1.2.3.4:8080") is True \ No newline at end of file diff --git a/tests/unit/test_web_request.py b/tests/unit/test_web_request.py new file mode 100644 index 00000000..a62aad79 --- /dev/null +++ b/tests/unit/test_web_request.py @@ -0,0 +1,182 @@ +# -*- coding: utf-8 -*- +""" +------------------------------------------------- + File Name： test_web_request.py + Description : WebRequest 单元测试 + Author : JHao + date： 2026/6/15 +------------------------------------------------- + Change Activity: + 2026/06/15: +------------------------------------------------- +""" +__author__ = 'JHao' + +import pytest +from unittest.mock import patch, MagicMock +from requests.models import Response + +from util.webRequest import WebRequest + + +def _mock_response(status_code=200, text=None, content=None, json_data=None): + """构造 mock Response""" + resp = Response() + resp.status_code = status_code + if json_data is not None: + import json + resp._content = json.dumps(json_data).encode("utf-8") + resp.json = lambda: json_data + elif content is not None: + resp._content = content + elif text is not None: + resp._content = text.encode("utf-8") + else: + resp._content = b"ok" + return resp + + +class TestWebRequestGet: + """get() 测试""" + + @patch("util.webRequest.time.sleep") + @patch("util.webRequest.requests.get") + def test_success_path(self, mock_get, mock_sleep): + """正常返回 -> self.response 被设置""" + mock_get.return_value = _mock_response(200, "hello") + wr = WebRequest() + result = wr.get("http://example.com", retry_time=1, retry_interval=0, timeout=1) + + assert result is wr + assert wr.response.status_code == 200 + assert wr.text == "hello" + + @patch("util.webRequest.time.sleep") + @patch("util.webRequest.requests.get") + def test_custom_header_merge(self, mock_get, mock_sleep): + """自定义 header 合并到默认 header""" + mock_get.return_value = _mock_response(200) + wr = WebRequest() + wr.get("http://example.com", header={"X-Custom": "v"}, retry_time=1, retry_interval=0, timeout=1) + + call_kwargs = mock_get.call_args[1] + assert call_kwargs["headers"]["X-Custom"] == "v" + assert "User-Agent" in call_kwargs["headers"] + + @patch("util.webRequest.time.sleep") + @patch("util.webRequest.requests.get") + def test_retry_exhaustion(self, mock_get, mock_sleep): + """全部失败 -> 返回 fallback（注意 get() 的 bug: 未赋值 self.response）""" + mock_get.side_effect = TimeoutError("timeout") + wr = WebRequest() + result = wr.get("http://example.com", retry_time=2, retry_interval=0, timeout=1) + + assert result is wr + # 注意：get() 在 retry 耗尽时创建了 resp 但未赋值给 self.response + # 所以 self.response 仍为初始的空 Response + assert mock_get.call_count == 2 + + +class TestWebRequestPost: + """post() 测试""" + + @patch("util.webRequest.time.sleep") + @patch("util.webRequest.requests.post") + def test_success_path(self, mock_post, mock_sleep): + """正常返回 -> self.response 被设置""" + mock_post.return_value = _mock_response(200, "posted") + wr = WebRequest() + result = wr.post("http://example.com", retry_time=1, retry_interval=0, timeout=1) + + assert result is wr + assert wr.response.status_code == 200 + assert wr.text == "posted" + + @patch("util.webRequest.time.sleep") + @patch("util.webRequest.requests.post") + def test_custom_header_merge(self, mock_post, mock_sleep): + """自定义 header 合并到默认 header""" + mock_post.return_value = _mock_response(200) + wr = WebRequest() + wr.post("http://example.com", header={"X-Custom": "v"}, retry_time=1, retry_interval=0, timeout=1) + + call_kwargs = mock_post.call_args[1] + assert call_kwargs["headers"]["X-Custom"] == "v" + assert "User-Agent" in call_kwargs["headers"] + + @patch("util.webRequest.time.sleep") + @patch("util.webRequest.requests.post") + def test_retry_exhaustion(self, mock_post, mock_sleep): + """全部失败 -> self.response 被正确赋值（与 get() 不同）""" + mock_post.side_effect = TimeoutError("timeout") + wr = WebRequest() + result = wr.post("http://example.com", retry_time=2, retry_interval=0, timeout=1) + + assert result is wr + # post() 正确赋值 self.response = resp + assert wr.response.status_code == 200 + assert mock_post.call_count == 2 + + +class TestWebRequestTree: + """tree 属性测试""" + + def test_empty_content_returns_none(self): + """空 content -> None""" + wr = WebRequest() + wr.response = _mock_response(200, content=b"") + assert wr.tree is None + + def test_valid_html_returns_element(self): + """有效 HTML -> lxml element""" + wr = WebRequest() + html = b"

hello

" + wr.response = _mock_response(200, content=html) + tree = wr.tree + assert tree is not None + assert tree.xpath("//p/text()") == ["hello"] + + +class TestWebRequestText: + """text 属性测试""" + + def test_returns_response_text(self): + """返回 response.text""" + wr = WebRequest() + wr.response = _mock_response(200, text="hello world") + assert wr.text == "hello world" + + +class TestWebRequestJson: + """json 属性测试""" + + def test_valid_json_returns_dict(self): + """有效 JSON -> dict""" + wr = WebRequest() + wr.response = _mock_response(200, json_data={"key": "val"}) + assert wr.json == {"key": "val"} + + def test_invalid_json_returns_empty_dict(self): + """无效 JSON -> {}""" + wr = WebRequest() + resp = _mock_response(200, content=b"not json") + resp.json = lambda: (_ for _ in ()).throw(ValueError("Invalid JSON")) + wr.response = resp + assert wr.json == {} + + +class TestWebRequestProperties: + """header/user_agent 属性测试""" + + def test_user_agent_returns_string(self): + wr = WebRequest() + ua = wr.user_agent + assert isinstance(ua, str) + assert len(ua) > 0 + + def test_header_contains_user_agent(self): + wr = WebRequest() + h = wr.header + assert "User-Agent" in h + assert "Accept" in h + assert "Connection" in h