437 lines
19 KiB
Python
437 lines
19 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""TDD for xcheck.py — 跨 vendor 对账域(spec §20.12.4, 2026-09-22)。
|
|
|
|
全 mock 零网络; 日期相对 today; 所有 ROOT 重定向 tmp_path。
|
|
覆盖: 对照纯函数四层漏斗/阈值分档/五级路由/trace 递增/findings 原子追加/
|
|
wash manifest 解析/fund_side 端到端。
|
|
"""
|
|
import datetime as dt
|
|
import json
|
|
from unittest.mock import MagicMock
|
|
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
from scripts.data_platform import xcheck as xc
|
|
|
|
|
|
@pytest.fixture
|
|
def roots(tmp_path, monkeypatch):
|
|
corpus = tmp_path / "corpus"
|
|
monkeypatch.setattr(xc, "XCHECK_ROOT", corpus / "xcheck")
|
|
monkeypatch.setattr(xc, "STATIC_ROOT", tmp_path / "static")
|
|
monkeypatch.setattr(xc, "DMSK_ROOT", tmp_path / "dmsk")
|
|
monkeypatch.setattr(xc, "VB_ROOT", tmp_path / "vb")
|
|
monkeypatch.setattr(xc, "WASH_MANIFEST_DIR", tmp_path / "manifests")
|
|
monkeypatch.setattr(xc, "RATE", {k: 0.0 for k in xc.RATE})
|
|
return tmp_path
|
|
|
|
|
|
class TestCompareNoticeDates:
|
|
def test_same_dates_no_finding(self):
|
|
rows = [{"REPORT_DATE": "2026-06-30", "NOTICE_DATE": "2026-08-15"}]
|
|
assert xc.compare_notice_dates("600519.SH", rows, rows) == []
|
|
|
|
def test_diff_one_day_l2_with_evidence(self):
|
|
s = [{"REPORT_DATE": "2026-06-30 00:00:00", "NOTICE_DATE": "2026-08-15"}]
|
|
d = [{"REPORT_DATE": "2026-06-30", "NOTICE_DATE": "2026-08-14"}]
|
|
out = xc.compare_notice_dates("600519.SH", s, d)
|
|
assert len(out) == 1
|
|
f = out[0]
|
|
assert f["level"] == "L2"
|
|
assert f["check"] == "notice_date"
|
|
vals = json.loads(f["values"]) # 证据快照内嵌
|
|
assert vals == {"static": "2026-08-15", "dmsk": "2026-08-14"}
|
|
|
|
def test_null_notice_skipped(self):
|
|
s = [{"REPORT_DATE": "2026-06-30", "NOTICE_DATE": None}]
|
|
d = [{"REPORT_DATE": "2026-06-30", "NOTICE_DATE": "2026-08-14"}]
|
|
assert xc.compare_notice_dates("600519.SH", s, d) == []
|
|
|
|
|
|
class TestComparePeriodKeys:
|
|
def test_identical_no_finding(self):
|
|
p = ["2026-06-30", "2026-03-31"]
|
|
assert xc.compare_period_keys("600519.SH", p, p) == []
|
|
|
|
def test_static_missing_period_l2(self):
|
|
out = xc.compare_period_keys(
|
|
"600519.SH", ["2026-06-30", "2026-03-31"], ["2026-03-31"])
|
|
assert len(out) == 1
|
|
vals = json.loads(out[0]["values"])
|
|
assert vals["sina_only"] == ["2026-06-30"] # static 缺 → 可抓 wash 漏洗
|
|
assert vals["static_only"] == []
|
|
|
|
def test_old_static_periods_out_of_window_silent(self):
|
|
# 09-22 冒烟 bug 修: static 全史老期(103 期)不入近12期窗, 不刷噪声
|
|
old = ["%d-12-31" % y for y in range(1998, 2002)] # 4 老期
|
|
recent = (["2023-%02d-30" % m for m in (3, 6, 9, 12)] +
|
|
["2024-%02d-30" % m for m in (3, 6, 9, 12)] +
|
|
["2025-%02d-30" % m for m in (3, 6, 9, 12)] +
|
|
["2026-03-31", "2026-06-30"]) # 14 新期
|
|
sina = recent[-12:] # 与 static 最近 12 对齐
|
|
assert xc.compare_period_keys("600519.SH", sina, old + recent) == []
|
|
|
|
|
|
class TestCompareValuation:
|
|
def _vb(self, pe, pb):
|
|
return {"peTTM": pe, "pbMRQ": pb, "date": "2026-09-21"}
|
|
|
|
def test_silent_below_note_tol(self):
|
|
assert xc.compare_valuation("600519", 18.30, 5.6, self._vb(18.30, 5.6)) == []
|
|
|
|
def test_l1_between_tols(self):
|
|
out = xc.compare_valuation("600519", 18.40, None, self._vb(18.30, 5.6))
|
|
assert len(out) == 1 and out[0]["level"] == "L1"
|
|
assert out[0]["column"] == "peTTM"
|
|
|
|
def test_l2_above_warn_tol(self):
|
|
out = xc.compare_valuation("600519", None, 6.2, self._vb(5.6, 5.6))
|
|
assert len(out) == 1 and out[0]["level"] == "L2"
|
|
vals = json.loads(out[0]["values"])
|
|
assert vals["qt"] == 6.2 and vals["vb"] == 5.6 and vals["vb_date"]
|
|
|
|
def test_vb_missing_skipped(self):
|
|
assert xc.compare_valuation("600519", 18.0, 5.0, None) == []
|
|
|
|
|
|
class TestQtParse:
|
|
@staticmethod
|
|
def _qt_line(sym, name, code, pe, pb):
|
|
parts = (["1", name, code, "10.0"] + ["x"] * 35 +
|
|
[str(pe)] + ["x"] * 6 + [str(pb)])
|
|
return 'v_%s="%s";' % (sym, "~".join(parts))
|
|
|
|
def test_parse_two_lines(self):
|
|
text = (self._qt_line("sh600519", "贵州茅台", "600519", 18.38, 5.67) +
|
|
self._qt_line("sz000001", "平安银行", "000001", 4.2, 0.55))
|
|
out = xc.parse_qt_payload(text)
|
|
assert out["600519"]["pe_ttm"] == 18.38 and out["600519"]["pb"] == 5.67
|
|
assert out["000001"]["pe_ttm"] == 4.2 and out["000001"]["pb"] == 0.55
|
|
|
|
|
|
class TestTraceAndFindings:
|
|
def test_trace_ids_increment_persisted(self, roots):
|
|
a = xc.next_trace_ids("20260922", 2)
|
|
b = xc.next_trace_ids("20260922", 1)
|
|
assert a == ["xc-20260922-0001", "xc-20260922-0002"]
|
|
assert b == ["xc-20260922-0003"]
|
|
|
|
def test_append_and_read_roundtrip(self, roots):
|
|
f = [{"trace_id": "xc-20260922-0001", "ts": "t", "level": "L2",
|
|
"check": "notice_date", "symbol": "600519.SH",
|
|
"report_date": "2026-06-30", "column": "NOTICE_DATE",
|
|
"values": "{}", "deviation": "1d+", "root_cause": "unknown",
|
|
"action": "reported", "status": "discovered"}]
|
|
assert xc.append_findings(f) == 1
|
|
df = xc.read_findings()
|
|
assert len(df) == 1 and df.iloc[0]["trace_id"] == "xc-20260922-0001"
|
|
# 追加第二片
|
|
assert xc.append_findings(f) == 1
|
|
assert len(xc.read_findings()) == 2
|
|
|
|
def test_empty_findings_no_write(self, roots):
|
|
assert xc.append_findings([]) == 0
|
|
assert not (roots / "corpus" / "xcheck" / "reports").exists()
|
|
|
|
|
|
class TestWashManifest:
|
|
def test_parse_manifest_lines(self, roots):
|
|
m = roots / "manifests"
|
|
m.mkdir()
|
|
(m / "20260922.list").write_text(
|
|
"income/600519.SH_income.parquet\n"
|
|
"balance/000001.SZ_balance.parquet\n"
|
|
"garbage_line\n", encoding="utf-8")
|
|
hits = xc._wash_hit_symbols()
|
|
assert hits == [("600519.SH", "income"), ("000001.SZ", "balance")]
|
|
|
|
def test_missing_dir_empty(self, roots):
|
|
assert xc._wash_hit_symbols() == []
|
|
|
|
|
|
class TestRouteL3:
|
|
def test_l3_off_by_default(self, roots, monkeypatch):
|
|
monkeypatch.setattr(xc, "L3_AUTO", False)
|
|
f = [{"level": "L3", "symbol": "600519.SH", "trace_id": "t1",
|
|
"action": "reported", "status": "discovered"}]
|
|
assert xc.route_l3(f) == 0
|
|
assert f[0]["action"] == "reported" # 未触发
|
|
|
|
def test_l3_on_writes_pending(self, roots, monkeypatch):
|
|
monkeypatch.setattr(xc, "L3_AUTO", True)
|
|
f = [{"level": "L3", "symbol": "600519.SH", "trace_id": "t1",
|
|
"check": "notice_date"}]
|
|
assert xc.route_l3(f) == 1
|
|
pend = roots / "pending.json" # = WASH_MANIFEST_DIR.parent/pending.json
|
|
j = json.loads(pend.read_text(encoding="utf-8"))
|
|
assert j["units"][0]["symbol"] == "600519.SH"
|
|
assert j["units"][0]["trace_id"] == "t1" # trace 反链
|
|
assert f[0]["status"] == "queued_wash"
|
|
|
|
|
|
class TestFundSideEndToEnd:
|
|
def _seed_static_dmsk(self, roots):
|
|
st = roots / "static" / "income"
|
|
st.mkdir(parents=True)
|
|
pd.DataFrame({
|
|
"REPORT_DATE": ["2026-03-31", "2026-06-30"],
|
|
"NOTICE_DATE": ["2026-04-16", "2026-08-15"],
|
|
"TOTAL_OPERATE_INCOME": [100.0, 100.5], # 期末 EQ 静默
|
|
"OPERATE_COST": [200.0, 200.0], # 期末 +5% → L2
|
|
}).to_parquet(st / "600519.SH_income.parquet", index=False)
|
|
dm = roots / "dmsk" / "income" / "dt=2026-06-30"
|
|
dm.mkdir(parents=True)
|
|
pd.DataFrame({
|
|
"SECURITY_CODE": ["600519", "000001"],
|
|
"REPORT_DATE": ["2026-06-30", "2026-06-30"],
|
|
"NOTICE_DATE": ["2026-08-14", "2026-08-20"], # 600519 差一天
|
|
}).to_parquet(dm / "part-0.parquet", index=False)
|
|
|
|
def test_fund_side_full_flow(self, roots, monkeypatch):
|
|
self._seed_static_dmsk(roots)
|
|
m = roots / "manifests"
|
|
m.mkdir()
|
|
(m / "20260922.list").write_text(
|
|
"income/600519.SH_income.parquet\n", encoding="utf-8")
|
|
|
|
def _fake_table(client, symbol, rtype):
|
|
if rtype != "lrb":
|
|
return {}
|
|
return {"2025-12-31": {"BIZTOTINCO": 90.0},
|
|
"2026-03-31": {"BIZTOTINCO": 100.0, "BIZCOST": 199.9},
|
|
"2026-06-30": {"BIZTOTINCO": 100.5, "BIZCOST": 210.0}}
|
|
|
|
monkeypatch.setattr(xc, "fetch_sina_table", MagicMock(side_effect=_fake_table))
|
|
monkeypatch.setattr(xc, "_drift_gate_statement",
|
|
MagicMock(return_value=False))
|
|
findings = xc.run_fund_side()
|
|
stamped = xc._stamp(findings)
|
|
checks = {f["check"] for f in stamped}
|
|
# 键完整性: static 缺 2025-12-31(近12期窗内) → L2
|
|
assert "period_keys" in checks
|
|
pk = [f for f in stamped if f["check"] == "period_keys"][0]
|
|
assert "2025-12-31" in json.loads(pk["values"])["sina_only"]
|
|
# 披露日: 08-15 vs 08-14 → L2 含证据
|
|
nd = [f for f in stamped if f["check"] == "notice_date"]
|
|
assert len(nd) == 1 and nd[0]["level"] == "L2"
|
|
assert json.loads(nd[0]["values"]) == {"static": "2026-08-15",
|
|
"dmsk": "2026-08-14"}
|
|
# 层③ 数值: TOTAL_OPERATE_INCOME 两期 EQ 静默; OPERATE_COST 期末
|
|
# 200→210 = 5% → L2, 03-31 期 200→199.9 = 0.05% → 静默
|
|
sv = [f for f in stamped if f["check"] == "statement_value"]
|
|
assert len(sv) == 1 and sv[0]["level"] == "L2"
|
|
assert sv[0]["column"] == "OPERATE_COST"
|
|
vals = json.loads(sv[0]["values"])
|
|
assert vals == {"sina": 210.0, "static": 200.0, "field": "BIZCOST"}
|
|
assert sv[0]["report_date"] == "2026-06-30"
|
|
# trace 全量唯一
|
|
tids = [f["trace_id"] for f in stamped]
|
|
assert len(set(tids)) == len(tids)
|
|
# 新浪原料全行项 dt 分区落盘(证据快照, 供回放)
|
|
day = dt.date.today().isoformat()
|
|
raw = (roots / "corpus" / "xcheck" / "sina_fin" /
|
|
("dt=" + day) / "600519.SH_income.parquet")
|
|
assert raw.exists()
|
|
rdf = pd.read_parquet(raw)
|
|
assert set(rdf.columns) == {"report_period", "item_field", "item_value"}
|
|
assert len(rdf) == 5 # 3+2+2 行项
|
|
|
|
def test_fund_side_no_manifests_noop(self, roots):
|
|
assert xc.run_fund_side() == []
|
|
|
|
|
|
class TestSinaFieldMap:
|
|
"""Phase 2b 映射表契约: 消费方声明列(fundamental_schema _TABLE_RAW+
|
|
_BALANCE_COLS 41 列镜像)全覆盖或显式不可对照。"""
|
|
|
|
DECLARED = set(
|
|
["TOTAL_OPERATE_INCOME", "OPERATE_COST", "PARENT_NETPROFIT",
|
|
"DEDUCT_PARENT_NETPROFIT", "TOTAL_PROFIT", "INVEST_INCOME",
|
|
"FAIRVALUE_CHANGE_INCOME", "ASSET_IMPAIRMENT_LOSS",
|
|
"CREDIT_IMPAIRMENT_LOSS", "RESEARCH_EXPENSE", "SALE_EXPENSE",
|
|
"FE_INTEREST_EXPENSE", "INCOME_TAX", "BASIC_EPS",
|
|
"TOTAL_ASSETS", "TOTAL_PARENT_EQUITY", "ACCOUNTS_RECE",
|
|
"OTHER_RECE", "GOODWILL", "SHARE_CAPITAL", "SHORT_LOAN",
|
|
"SHORT_FIN_PAYABLE", "NONCURRENT_LIAB_1YEAR", "LONG_LOAN",
|
|
"BOND_PAYABLE", "LEASE_LIAB", "INVENTORY", "MONETARYFUNDS",
|
|
"ACCOUNTS_PAYABLE",
|
|
"NETCASH_OPERATE", "SALES_SERVICES", "ACCEPT_INVEST_CASH",
|
|
"FA_IR_DEPR", "IA_AMORTIZE", "LPE_AMORTIZE",
|
|
"USERIGHT_ASSET_AMORTIZE", "CONSTRUCT_LONG_ASSET",
|
|
"RECEIVE_LOAN_CASH", "ISSUE_BOND", "PAY_DEBT_CASH"])
|
|
|
|
def test_declared_cols_covered(self):
|
|
mapped = {c for dom in xc.SINA_FIELD_MAP.values() for c in dom}
|
|
assert self.DECLARED - mapped - set(xc.SINA_UNMAPPABLE) == set()
|
|
|
|
def test_every_entry_nonempty_fields(self):
|
|
for dom, cols in xc.SINA_FIELD_MAP.items():
|
|
assert dom in ("income", "balance", "cashflow")
|
|
for col, fields in cols.items():
|
|
assert fields, (dom, col)
|
|
|
|
def test_variant_fallback_resolves_bank(self):
|
|
vals = {"PARENETP": None, "NETPARECOMPPROF": "76445000000.0"}
|
|
assert xc.resolve_sina_value(vals, xc.SINA_FIELD_MAP["income"]
|
|
["PARENT_NETPROFIT"]) == 76445000000.0
|
|
|
|
def test_variant_fallback_all_null(self):
|
|
vals = {"PARENETP": None, "NETPARECOMPPROF": None}
|
|
assert xc.resolve_sina_value(vals, xc.SINA_FIELD_MAP["income"]
|
|
["PARENT_NETPROFIT"]) is None
|
|
|
|
|
|
class TestSinaFetch:
|
|
"""fetch_sina_table: 期键归一(20260630→2026-06-30, 修层①空转 bug)+行项解析。"""
|
|
|
|
@staticmethod
|
|
def _resp(report_list):
|
|
m = MagicMock()
|
|
m.json.return_value = {"result": {"data": {"report_list": report_list}}}
|
|
return m
|
|
|
|
def test_periods_normalized_and_values_parsed(self, monkeypatch):
|
|
cl = MagicMock()
|
|
cl.get.return_value = self._resp({
|
|
"20260630": {"data": [
|
|
{"item_field": "BIZTOTINCO", "item_value": "100.5"},
|
|
{"item_field": "PARENETP", "item_value": None},
|
|
{"item_field": None, "item_value": "1.0"}, # 无字段行丢弃
|
|
]},
|
|
"20260331": {"data": [
|
|
{"item_field": "BIZTOTINCO", "item_value": "abc"}, # 非数 → None
|
|
]},
|
|
})
|
|
out = xc.fetch_sina_table(cl, "600519.SH", "lrb")
|
|
assert sorted(out) == ["2026-03-31", "2026-06-30"]
|
|
assert out["2026-06-30"] == {"BIZTOTINCO": 100.5, "PARENETP": None}
|
|
assert out["2026-03-31"] == {"BIZTOTINCO": None}
|
|
|
|
def test_dashed_period_passthrough(self, monkeypatch):
|
|
cl = MagicMock()
|
|
cl.get.return_value = self._resp({
|
|
"2026-06-30": {"data": [{"item_field": "BIZTOTINCO",
|
|
"item_value": "1.0"}]}})
|
|
out = xc.fetch_sina_table(cl, "600519.SH", "lrb")
|
|
assert sorted(out) == ["2026-06-30"]
|
|
|
|
|
|
class TestCompareStatementValues:
|
|
"""层③ 三表数值对照: 阈值分档沿用估值域, 单侧空 L1 coverage。"""
|
|
|
|
ROWS = [{"REPORT_DATE": "2026-06-30", "TOTAL_OPERATE_INCOME": 100.5,
|
|
"OPERATE_COST": 200.0, "GOODWILL": float("nan")}]
|
|
|
|
def _sina(self, **over):
|
|
vals = {"BIZTOTINCO": 100.5, "BIZCOST": 200.0, "GOODWILL": None}
|
|
vals.update(over)
|
|
return {"2026-06-30": vals}
|
|
|
|
def test_eq_silent(self):
|
|
assert xc.compare_statement_values(
|
|
"600519.SH", "income", self._sina(), self.ROWS) == []
|
|
|
|
def test_mid_band_l1(self):
|
|
out = xc.compare_statement_values(
|
|
"600519.SH", "income", self._sina(BIZTOTINCO=101.0), self.ROWS)
|
|
assert len(out) == 1 and out[0]["level"] == "L1"
|
|
assert json.loads(out[0]["values"]) == {
|
|
"sina": 101.0, "static": 100.5, "field": "BIZTOTINCO"}
|
|
|
|
def test_above_warn_l2(self):
|
|
out = xc.compare_statement_values(
|
|
"600519.SH", "income", self._sina(BIZCOST=210.0), self.ROWS)
|
|
assert len(out) == 1 and out[0]["level"] == "L2"
|
|
assert out[0]["deviation"] == "0.0500"
|
|
|
|
def test_one_side_null_l1_coverage(self):
|
|
out = xc.compare_statement_values(
|
|
"600519.SH", "income", self._sina(BIZTOTINCO=100.5), [
|
|
{"REPORT_DATE": "2026-06-30", "TOTAL_OPERATE_INCOME": None,
|
|
"OPERATE_COST": 200.0}])
|
|
assert len(out) == 1 and out[0]["level"] == "L1"
|
|
assert out[0]["deviation"] == "one_side_null"
|
|
|
|
def test_static_zero_routes_coverage(self):
|
|
out = xc.compare_statement_values(
|
|
"600519.SH", "income", self._sina(BIZCOST=210.0), [
|
|
{"REPORT_DATE": "2026-06-30", "TOTAL_OPERATE_INCOME": 100.5,
|
|
"OPERATE_COST": 0.0}])
|
|
assert len(out) == 1 and out[0]["deviation"] == "one_side_null"
|
|
|
|
def test_both_null_silent(self):
|
|
assert xc.compare_statement_values(
|
|
"600519.SH", "income", self._sina(BIZCOST=None), [
|
|
{"REPORT_DATE": "2026-06-30", "TOTAL_OPERATE_INCOME": 100.5,
|
|
"OPERATE_COST": float("nan")}]) == []
|
|
|
|
def test_em_column_missing_skipped(self):
|
|
# 银行/保险变体: static 缺列 → 跳过非 finding
|
|
out = xc.compare_statement_values(
|
|
"600519.SH", "income", self._sina(), [
|
|
{"REPORT_DATE": "2026-06-30", "OPERATE_COST": 200.0}])
|
|
assert out == []
|
|
|
|
def test_period_intersection_only(self):
|
|
out = xc.compare_statement_values(
|
|
"600519.SH", "income",
|
|
{"2026-06-30": {"BIZTOTINCO": 100.5, "BIZCOST": 200.0},
|
|
"2026-09-30": {"BIZTOTINCO": 999.0}}, self.ROWS)
|
|
assert out == [] # 09-30 static 无 → 不对照; 06-30 全 EQ
|
|
|
|
|
|
class TestDriftLanes:
|
|
"""兜底档: 周期门(估值周一/三表月初)+分布聚合。"""
|
|
|
|
def test_valuation_gate_monday_only(self):
|
|
monday = dt.date(2026, 9, 21)
|
|
assert xc._drift_gate_valuation(monday) is True
|
|
assert xc._drift_gate_valuation(monday + dt.timedelta(days=1)) is False
|
|
|
|
def test_statement_gate_month_start_window(self):
|
|
assert xc._drift_gate_statement(dt.date(2026, 10, 3)) is True
|
|
assert xc._drift_gate_statement(dt.date(2026, 10, 8)) is False
|
|
|
|
def test_gates_env_off(self, monkeypatch):
|
|
monkeypatch.setenv("XCHECK_DRIFT", "0")
|
|
assert xc._drift_gate_valuation(dt.date(2026, 9, 21)) is False
|
|
assert xc._drift_gate_statement(dt.date(2026, 10, 3)) is False
|
|
|
|
def test_drift_valuation_silent_when_close(self):
|
|
snap = {"600%03d" % i: {"pe_ttm": 20.0, "pb": 3.0} for i in range(150)}
|
|
vb = pd.DataFrame(
|
|
{"peTTM": [20.05] * 150, "pbMRQ": [3.0] * 150,
|
|
"date": ["2026-09-21"] * 150, "exchange": ["SSE"] * 150},
|
|
index=list(snap))
|
|
assert xc.drift_valuation(snap, vb) == []
|
|
|
|
def test_drift_valuation_median_shift_l2(self):
|
|
snap = {"600%03d" % i: {"pe_ttm": 20.0, "pb": 3.0} for i in range(150)}
|
|
vb = pd.DataFrame(
|
|
{"peTTM": [23.0] * 150, "pbMRQ": [3.0] * 150,
|
|
"date": ["2026-09-21"] * 150, "exchange": ["SSE"] * 150},
|
|
index=list(snap))
|
|
out = xc.drift_valuation(snap, vb)
|
|
assert len(out) == 1 and out[0]["level"] == "L2"
|
|
assert out[0]["column"] == "peTTM" and out[0]["symbol"] == "__MARKET__"
|
|
vals = json.loads(out[0]["values"])
|
|
assert vals["n"] == 150
|
|
|
|
def test_drift_statement_aggregate_below_rate_silent(self):
|
|
assert xc.drift_statement_aggregate(
|
|
{"checked": 30, "l2": 1, "offenders": []}) == []
|
|
|
|
def test_drift_statement_aggregate_above_rate_l2(self):
|
|
out = xc.drift_statement_aggregate({
|
|
"checked": 20, "l2": 5,
|
|
"offenders": [{"symbol": "600519.SH", "n_l2": 3}]})
|
|
assert len(out) == 1 and out[0]["level"] == "L2"
|
|
assert out[0]["symbol"] == "__SAMPLE__"
|
|
assert json.loads(out[0]["values"])["checked"] == 20
|
|
assert json.loads(out[0]["values"])["l2"] == 5
|
|
|
|
def test_drift_statement_aggregate_zero_checked(self):
|
|
assert xc.drift_statement_aggregate({"checked": 0, "l2": 0,
|
|
"offenders": []}) == []
|