SHA256
30 lines
1.3 KiB
Python
30 lines
1.3 KiB
Python
"""验证 sector.fetch_sector 概念板块抓取逻辑"""
|
|
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
from src.fetchers import sector
|
|
|
|
|
|
def test_concept_skips_when_enough_data():
|
|
"""已有数量达到当前板块数量时，跳过抓取。"""
|
|
mock_session = MagicMock()
|
|
mock_session.execute.side_effect = [MagicMock(scalar=MagicMock(return_value=150000)),
|
|
MagicMock(scalar=MagicMock(return_value=460))]
|
|
with patch.object(sector, "get_session", return_value=mock_session), \
|
|
patch.object(sector, "_fetch_concept_list", return_value=[{"code": "880001", "name": "测试板块"}]) as mock_list:
|
|
sector.fetch_sector()
|
|
|
|
mock_list.assert_called_once_with()
|
|
|
|
|
|
def test_concept_fetches_when_insufficient():
|
|
"""板块数量不足时，执行抓取。"""
|
|
mock_session = MagicMock()
|
|
mock_session.execute.side_effect = [MagicMock(scalar=MagicMock(return_value=0)),
|
|
MagicMock(scalar=MagicMock(return_value=0))]
|
|
|
|
with patch.object(sector, "get_session", return_value=mock_session), \
|
|
patch.object(sector, "_fetch_concept_list", return_value=[{"code": "880001", "name": "测试板块"}]), \
|
|
patch.object(sector, "_fetch_concept_stocks", return_value=[]):
|
|
sector.fetch_sector()
|