FinDataHub
11d36d1c1e
问题 - 索引中该目录记为小写 `scripts/`，工作区实为 `Scripts/`。因 core.ignorecase=true， 本地 `git diff -- Scripts/bin/x.py` 会**静默返回空**（看似无改动）， 且在大小写敏感环境（Linux CI / Docker 挂载）会检出 `scripts/`， 与全仓所有 `Scripts/...` 引用不匹配。 - 该目录此前 10 个脚本（check_python_version.py、perf_snapshot.py、 install_collector_launchd.py、mysql_toggle.sh 等）从未入库， 而 CI 第 31 行正调用 `Scripts/bin/check_python_version.py --strict` —— 全新检出必然失败。 改动 - 索引路径统一为 `Scripts/`（工作区文件未移动，仅索引改名）。 - 补齐上述 10 个未跟踪脚本及其测试。 影响 - 修正前索引 35 条（全为小写路径），修正后 45 条。
205 lines
7.9 KiB
Python
205 lines
7.9 KiB
Python
# coding=utf-8
|
||
"""newsflash.py 快讯查询脚本单元测试.
|
||
|
||
重点覆盖：
|
||
- _tabulate：mysql 制表符输出解析
|
||
- run_sql：pymysql 路径（None→空串、异常降级）+ mysql 命令行 fallback（参数转义）
|
||
- main：SQL 条件拼接（来源/关键词/时间参数化、分页、无筛选时统计查询）
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import sys
|
||
from types import SimpleNamespace
|
||
from unittest.mock import MagicMock
|
||
|
||
import pytest
|
||
|
||
# 将 Scripts/bin 加入 sys.path
|
||
_bin_dir = os.path.join(os.path.dirname(__file__), "..")
|
||
if _bin_dir not in sys.path:
|
||
sys.path.insert(0, _bin_dir)
|
||
|
||
import newsflash
|
||
|
||
# conftest 注入的 pymysql mock 缺少 cursors.DictCursor，补齐
|
||
_pymysql_mock = sys.modules.get("pymysql")
|
||
if _pymysql_mock is not None and not hasattr(_pymysql_mock, "cursors"):
|
||
_pymysql_mock.cursors = SimpleNamespace(DictCursor=dict)
|
||
|
||
|
||
def _run_main(monkeypatch, argv, run_sql_result=None):
|
||
"""mock run_sql 并执行 main，返回记录的 (sql, params) 调用列表."""
|
||
calls = []
|
||
|
||
def fake_run_sql(sql, params=None):
|
||
calls.append((sql, params or ()))
|
||
return run_sql_result if run_sql_result is not None else []
|
||
|
||
monkeypatch.setattr(newsflash, "run_sql", fake_run_sql)
|
||
monkeypatch.setattr(sys, "argv", ["newsflash.py"] + argv)
|
||
newsflash.main()
|
||
return calls
|
||
|
||
|
||
# ── _tabulate ──────────────────────────────────────────────
|
||
|
||
|
||
class TestTabulate:
|
||
def test_parses_header_and_rows(self):
|
||
out = "id\tname\n1\tabc\n2\tdef\n"
|
||
rows = newsflash._tabulate(out)
|
||
assert rows == [{"id": "1", "name": "abc"}, {"id": "2", "name": "def"}]
|
||
|
||
def test_empty_output(self):
|
||
assert newsflash._tabulate("") == []
|
||
|
||
def test_blank_lines_ignored(self):
|
||
out = "id\tname\n\n1\tabc\n\n"
|
||
assert len(newsflash._tabulate(out)) == 1
|
||
|
||
def test_missing_tabs(self):
|
||
out = "a\tb\n1\n"
|
||
rows = newsflash._tabulate(out)
|
||
assert rows[0]["a"] == "1"
|
||
assert "b" not in rows[0]
|
||
|
||
|
||
# ── run_sql（pymysql 路径）─────────────────────────────────
|
||
|
||
|
||
class TestRunSql:
|
||
def _patch_db(self, monkeypatch, rows):
|
||
"""mock get_conn/put_conn，返回可遍历的 dict 行."""
|
||
conn = MagicMock()
|
||
cur = MagicMock()
|
||
cur.fetchall.return_value = rows
|
||
conn.cursor.return_value = cur
|
||
get_conn = MagicMock(return_value=conn)
|
||
put_conn = MagicMock()
|
||
monkeypatch.setattr(newsflash, "get_conn", get_conn)
|
||
monkeypatch.setattr(newsflash, "put_conn", put_conn)
|
||
return get_conn, put_conn, conn, cur
|
||
|
||
def test_returns_rows_with_none_as_empty(self, monkeypatch):
|
||
_, _, _, cur = self._patch_db(monkeypatch, [{"id": 1, "name": None}])
|
||
rows = newsflash.run_sql("SELECT 1", (1,))
|
||
assert rows == [{"id": "1", "name": ""}]
|
||
cur.execute.assert_called_once_with("SELECT 1", (1,))
|
||
|
||
def test_empty_rows(self, monkeypatch):
|
||
self._patch_db(monkeypatch, [])
|
||
assert newsflash.run_sql("SELECT 1") == []
|
||
|
||
def test_exception_returns_none(self, monkeypatch):
|
||
conn = MagicMock()
|
||
conn.cursor.side_effect = RuntimeError("boom")
|
||
monkeypatch.setattr(newsflash, "get_conn", MagicMock(return_value=conn))
|
||
monkeypatch.setattr(newsflash, "put_conn", MagicMock())
|
||
assert newsflash.run_sql("SELECT 1") is None
|
||
|
||
def test_put_conn_called(self, monkeypatch):
|
||
_, put_conn, _, _ = self._patch_db(monkeypatch, [])
|
||
newsflash.run_sql("SELECT 1")
|
||
put_conn.assert_called_once()
|
||
|
||
|
||
# ── run_sql（mysql 命令行 fallback）────────────────────────
|
||
|
||
|
||
class TestRunSqlFallback:
|
||
def _enable_fallback(self, monkeypatch):
|
||
"""切换 HAS_PYMYSQL=False 并注入 fake subprocess."""
|
||
monkeypatch.setattr(newsflash, "HAS_PYMYSQL", False)
|
||
fake_subprocess = MagicMock()
|
||
monkeypatch.setattr(newsflash, "subprocess", fake_subprocess, raising=False)
|
||
return fake_subprocess
|
||
|
||
def test_escapes_string_params(self, monkeypatch):
|
||
fake_subprocess = self._enable_fallback(monkeypatch)
|
||
fake_subprocess.run.return_value = SimpleNamespace(returncode=0, stdout="a\tb\n1\t2\n")
|
||
newsflash.run_sql("SELECT %s", ["o'brien"])
|
||
cmd = fake_subprocess.run.call_args.args[0]
|
||
# 单引号被转义为 ''，占位符被替换
|
||
assert "o''brien" in cmd[-1]
|
||
assert "%s" not in cmd[-1]
|
||
|
||
def test_success_parses_output(self, monkeypatch):
|
||
fake_subprocess = self._enable_fallback(monkeypatch)
|
||
fake_subprocess.run.return_value = SimpleNamespace(returncode=0, stdout="id\tname\n1\tabc\n")
|
||
assert newsflash.run_sql("SELECT 1") == [{"id": "1", "name": "abc"}]
|
||
|
||
def test_failure_returns_none(self, monkeypatch):
|
||
fake_subprocess = self._enable_fallback(monkeypatch)
|
||
fake_subprocess.run.return_value = SimpleNamespace(returncode=1, stdout="error")
|
||
assert newsflash.run_sql("SELECT 1") is None
|
||
|
||
|
||
# ── main：SQL 条件拼接 ─────────────────────────────────────
|
||
|
||
|
||
class TestMain:
|
||
def test_no_filter_runs_stats_queries(self, monkeypatch, capsys):
|
||
_run_main(monkeypatch, [])
|
||
result = json.loads(capsys.readouterr().out)
|
||
assert result["status"] == "ok"
|
||
assert result["latest"] == []
|
||
|
||
def test_source_filter_parameterized(self, monkeypatch, capsys):
|
||
calls = _run_main(monkeypatch, ["--source", "cls"])
|
||
result = json.loads(capsys.readouterr().out)
|
||
list_sql, list_params = calls[0]
|
||
assert "s.code = %s" in list_sql
|
||
assert "cls" in list_params
|
||
# 有筛选时不执行统计查询
|
||
all_sql = "\n".join(s for s, _ in calls)
|
||
assert "GROUP BY s.name" not in all_sql
|
||
assert result["status"] == "ok"
|
||
|
||
def test_keyword_filter_parameterized(self, monkeypatch, capsys):
|
||
calls = _run_main(monkeypatch, ["--keyword", "央行"])
|
||
list_sql, list_params = calls[0]
|
||
assert "f.title LIKE %s" in list_sql
|
||
assert "%央行%" in list_params
|
||
|
||
def test_hours_filter(self, monkeypatch, capsys):
|
||
calls = _run_main(monkeypatch, ["--hours", "24"])
|
||
list_sql, list_params = calls[0]
|
||
assert "f.published_at >= %s" in list_sql
|
||
assert isinstance(list_params[-2], int) # 毫秒时间戳
|
||
|
||
def test_pagination_params(self, monkeypatch, capsys):
|
||
calls = _run_main(monkeypatch, ["--limit", "20", "--offset", "40"])
|
||
list_sql, list_params = calls[0]
|
||
assert "LIMIT %s OFFSET %s" in list_sql
|
||
assert list_params[-2:] == [20, 40]
|
||
|
||
def test_stats_queries_present_when_no_filter(self, monkeypatch, capsys):
|
||
calls = _run_main(monkeypatch, [])
|
||
all_sql = "\n".join(s for s, _ in calls)
|
||
assert "GROUP BY s.name" in all_sql
|
||
assert "GROUP BY day" in all_sql
|
||
assert "GROUP BY importance" in all_sql
|
||
assert "importance >= 3" in all_sql
|
||
assert "daily_summary" in all_sql
|
||
|
||
def test_result_aggregation(self, monkeypatch, capsys):
|
||
"""统计查询返回数据时正确聚合到 result."""
|
||
|
||
def fake_run_sql(sql, params=None):
|
||
if "COUNT(*) as total" in sql and "news_flash f" in sql:
|
||
return [{"total": 10}]
|
||
if "GROUP BY importance" in sql:
|
||
return [{"importance": 5, "count": 3}]
|
||
if "FROM daily_summary" in sql:
|
||
return [{"summary": "总结", "updated_at": "2026-08-03"}]
|
||
return []
|
||
|
||
monkeypatch.setattr(newsflash, "run_sql", fake_run_sql)
|
||
monkeypatch.setattr(sys, "argv", ["newsflash.py"])
|
||
newsflash.main()
|
||
result = json.loads(capsys.readouterr().out)
|
||
assert result["total"] == 10
|
||
assert result["importance_dist"] == {"5": 3}
|
||
assert result["daily_summary"]["summary"] == "总结"
|