Commit 9326844a authored by wenwen.tang's avatar wenwen.tang 😕

test: 补行粒度核对与楼层总表的单测,回归脚本改数据驱动并加沙箱闸门

- 新增 tools/test_round6.py(39 项):覆盖行粒度核对全链路 ——
  进度与部分确认、旧 units 展开成行、只核单一类目、确认后取消勾选会撤销确认、
  未知行过滤、账期级部分确认放行、批次摘要字段
- 新增 tools/test_masters.py:楼层总表 公共/套内/空调总 的系数与金额口径
- test_rules.py:免租期断言改为与 `rentfree.windows()` 对拍的数据驱动写法,
  不再写死「3 段 / 只有 302室」(台账新增 801/802 后会过期)
- ui_verify.py/2/3/4/5/6.py:**未设 RENTOS_BASE 时直接拒绝运行**。
  此前 ui_verify2 直连 8899 真实服务,往 `data/batches/` 写了 17 个测试批次、
  还改脏了台账 2 行。跑这批脚本必须经 `tools/ui_verify_runner.py` 沙箱
- ui_verify2/5/7.py:批次台账改为按账期汇总默认折叠后,脚本改为只点
  `.collapsed` 分组(hash 导航不刷新页面,折叠状态会延续)
- 新增 tools/ui_verify11.py(85 项):核对按行 + 列显示 + 部分确认全链路
parent 3aa94d82
# -*- coding: utf-8 -*-
"""
电费「总表」(3F / 6F / 7F)单元测试(2026-10-13)
==================================================
对应需求(巴特菲):
在录入本期电费时**一并录入** 3F、6F、7F 总表,列有
明细类目 / 上期抄表数 / 本期抄表数 / 用量 / 电价 / 系数;**不发账单,只支持导出**。
覆盖面:
· masters.floor_of / label_of:楼层与类目的别名归一(3 / 3楼 / 3层 → 3F,公摊 → 公共)
· masters.build:默认系数(公共 12.25 / 套内 1.00 / 空调总 12.25)、
用量 = 本期 − 上期、金额 = 用量 × 电价 × 系数、只填上期 → pending、
整行留空 / 未知楼层 / 未知类目一律丢弃、允许逐行改电价与系数
· masters.totals / spec / blank_rows
· service.ingest_rows(masters=...):总表随批次留档、汇总口径单列,
**不参与出单**(通知书仍只按逐室明细生成)
· service.reingest:没传 masters 就沿用原值、传了才覆盖
· store.list_batches:总表行数 / 金额摘要
· export_batch_xlsx:有总表时多一个「总表」Sheet(含合计行),没有则仍是三个 Sheet
运行:python -m unittest discover -s tools -p "test_*.py"
"""
import os
import shutil
import tempfile
import unittest
from unittest import mock
from test_round6 import FakeElecBackend, rows_of # 复用假后端与造行工具
from rentos import config as C # noqa: E402
from rentos import masters as MST # noqa: E402
from rentos import service # noqa: E402
def mrow(floor, label, prev=None, cur=None, **kw):
d = {"floor": floor, "label": label, "prev": prev, "cur": cur}
d.update(kw)
return d
# ================================================================ 1. 规格与归一
class TestSpec(unittest.TestCase):
def test_spec_floors_and_items(self):
s = MST.spec()
self.assertEqual(s["floors"], ["3F", "6F", "7F"])
self.assertEqual([i["label"] for i in s["items"]], ["公共", "套内", "空调总"])
self.assertEqual([i["factor"] for i in s["items"]], [12.25, 1.00, 12.25])
def test_floor_aliases(self):
for v in ("3F", "3f", "3", "3楼", "3层", " 3楼总表 "):
self.assertEqual(MST.floor_of(v), "3F", v)
for v in ("6f", "7楼", "7"):
self.assertIn(MST.floor_of(v), ("6F", "7F"), v)
for bad in ("5F", "4楼", "", None, "一楼"):
self.assertEqual(MST.floor_of(bad), "", repr(bad))
def test_label_aliases(self):
self.assertEqual(MST.label_of("公共"), "公共")
self.assertEqual(MST.label_of("公摊"), "公共")
self.assertEqual(MST.label_of("套内"), "套内")
self.assertEqual(MST.label_of("室内"), "套内")
self.assertEqual(MST.label_of("空调总"), "空调总")
self.assertEqual(MST.label_of("空调总表"), "空调总")
for bad in ("餐饮", "", None):
self.assertEqual(MST.label_of(bad), "", repr(bad))
def test_blank_rows_shape(self):
rows = MST.blank_rows()
self.assertEqual(len(rows), 9) # 3 层 × 3 类目
self.assertEqual([r["floor"] for r in rows[:3]], ["3F"] * 3)
self.assertEqual([r["label"] for r in rows[:3]], ["公共", "套内", "空调总"])
self.assertEqual(rows[1]["factor"], 1.00)
self.assertIsNone(rows[0]["prev"])
# ================================================================ 2. 计费
class TestBuild(unittest.TestCase):
def test_default_price_and_factor(self):
rows = MST.build([mrow("7F", "公共", 12982.50, 13482.50),
mrow("7F", "套内", 1533.90, 1733.90),
mrow("7F", "空调总", 2967.50, 3067.50)])
self.assertEqual([r["label"] for r in rows], ["公共", "套内", "空调总"])
# 公共:500 × 1.20 × 12.25
self.assertEqual(rows[0]["qty"], 500)
self.assertAlmostEqual(rows[0]["amount"], round(500 * 1.2 * 12.25, 2))
# 套内:200 × 1.20 × 1.00
self.assertAlmostEqual(rows[1]["amount"], round(200 * 1.2 * 1.0, 2))
# 空调总:100 × 1.20 × 12.25
self.assertAlmostEqual(rows[2]["amount"], round(100 * 1.2 * 12.25, 2))
self.assertEqual(rows[0]["price"], C.DEFAULT_CONFIG["elec"]["price"])
def test_price_argument_overrides_default(self):
rows = MST.build([mrow("3F", "套内", 0, 100)], price=2)
self.assertEqual(rows[0]["price"], 2)
self.assertAlmostEqual(rows[0]["amount"], 200.0)
def test_per_row_price_and_factor_win(self):
rows = MST.build([mrow("6F", "空调总", 0, 10, price=1.5, factor=2)])
self.assertEqual(rows[0]["price"], 1.5)
self.assertEqual(rows[0]["factor"], 2)
self.assertAlmostEqual(rows[0]["amount"], 30.0)
def test_prev_only_is_pending(self):
rows = MST.build([mrow("3F", "公共", 12982.50)])
self.assertEqual(len(rows), 1)
self.assertTrue(rows[0]["pending"])
self.assertIsNone(rows[0]["qty"])
self.assertIsNone(rows[0]["amount"])
def test_empty_and_unknown_rows_dropped(self):
rows = MST.build([
mrow("3F", "公共"), # 什么都没填 → 丢
mrow("5F", "公共", 0, 10), # 楼层不在名单 → 丢
mrow("6F", "餐饮", 0, 10), # 类目不认识 → 丢
mrow("6F", "套内", 0, 10), # 正常
])
self.assertEqual(len(rows), 1)
self.assertEqual((rows[0]["floor"], rows[0]["label"]), ("6F", "套内"))
def test_qty_without_readings(self):
rows = MST.build([mrow("7F", "公共", qty=100)])
self.assertEqual(rows[0]["qty"], 100)
self.assertAlmostEqual(rows[0]["amount"], round(100 * 1.2 * 12.25, 2))
def test_order_is_floor_then_item(self):
rows = MST.build([mrow("7F", "空调总", 0, 1), mrow("3F", "公共", 0, 1),
mrow("6F", "套内", 0, 1)])
self.assertEqual([(r["floor"], r["label"]) for r in rows],
[("3F", "公共"), ("6F", "套内"), ("7F", "空调总")])
def test_totals(self):
rows = MST.build([mrow("3F", "公共", 0, 10), mrow("3F", "套内", 5)])
t = MST.totals(rows)
self.assertEqual(t["rows"], 2)
self.assertEqual(t["pending"], 1)
self.assertEqual(t["qty"], 10)
self.assertAlmostEqual(t["amount"], round(10 * 1.2 * 12.25, 2))
self.assertEqual(MST.totals(None), {"rows": 0, "pending": 0, "qty": 0, "amount": 0})
# ================================================================ 3. 录入 / 重编
class TestIngestWithMasters(FakeElecBackend):
def _masters(self):
return [mrow("3F", "公共", 0, 100), mrow("6F", "套内", 0, 50),
mrow("7F", "空调总", 0, 10)]
def test_batch_keeps_masters_and_never_bills_them(self):
b = service.ingest_rows(rows_of(("302室", "照明", 100, 180)),
period_label="2026年10月", masters=self._masters())
self.assertEqual(len(b["masters"]), 3)
t = b["totals"]
self.assertEqual(t["master_rows"], 3)
self.assertAlmostEqual(t["master_amount"],
round(100 * 1.2 * 12.25 + 50 * 1.2 * 1.0 + 10 * 1.2 * 12.25, 2))
# 总表不产生通知书:单据只有房间那一份,合计金额也只算房间
self.assertEqual(len(b["documents"]), 1)
self.assertAlmostEqual(t["total_amount"], round(80 * 1.2 * 1.0, 2))
self.assertAlmostEqual(t["document_amount"], t["total_amount"])
self.assertEqual(t["row_count"], 1)
self.assertNotIn("公共", str(b["documents"]))
def test_no_masters_is_fine(self):
b = service.ingest_rows(rows_of(("302室", "照明", 100, 180)),
period_label="2026年10月")
self.assertEqual(b["masters"], [])
self.assertEqual(b["totals"]["master_rows"], 0)
self.assertEqual(b["totals"]["master_amount"], 0)
def test_reingest_without_masters_keeps_old(self):
b = service.ingest_rows(rows_of(("302室", "照明", 100, 180)),
period_label="2026年10月", masters=self._masters())
b2 = service.reingest(b["batch_id"], rows_of(("302室", "照明", 100, 200)))
self.assertEqual(len(b2["masters"]), 3)
self.assertEqual(b2["totals"]["master_rows"], 3)
def test_reingest_with_masters_overwrites(self):
b = service.ingest_rows(rows_of(("302室", "照明", 100, 180)),
period_label="2026年10月", masters=self._masters())
b2 = service.reingest(b["batch_id"], rows_of(("302室", "照明", 100, 200)),
masters=[mrow("3F", "公共", 0, 20)])
self.assertEqual(len(b2["masters"]), 1)
self.assertEqual(b2["masters"][0]["floor"], "3F")
self.assertEqual(b2["totals"]["master_rows"], 1)
def test_reingest_with_empty_masters_clears(self):
b = service.ingest_rows(rows_of(("302室", "照明", 100, 180)),
period_label="2026年10月", masters=self._masters())
b2 = service.reingest(b["batch_id"], rows_of(("302室", "照明", 100, 200)),
masters=[])
self.assertEqual(b2["masters"], [])
# ================================================================ 4. 列表摘要 / 导出
class TestBatchListAndExport(FakeElecBackend):
def setUp(self):
super().setUp()
self.xdir = os.path.join(self.tmpdir, "exports")
os.makedirs(self.xdir, exist_ok=True)
p = mock.patch.object(C, "BATCH_EXPORT_DIR", self.xdir)
p.start()
self.addCleanup(p.stop)
def _batch(self, with_masters=True):
return service.ingest_rows(
rows_of(("302室", "照明", 100, 180)),
period_label="2026年10月",
masters=([mrow("3F", "公共", 0, 100), mrow("7F", "空调总", 0, 10)]
if with_masters else None))
# list_batches 走真实文件,故其总表摘要字段的断言放在 test_round6.TestBatchListSummary
def test_export_adds_master_sheet(self):
import openpyxl
b = self._batch()
p = service.export_batch_xlsx(b["batch_id"])
self.assertTrue(p.startswith(self.xdir))
wb = openpyxl.load_workbook(p)
try:
self.assertEqual(wb.sheetnames, ["单据", "明细", "总表", "汇总"])
ws = wb["总表"]
head = [ws.cell(3, i).value for i in range(1, 11)]
self.assertEqual(head, ["序号", "楼层", "明细类目", "上期抄表数", "本期抄表数",
"用量", "电价", "系数", "金额", "状态"])
self.assertEqual(ws.cell(4, 2).value, "3F") # 第一行 3F 公共
self.assertEqual(ws.cell(4, 3).value, "公共")
self.assertEqual(ws.cell(4, 6).value, 100)
self.assertAlmostEqual(ws.cell(4, 9).value, round(100 * 1.2 * 12.25, 2))
self.assertEqual(ws.cell(5, 2).value, "7F") # 第二行 7F 空调总
self.assertEqual(ws.cell(6, 2).value, "合计")
self.assertAlmostEqual(ws.cell(6, 9).value,
round((100 + 10) * 1.2 * 12.25, 2))
# 汇总里也要能看到总表金额
kv = {wb["汇总"].cell(i, 1).value: wb["汇总"].cell(i, 2).value
for i in range(1, wb["汇总"].max_row + 1)}
self.assertEqual(kv["总表行数"], 2)
self.assertAlmostEqual(kv["总表金额"], round(110 * 1.2 * 12.25, 2))
finally:
wb.close()
def test_export_without_masters_still_three_sheets(self):
import openpyxl
b = self._batch(with_masters=False)
wb = openpyxl.load_workbook(service.export_batch_xlsx(b["batch_id"]))
try:
self.assertEqual(wb.sheetnames, ["单据", "明细", "汇总"])
finally:
wb.close()
if __name__ == "__main__":
unittest.main(verbosity=2)
# -*- coding: utf-8 -*-
"""
第六轮需求单元测试(2026-10-13)
================================
对应需求(巴特菲):
1. 电费通知单批次台账里的数据要支持「重新编辑」—— 载入录入页改完覆盖原批次
2. 批次台账增加「核对」按钮:逐室 ✔ 核对,批次详情显示核对状态,
**确认核对后才能发送账单**
3. 电费全部手动发送,不再自动 —— 设置里去掉电费自动发送
覆盖面:
· plans:elec 计划 auto=False,配置里硬写 enabled=True 也被强制关闭;AUTO_KEYS 只剩房租
· scheduler:due_plans / fire() 默认不碰电费;plan_info 对电费返回 manual
· service.build_details:车位丢弃 / 空明细报错 / 计费口径(空调 ×12.25、照明 ×1)
· service.ingest_parts:新批次带空核对状态
· service.reingest:覆盖式重编(batch_id 不变、created_at 保留)、
发送状态按文件名沿用、核对状态重置、重编次数累加、清旧 PDF、
房租批次与不存在的批次一律拒绝
· verify_summary / set_verify:逐室进度、未勾完不许确认、确认后 ok、
勾选回退自动撤销确认、未知房间被过滤、房租批次不走核对
· store.list_batches:核对摘要字段(verify_total / verify_units / verify_ok)
运行:python -m unittest discover -s tools -p "test_*.py"
"""
import json
import os
import shutil
import sys
import tempfile
import unittest
from unittest import mock
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from rentos import attachments # noqa: E402
from rentos import config as C # noqa: E402
from rentos import elec_parser as EP # noqa: E402
from rentos import elec_render as ER # noqa: E402
from rentos import plans # noqa: E402
from rentos import scheduler # noqa: E402
from rentos import service # noqa: E402
from rentos import store # noqa: E402
FAKE_OWNERS = {
"精锐": {"name": "精锐 (上海) 仓储有限公司"},
"科博": {"name": "科博(上海)仓储有限公司"},
}
def rows_of(*specs):
"""rows_of(("302室", "照明", 100, 180), ...) → 录入行列表
带一个公司名:没有公司(或「(空置)」)的行会被 group_by_company 丢掉、不出单,
这样造出来的批次才有真实的通知书可断言。
"""
out = []
for unit, meter, prev, cur in specs:
out.append({"unit": unit, "meter": meter, "prev": prev, "cur": cur,
"company": FAKE_COMPANY})
return out
FAKE_COMPANY = "科博(上海)仓储有限公司"
def doc_name(units):
return "%s-%s-电费收费通知书.pdf" % (FAKE_COMPANY, "|".join(units))
# ================================================================ 假后端
class FakeElecBackend(unittest.TestCase):
"""电费链路的纯内存后端:配置 / 附件 / 单元映射 / 渲染全部替换掉。"""
def setUp(self):
self.batches = {}
self.pruned = []
self.tmpdir = tempfile.mkdtemp(prefix="rentos-r6-")
self.addCleanup(lambda: shutil.rmtree(self.tmpdir, ignore_errors=True))
def fake_render(docs, notice_date, owners, batch_id, due_date=None,
with_logo=True):
out = []
for d in docs:
units = [u["unit"] for u in d["units"]]
rows = [r for u in d["units"] for r in u["rows"]]
out.append({
"owner": d["owner"],
"owner_name": (owners.get(d["owner"]) or {}).get("name", ""),
"owner_missing": d["owner"] not in owners,
"company": d["company"],
"units": units,
"unit_count": len(units),
"row_count": len(rows),
"total_amount": round(sum(r["amount"] or 0 for r in rows), 2),
# 文件名只由「公司 + 单元」决定:内容不变则文件名不变
"file": doc_name(units),
"abs_path": os.path.join(self.tmpdir, "x.pdf"),
"send_state": "待发送", "sent_at": "", "send_error": "",
"period_label": d.get("period_label") or "",
})
return out
self._patches = [
mock.patch.object(store, "load_config",
lambda: {"owners": None, "elec": {}}),
mock.patch.object(store, "save_batch",
lambda b: self.batches.__setitem__(b["batch_id"], b) or b),
mock.patch.object(store, "load_batch",
lambda bid: self.batches.get(bid)),
mock.patch.object(store, "load_all_batches", self._all_batches),
mock.patch.object(service, "owners_config", lambda cfg=None: FAKE_OWNERS),
mock.patch.object(ER, "render_batch", fake_render),
mock.patch.object(EP, "load_unit_map", lambda *a, **k: {}),
mock.patch.object(EP, "load_owner_map", lambda *a, **k: {}),
mock.patch.object(attachments, "normalize", lambda items, username="": []),
mock.patch.object(store, "batch_path",
lambda bid: os.path.join(self.tmpdir, "%s.json" % bid)),
mock.patch.object(service, "prune_output",
lambda bid, keep, kind="elec": self.pruned.append((bid, set(keep)))),
]
for p in self._patches:
p.start()
self.addCleanup(lambda: [p.stop() for p in self._patches])
# ---- 造数据
def mk_batch(self, *specs, **kw):
b = service.ingest_rows(rows_of(*specs), period_label="2026年10月",
username="tester", **kw)
return self.batches[b["batch_id"]]
def _all_batches(self, kind=None):
"""内存版 store.load_all_batches —— 账期分组别去读真实 data/batches/。"""
out = [b for b in self.batches.values() if not kind or (b.get("kind") or "elec") == kind]
out.sort(key=lambda x: x.get("created_at") or "", reverse=True)
return out
def units_of(self, batch):
return sorted({d["unit"] for d in batch["details"]})
# ================================================================ 1. 计划:电费不自动发
class TestPlanNoAuto(unittest.TestCase):
def test_elec_plan_is_manual(self):
self.assertFalse(plans.is_auto("elec"))
self.assertTrue(plans.is_auto("rent_parking"))
def test_auto_keys_only_rent(self):
self.assertEqual(list(plans.AUTO_KEYS), ["rent_parking"])
self.assertIn("elec", plans.KEYS) # 计划本身还在(还有免租期等设置)
def test_config_cannot_turn_elec_auto_back_on(self):
"""老配置里如果写着 elec.enabled=True,也必须被强制关掉。"""
cfg = {"notify": {"enabled": True,
"plans": {"elec": {"enabled": True, "advance_days": 0,
"send_time": "08:00", "skip_rentfree": False}}}}
pc = plans.of(cfg, "elec")
self.assertFalse(pc["enabled"])
self.assertEqual(pc["send_time"], "08:00") # 其它设置保留
self.assertFalse(pc["skip_rentfree"])
def test_elec_spec_declares_auto_false(self):
self.assertIs(plans.meta("elec").get("auto"), False)
self.assertIs(plans.meta("rent_parking").get("auto"), True)
# ================================================================ 2. 调度器:不碰电费
class TestSchedulerSkipsElec(FakeElecBackend):
def test_plan_info_manual(self):
info = scheduler.plan_info("elec", plans.of({}, "elec"), True, "")
self.assertEqual(info["state"], "manual")
self.assertIn("手动", info["note"])
def test_fire_defaults_to_auto_plans_only(self):
"""不带 plan_key 的手动触发只能跑自动计划(电费不在内)。"""
calls = []
def fake_fire_one(key, reason):
calls.append(key)
return None
with mock.patch.object(scheduler, "_fire_one", fake_fire_one):
scheduler.fire("测试")
self.assertEqual(calls, ["rent_parking"])
def test_fire_can_still_target_elec_explicitly(self):
calls = []
with mock.patch.object(scheduler, "_fire_one",
lambda key, reason: calls.append(key)):
scheduler.fire("测试", plan_key="elec")
self.assertEqual(calls, ["elec"])
def test_due_plans_ignores_elec_even_if_enabled(self):
cfg = {"notify": {"enabled": True,
"plans": {"elec": {"enabled": True, "send_time": "00:00"}}}}
with mock.patch.object(store, "load_config", lambda: cfg), \
mock.patch.object(scheduler, "pending_batch", lambda k: {"batch_id": "x"}), \
mock.patch.object(scheduler, "_trigger_at",
lambda k, pc, b: (None, "", "")):
self.assertNotIn("elec", scheduler.due_plans())
def test_status_marks_manual(self):
with mock.patch.object(store, "load_config", lambda: {}), \
mock.patch.object(scheduler, "load_state",
lambda: {"fired": {}, "last_fired_date": "",
"last_fired_at": "", "last_result": ""}):
st = scheduler.status()
elec = next(p for p in st["plans"] if p["key"] == "elec")
self.assertFalse(elec["auto"])
self.assertEqual(elec["state"], "manual")
self.assertFalse(elec["enabled"])
self.assertEqual(st["manual_count"], 1)
# ================================================================ 3. 录入 / 计费
class TestBuildDetails(FakeElecBackend):
def test_park_rows_dropped(self):
details, parks = service.build_details(rows_of(("302室", "照明", 100, 180),
("190车位", "照明", 1, 2)))
self.assertEqual([d["unit"] for d in details], ["302室"])
self.assertEqual(parks, ["190车位"])
def test_empty_rows_raise(self):
with self.assertRaises(ValueError):
service.build_details([])
def test_meter_factor_rules(self):
details, _ = service.build_details(rows_of(("302室", "空调", 1000, 1010),
("303室", "照明", 100, 140)))
ac, room = details
self.assertEqual(ac["qty"], 10)
self.assertAlmostEqual(ac["amount"], round(10 * 1.2 * 12.25, 2))
self.assertEqual(room["qty"], 40)
self.assertAlmostEqual(room["amount"], round(40 * 1.2 * 1.0, 2))
def test_pending_when_no_current(self):
details, _ = service.build_details([{"unit": "302室", "meter": "照明", "prev": 100}])
self.assertTrue(details[0]["pending"])
self.assertIsNone(details[0]["amount"])
def test_new_batch_has_blank_verify(self):
b = self.mk_batch(("302室", "照明", 100, 180))
self.assertEqual(b["verify"], {"units": [], "rows": [], "done": False,
"at": "", "by": "", "scope": []})
self.assertEqual(service.verify_summary(b)["total"], 1)
self.assertFalse(service.verify_summary(b)["ok"])
# ================================================================ 4. 重新编辑(重编)
class TestReingest(FakeElecBackend):
def test_overwrites_same_batch_id(self):
b = self.mk_batch(("302室", "照明", 100, 180), ("303室", "照明", 100, 140))
bid, created = b["batch_id"], b["created_at"]
b2 = service.reingest(bid, rows_of(("302室", "照明", 100, 200)),
username="batefei")
self.assertEqual(b2["batch_id"], bid) # 批次号不变
self.assertEqual(b2["created_at"], created) # 创建时间不变
self.assertEqual(self.units_of(b2), ["302室"]) # 明细被覆盖
self.assertEqual(b2["reingest_count"], 1)
self.assertEqual(b2["reingest_by"], "batefei")
self.assertTrue(b2["reingest_at"])
self.assertEqual(len(self.batches), 1) # 没有多出一个批次
def test_reingest_keeps_changed_docs_pending(self):
b = self.mk_batch(("302室", "照明", 100, 180))
for d in b["documents"]:
d["send_state"], d["sent_at"] = "已发送", "2026-10-13 09:00:00"
self.batches[b["batch_id"]] = b
# ① 内容没变 → 文件名一样 → 发送状态沿用到新单据上
b2 = service.reingest(b["batch_id"], rows_of(("302室", "照明", 100, 180)))
self.assertEqual(b2["documents"][0]["send_state"], "已发送")
self.assertEqual(b2["send_state"], "已发送")
# ② 单元变了 → 文件名变了 → 新单据回到待发送(旧的不留)
b3 = service.reingest(b["batch_id"], rows_of(("302室", "照明", 100, 260),
("303室", "照明", 100, 300)))
states = {d["file"]: d["send_state"] for d in b3["documents"]}
self.assertEqual(states.get(doc_name(["302室", "303室"])), "待发送")
def test_reingest_resets_verify(self):
b = self.mk_batch(("302室", "照明", 100, 180), ("303室", "照明", 100, 140))
service.set_verify(b["batch_id"], units=["302室", "303室"], done=True)
self.assertTrue(service.verify_summary(self.batches[b["batch_id"]])["ok"])
b2 = service.reingest(b["batch_id"], rows_of(("302室", "照明", 100, 190),
("303室", "照明", 100, 150)))
vs = service.verify_summary(b2)
self.assertFalse(vs["done"])
self.assertEqual(vs["verified"], 0)
self.assertFalse(vs["ok"])
def test_reingest_rejects_rent_and_missing(self):
with self.assertRaises(ValueError):
service.reingest("NO-SUCH", rows_of(("302室", "照明", 1, 2)))
self.batches["rent-1"] = {"batch_id": "rent-1", "kind": "rent_parking",
"details": [], "documents": []}
with self.assertRaises(ValueError):
service.reingest("rent-1", rows_of(("302室", "照明", 1, 2)))
def test_reingest_prunes_stale_pdfs(self):
b = self.mk_batch(("302室", "照明", 100, 180), ("303室", "照明", 100, 140))
self.pruned.clear()
service.reingest(b["batch_id"], rows_of(("302室", "照明", 100, 200)))
self.assertTrue(self.pruned)
bid, keep = self.pruned[-1]
self.assertEqual(bid, b["batch_id"])
self.assertEqual(len(keep), 1)
self.assertIn("302室", list(keep)[0])
def test_reingest_keeps_batch_totals_in_sync(self):
b = self.mk_batch(("302室", "照明", 100, 180))
before = b["totals"]["total_amount"]
b2 = service.reingest(b["batch_id"], rows_of(("302室", "照明", 100, 260)))
after = b2["totals"]["total_amount"]
# 照明系数 1.0:160 度 × 1.20 × 1.0
self.assertAlmostEqual(after, round(160 * 1.2 * 1.0, 2))
self.assertNotEqual(before, after)
self.assertEqual(b2["totals"]["row_count"], 1)
# ================================================================ 5. 核对(按行:单元|类目)
class TestVerify(FakeElecBackend):
def _b3(self):
return self.mk_batch(("302室", "照明", 100, 180), ("303室", "照明", 100, 140),
("305室", "空调", 100, 110))
def test_progress_and_partial_confirm(self):
"""核对粒度是「行」:可以只核一部分房间,也可以只核对某一类目;部分核对也能确认。"""
b = self._b3()
bid = b["batch_id"]
_, vs = service.set_verify(bid, rows=["302室|照明"])
self.assertEqual((vs["total"], vs["verified"]), (3, 1))
self.assertEqual(vs["missing"], ["303室|照明", "305室|空调"])
self.assertFalse(vs["complete"])
self.assertFalse(vs["ok"])
# 部分核对也允许「确认核对」(ok = 已确认,闸门放行)
_, vs = service.set_verify(bid, done=True, username="batefei")
self.assertTrue(vs["ok"])
self.assertTrue(vs["partial"]) # 确认了但还有行没勾
self.assertEqual(vs["by"], "batefei")
self.assertTrue(vs["at"])
self.assertEqual(vs["missing_units"], ["303室", "305室"])
def test_legacy_units_expand_to_all_rows(self):
"""旧口径 units(房间)→ 等价于把该房间的所有行都勾上。"""
b = self._b3()
_, vs = service.set_verify(b["batch_id"], units=["302室", "303室"])
self.assertEqual((vs["total"], vs["verified"]), (3, 2))
self.assertEqual(sorted(vs["units"]), ["302室", "303室"])
self.assertEqual(vs["rows"], ["302室|照明", "303室|照明"])
def test_only_one_meter_category(self):
"""只核对「空调」:同房间的照明行仍未核对。"""
b = self.mk_batch(("302室", "照明", 100, 180), ("302室", "空调", 100, 200),
("303室", "空调", 100, 210))
_, vs = service.set_verify(b["batch_id"], rows=["302室|空调", "303室|空调"])
self.assertEqual((vs["total"], vs["verified"]), (3, 2))
self.assertEqual(vs["missing"], ["302室|照明"])
self.assertFalse(vs["complete"])
service.set_verify(b["batch_id"], done=True)
self.assertTrue(service.verify_summary(self.batches[b["batch_id"]])["partial"])
def test_unchecking_after_confirm_revokes_it(self):
b = self._b3()
bid = b["batch_id"]
service.set_verify(bid, rows=["302室|照明", "303室|照明"], done=True)
self.assertTrue(service.verify_summary(self.batches[bid])["ok"])
_, vs = service.set_verify(bid, rows=["302室|照明"]) # 少勾了一行
self.assertFalse(vs["done"])
self.assertFalse(vs["ok"])
self.assertEqual(vs["at"], "")
def test_unknown_rows_filtered(self):
b = self.mk_batch(("302室", "照明", 100, 180))
_, vs = service.set_verify(b["batch_id"], rows=["302室|照明", "999室|空调"])
self.assertEqual(vs["rows"], ["302室|照明"])
self.assertEqual(vs["total"], 1)
def test_rows_must_be_list(self):
b = self.mk_batch(("302室", "照明", 100, 180))
with self.assertRaises(ValueError):
service.set_verify(b["batch_id"], rows="302室|照明")
def test_cancel_confirm_explicitly(self):
b = self._b3()
bid = b["batch_id"]
service.set_verify(bid, rows=["302室|照明", "303室|照明"], done=True)
_, vs = service.set_verify(bid, done=False)
self.assertFalse(vs["done"])
self.assertTrue(vs["complete"] is False) # 305室 那行仍没勾
self.assertEqual(vs["verified"], 2)
def test_rent_batch_and_missing_rejected(self):
with self.assertRaises(ValueError):
service.set_verify("NO-SUCH", rows=[])
self.batches["rent-1"] = {"batch_id": "rent-1", "kind": "rent_parking",
"details": [{"unit": "302室"}], "documents": []}
with self.assertRaises(ValueError):
service.set_verify("rent-1", rows=["302室|照明"])
def test_summary_of_empty_batch_is_ok_only_when_confirmed(self):
vs = service.verify_summary({"details": [], "documents": []})
self.assertEqual(vs["total"], 0)
self.assertTrue(vs["complete"])
self.assertFalse(vs["ok"]) # 没点确认仍不放行
# ================================================================ 6. 批次摘要字段
class TestBatchListSummary(FakeElecBackend):
"""list_batches 走真实文件,所以这里单独换一个真实临时 BATCH_DIR。"""
def setUp(self):
super().setUp()
self.bdir = os.path.join(self.tmpdir, "batches")
os.makedirs(self.bdir, exist_ok=True)
p = mock.patch.object(C, "BATCH_DIR", self.bdir)
p.start()
self.addCleanup(p.stop)
# load_batch / save_batch 改回真实实现(用临时目录)
mock.patch.stopall() # 先停掉父类的内存桩
self._patches2 = [
mock.patch.object(store, "load_config",
lambda: {"owners": None, "elec": {}}),
mock.patch.object(service, "owners_config", lambda cfg=None: FAKE_OWNERS),
mock.patch.object(ER, "render_batch", self._real_render),
mock.patch.object(EP, "load_unit_map", lambda *a, **k: {}),
mock.patch.object(EP, "load_owner_map", lambda *a, **k: {}),
mock.patch.object(attachments, "normalize", lambda items, username="": []),
]
for q in self._patches2:
q.start()
self.addCleanup(lambda: [q.stop() for q in self._patches2])
def _real_render(self, docs, notice_date, owners, batch_id, due_date=None,
with_logo=True):
return [{"owner": d["owner"], "owner_name": "", "owner_missing": False,
"company": d["company"],
"units": [u["unit"] for u in d["units"]],
"unit_count": len(d["units"]),
"row_count": sum(len(u["rows"]) for u in d["units"]),
"total_amount": 0.0, "file": doc_name([u["unit"] for u in d["units"]]),
"abs_path": "/tmp/a.pdf", "send_state": "待发送",
"sent_at": "", "send_error": ""} for d in docs]
def test_verify_fields_present(self):
b = service.ingest_rows(rows_of(("302室", "照明", 100, 180),
("303室", "照明", 100, 140)),
period_label="2026年10月")
rows = store.list_batches("elec")
me = next(r for r in rows if r["batch_id"] == b["batch_id"])
self.assertEqual(me["verify_rows"], 2) # 核对粒度是「行」
self.assertEqual(me["verify_rows_done"], 0)
self.assertFalse(me["verify_ok"])
self.assertEqual(me["reingest_count"], 0)
bid = b["batch_id"]
service.set_verify(bid, rows=["302室|照明"]) # 只勾一行(部分核对)
me = next(r for r in store.list_batches("elec") if r["batch_id"] == bid)
self.assertEqual(me["verify_rows_done"], 1)
self.assertFalse(me["verify_partial"]) # 还没确认
service.set_verify(bid, done=True, username="u")
me = next(r for r in store.list_batches("elec") if r["batch_id"] == bid)
self.assertTrue(me["verify_ok"])
self.assertTrue(me["verify_partial"]) # 确认了但还有一行没勾
self.assertTrue(me["verify_at"])
def test_master_summary_fields(self):
"""总表(3F/6F/7F)只记账:列表里单列行数与金额,不并入出单金额。"""
b = service.ingest_rows(
rows_of(("302室", "照明", 100, 180)), period_label="2026年10月",
masters=[{"floor": "3F", "label": "公共", "prev": 0, "cur": 100},
{"floor": "7F", "label": "空调总", "prev": 0, "cur": 10}])
me = next(r for r in store.list_batches("elec") if r["batch_id"] == b["batch_id"])
self.assertEqual(me["master_rows"], 2)
self.assertAlmostEqual(me["master_amount"],
round((100 + 10) * 1.2 * 12.25, 2))
self.assertEqual(me["tenant_count"], 1) # 总表不算单据
# ================================================================ 4. 账期级核对
class TestPeriodVerify(FakeElecBackend):
"""按**账期**统一核对:一个账期一个核对入口,一次确认覆盖该账期全部批次。
口径(见 service.period_summary):
· 房间 = 该账期所有批次的并集;
· 一个房间算「已核对」⇔ **含它的每一张批次**都勾了它;
· 账期 ok ⇔ 每个房间都勾上 且 每张批次都点了「确认核对」。
"""
def setUp(self):
super().setUp()
self._seq = 0
def _bid(*a, **k): # 同一测试里连造多张批次也不会撞号
self._seq += 1
return "P%02d" % self._seq
p = mock.patch.object(service, "make_batch_id", _bid)
p.start()
self.addCleanup(p.stop)
def _ingest(self, *specs, period="2026年10月"):
b = service.ingest_rows(rows_of(*specs), period_label=period, username="t")
return self.batches[b["batch_id"]]
def test_period_groups_batches_and_unions_rooms(self):
a = self._ingest(("302室", "照明", 100, 180), ("303室", "照明", 100, 140))
c = self._ingest(("303室", "照明", 200, 260), ("304室", "照明", 10, 40))
ps = service.period_summary("2026年10月")
self.assertEqual(ps["batch_count"], 2)
self.assertEqual(sorted(ps["batch_ids"]), sorted([a["batch_id"], c["batch_id"]]))
self.assertEqual(sorted(ps["rooms"]), ["302室", "303室", "304室"])
self.assertEqual((ps["total"], ps["verified"]), (3, 0))
self.assertFalse(ps["complete"])
self.assertFalse(ps["ok"])
def test_row_checked_only_if_every_batch_checked_it(self):
a = self._ingest(("302室", "照明", 100, 180), ("303室", "照明", 100, 140))
self._ingest(("303室", "照明", 200, 260), ("304室", "照明", 10, 40))
# 只有 A 勾了 303室(C 也含 303室 却没勾)→ 该行在账期口径下仍算「未核对」
service.set_verify(a["batch_id"], rows=["302室|照明", "303室|照明"])
ps = service.period_summary("2026年10月")
self.assertEqual(ps["checked"], ["302室|照明"]) # 302室 只在 A 里 → 已核对
self.assertIn("303室|照明", ps["missing"])
def test_done_allows_partial(self):
"""账期确认核对**不要求全部行勾完**(部分核对也能确认)。"""
self._ingest(("302室", "照明", 100, 180))
self._ingest(("303室", "照明", 100, 140))
ps, bs = service.set_period_verify(
"2026年10月", rows=["302室|照明"], done=True, username="fate")
self.assertEqual(len(bs), 2)
self.assertTrue(ps["ok"])
self.assertTrue(ps["partial"]) # 303室 那行没勾
for b in bs: # 每张批次都成了「已核对」
self.assertTrue(service.verify_summary(b)["ok"])
again = service.period_summary("2026年10月")
self.assertEqual(again["checked"], ["302室|照明"])
self.assertEqual(again["missing"], ["303室|照明"])
self.assertEqual(again["by"], "fate")
def test_done_writes_every_batch(self):
self._ingest(("302室", "照明", 100, 180), ("303室", "照明", 100, 140))
self._ingest(("303室", "照明", 200, 260), ("304室", "照明", 10, 40))
ps, bs = service.set_period_verify(
"2026年10月", rows=["302室|照明", "303室|照明", "304室|照明"],
done=True, username="fate")
self.assertEqual(len(bs), 2)
for b in bs: # 每张批次都成了「已核对」(发送闸门看这个)
self.assertTrue(service.verify_summary(b)["ok"])
again = service.period_summary("2026年10月")
self.assertTrue(again["ok"])
self.assertTrue(again["done"])
self.assertFalse(again["partial"])
self.assertEqual(sorted(again["checked"]),
["302室|照明", "303室|照明", "304室|照明"])
self.assertEqual(again["by"], "fate")
def test_later_uncheck_revokes_period_ok(self):
a = self._ingest(("302室", "照明", 100, 180))
service.set_period_verify("2026年10月", rows=["302室|照明"], done=True,
username="fate")
self.assertTrue(service.period_summary("2026年10月")["ok"])
service.set_verify(a["batch_id"], rows=[]) # 撤回勾选 → 自动撤销确认
self.assertFalse(service.period_summary("2026年10月")["ok"])
def test_periods_are_isolated_by_label(self):
a = self._ingest(("302室", "照明", 100, 180))
b = self._ingest(("901室", "照明", 100, 180), period="2026年11月")
self.assertEqual(service.period_summary("2026年10月")["batch_ids"], [a["batch_id"]])
self.assertEqual(service.period_summary("2026年11月")["batch_ids"], [b["batch_id"]])
def test_list_periods_counts_per_period(self):
self._ingest(("302室", "照明", 100, 180))
self._ingest(("303室", "照明", 100, 140))
self._ingest(("901室", "照明", 100, 180), period="2026年11月")
got = {p["period"]: p["batch_count"] for p in service.list_periods("elec")}
self.assertEqual(got, {"2026年10月": 2, "2026年11月": 1})
def test_empty_period_rejected(self):
self._ingest(("302室", "照明", 100, 180))
with self.assertRaises(ValueError):
service.set_period_verify("2026年12月", rows=[], done=True)
if __name__ == "__main__":
unittest.main(verbosity=2)
...@@ -41,8 +41,10 @@ class TestPlans(unittest.TestCase): ...@@ -41,8 +41,10 @@ class TestPlans(unittest.TestCase):
"skip_rentfree": 0}) "skip_rentfree": 0})
self.assertEqual(p["advance_days"], plans.MAX_ADVANCE) self.assertEqual(p["advance_days"], plans.MAX_ADVANCE)
self.assertEqual(p["send_time"], "08:05") self.assertEqual(p["send_time"], "08:05")
self.assertTrue(p["enabled"])
self.assertFalse(p["skip_rentfree"]) self.assertFalse(p["skip_rentfree"])
# 电费已改「全部手动发送」:配置里写 enabled=true 也必须被强制关掉(其余设置保留)
self.assertFalse(p["enabled"])
self.assertTrue(plans.normalize("rent_parking", {"enabled": 1})["enabled"])
bad = plans.normalize("elec", {"advance_days": "abc", "send_time": "25:00", "enabled": False}) bad = plans.normalize("elec", {"advance_days": "abc", "send_time": "25:00", "enabled": False})
self.assertEqual(bad["advance_days"], plans.BY_KEY["elec"]["advance_days"]) self.assertEqual(bad["advance_days"], plans.BY_KEY["elec"]["advance_days"])
...@@ -90,22 +92,37 @@ class TestRentfree(unittest.TestCase): ...@@ -90,22 +92,37 @@ class TestRentfree(unittest.TestCase):
self.assertEqual(rentfree.norm("706b"), "706B") self.assertEqual(rentfree.norm("706b"), "706B")
def test_windows_from_real_table(self): def test_windows_from_real_table(self):
"""真实台账里的三段免租期:315室 / 302室 / 303室(各 0.5 个月)。""" """真实台账里的免租窗口:首N月(315室 / 302室 / 303室)+ 每年8-9月(逸腾 801/802)。
ws = {w["unit"]: w for w in rentfree.windows()}
self.assertIn("302室", ws) 只断言结构与口径不变,**不写死日期集合** —— 台账新增/删除免租行都不该让本用例变红。
self.assertEqual(ws["302室"]["begin"], "2026-08-16") """
self.assertEqual(ws["302室"]["end"], "2026-08-31") ws = rentfree.windows()
for w in ws.values(): by_unit = {}
for w in ws:
by_unit.setdefault(w["unit"], []).append(w)
for u in ("302室", "303室", "315室"):
self.assertIn(u, by_unit)
# 302室首N月窗口:8-16 起(含起始不含结束)
first = [w for w in by_unit["302室"] if w["type"] == "首N月免租"][0]
self.assertEqual(first["begin"], "2026-08-16")
self.assertEqual(first["end"], "2026-08-31")
for w in ws:
self.assertGreater(w["months"], 0) self.assertGreater(w["months"], 0)
self.assertLess(w["begin"], w["end"]) self.assertLess(w["begin"], w["end"])
def test_active_and_hit(self): def test_active_and_hit(self):
self.assertEqual([a["unit"] for a in rentfree.active("2026-08-20")], ["302室"]) """active() 与 windows() 同一口径(含起始不含结束)——用 windows 对拍,数据驱动。"""
self.assertEqual([a["unit"] for a in rentfree.active("2026-09-10")], ["303室"]) for d in ("2026-01-01", "2026-01-16", "2026-08-16", "2026-08-20", "2026-08-31",
self.assertEqual(rentfree.active("2026-10-09"), []) # 今天都不在免租期 "2026-09-10", "2026-10-01", "2026-10-09", "2027-08-01"):
# 边界:结束日当天已恢复正常(含起始不含结束) day = dt.date.fromisoformat(d)
self.assertEqual(rentfree.active("2026-08-31"), []) want = sorted({w["unit"] for w in rentfree.windows()
self.assertEqual([a["unit"] for a in rentfree.active("2026-08-16")], ["302室"]) if w["_begin"] <= day < w["_end"]})
self.assertEqual(sorted(a["unit"] for a in rentfree.active(d)), want,
msg=f"active({d}) 与 windows() 口径不一致")
# 302室 0.5 月窗口的边界:含起始不含结束
self.assertIn("302室", [a["unit"] for a in rentfree.active("2026-08-16")])
self.assertNotIn("302室", [a["unit"] for a in rentfree.active("2026-08-31")])
self.assertNotIn("302室", [a["unit"] for a in rentfree.active("2026-10-09")])
hits = rentfree.hit(["302室", "306室"], "2026-08-20") hits = rentfree.hit(["302室", "306室"], "2026-08-20")
self.assertEqual([h["matched"] for h in hits], ["302室"]) self.assertEqual([h["matched"] for h in hits], ["302室"])
...@@ -216,24 +233,33 @@ class TestScheduler(unittest.TestCase): ...@@ -216,24 +233,33 @@ class TestScheduler(unittest.TestCase):
return scheduler.status() return scheduler.status()
def test_trigger_date_is_due_minus_advance(self): def test_trigger_date_is_due_minus_advance(self):
pc = plans.all_of(CFG_ON)["elec"] """房租车位费:next_at = 「下一账单日」− 提前天数。
(电费 2026-10-13 起改为 100% 手动发送,不再排期,见 TestPlans /
test_round6.TestSchedulerSkipsElec)
"""
from rentos import tables
rows = [{"_id": "x", "A": "精锐", "B": "3楼", "C": "302室", "F": "逢林",
"I": "2026-01-01", "J": "2028-12-31", "K": 24, "L": 0.5, "N": 3,
"O": 1008, "P": 6570, "AC": "2026-12-01", "AJ": "2026-12-01"}]
pc = {"advance_days": 5, "enabled": True, "send_time": "09:00"}
def info(now, fired=""): def info(now, fired=""):
with mock.patch.object(store, "list_batches", with mock.patch.object(tables, "load_table",
return_value=[{"batch_id": BATCH["batch_id"]}]), \ return_value={"rows": rows, "columns": []}), \
mock.patch.object(store, "load_batch", return_value=dict(BATCH)): mock.patch.object(scheduler, "_auto_pending", return_value=None):
return scheduler.plan_info("elec", pc, True, fired, now) return scheduler.plan_info("rent_parking", pc, True, fired, now)
# 缴费期限 08-24、提前 3 天 → 触发日 08-21 09:00 # 下一账单日 12-01、提前 5 天 → 推送日 11-26 09:00
wait = info(dt.datetime(2026, 8, 20, 10, 0)) wait = info(dt.datetime(2026, 11, 20, 10, 0))
self.assertEqual(wait["next_at"], "2026-08-21 09:00") self.assertEqual(wait["next_at"], "2026-11-26 09:00")
self.assertEqual(wait["state"], "wait") self.assertEqual(wait["state"], "wait")
self.assertIn("还有 1 天", wait["note"]) self.assertIn("还有 6 天", wait["note"])
due = info(dt.datetime(2026, 8, 21, 9, 5)) due = info(dt.datetime(2026, 11, 26, 9, 5))
self.assertEqual(due["state"], "due") self.assertEqual(due["state"], "due")
fired = info(dt.datetime(2026, 8, 21, 9, 5), "2026-08-21") fired = info(dt.datetime(2026, 11, 26, 9, 5), "2026-11-26")
self.assertEqual(fired["state"], "fired") self.assertEqual(fired["state"], "fired")
def test_rent_parking_schedules_by_next_bill(self): def test_rent_parking_schedules_by_next_bill(self):
...@@ -275,7 +301,12 @@ class TestScheduler(unittest.TestCase): ...@@ -275,7 +301,12 @@ class TestScheduler(unittest.TestCase):
def test_master_switch_off(self): def test_master_switch_off(self):
cfg = {"notify": dict(CFG_ON["notify"], enabled=False)} cfg = {"notify": dict(CFG_ON["notify"], enabled=False)}
st = self._status(cfg, {"fired": {}}) st = self._status(cfg, {"fired": {}})
self.assertTrue(all(p["state"] == "off" for p in st["plans"])) by = {p["key"]: p for p in st["plans"]}
self.assertEqual(by["rent_parking"]["state"], "off")
# 电费是「手动发送」:总开关开不开都不排期
self.assertEqual(by["elec"]["state"], "manual")
self.assertEqual(st["enabled_count"], 0)
self.assertEqual(st["manual_count"], 1)
def test_due_plans_respects_flags_and_date(self): def test_due_plans_respects_flags_and_date(self):
def due(now, fired=None, cfg=CFG_ON): def due(now, fired=None, cfg=CFG_ON):
...@@ -286,11 +317,11 @@ class TestScheduler(unittest.TestCase): ...@@ -286,11 +317,11 @@ class TestScheduler(unittest.TestCase):
mock.patch.object(scheduler, "load_state", return_value={"fired": fired or {}}): mock.patch.object(scheduler, "load_state", return_value={"fired": fired or {}}):
return scheduler.due_plans(now) return scheduler.due_plans(now)
self.assertEqual(due(dt.datetime(2026, 8, 20, 9, 30)), []) # 未到触发日 # 电费已改全手动:即使到了「缴费期限 − 提前天数」那天也不该被自动触发
self.assertEqual(due(dt.datetime(2026, 8, 21, 8, 59)), []) # 到日未到时点 self.assertEqual(due(dt.datetime(2026, 8, 20, 9, 30)), [])
self.assertEqual(due(dt.datetime(2026, 8, 21, 9, 0)), ["elec"]) # 到日到时点 self.assertEqual(due(dt.datetime(2026, 8, 21, 9, 0)), [])
self.assertEqual(due(dt.datetime(2026, 8, 22, 10, 0), self.assertEqual(due(dt.datetime(2026, 8, 22, 10, 0),
fired={"elec": "2026-08-22"}), []) # 当天已触发 fired={"elec": "2026-08-22"}), [])
off = {"notify": dict(CFG_ON["notify"], enabled=False)} off = {"notify": dict(CFG_ON["notify"], enabled=False)}
self.assertEqual(due(dt.datetime(2026, 8, 25, 10, 0), cfg=off), []) self.assertEqual(due(dt.datetime(2026, 8, 25, 10, 0), cfg=off), [])
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment