Commit bcbfeeba authored by wenwen.tang's avatar wenwen.tang 😕

test(contracts): 单测 20 项 + API 冒烟 24 项 + UI 回归 30 项;沙箱补 CONTRACT_DIR

- test_contracts.py(新,20 项):解码白名单 / 超限 / 非法 base64、存取与路径穿越、
  列值 parse/dump/describe 容错、AL 列登记、coerce、files_text、
  台账写读 + 导出 xlsx 是文件名、合同目录与备注图片目录互不干扰
- smoke_verify_contracts.py(新,24 项,经 runner 沙箱):上传 → 查阅 200 +
  application/pdf + 字节一致 → ?dl=1 是 attachment → 写 AL → 导出 xlsx 是文件名 →
  访客上传/删 403 但可查阅 → 非法类型 400 → 穿越 404 → 清空 AL → 删文件后 404
- ui_verify12.py(新,30 项,自带沙箱 + 真浏览器):表头「合同原件」在最右、
  只读无上传入口、编辑态「+ 合同」→ 浮层 → 真实文件选择器上传 → 三个操作齐全 →
  查阅链接带 token 且返回原字节 → 关浮层变「📎 1 份」且 pending → 保存落盘 →
  只读浮层只有查阅/下载 → 移除后 AL 清空但服务器文件仍在 → 无 console 错误
- 沙箱常量清单补 CONTRACT_DIR(runner + ui_verify7/8/9/10/11 + check_scheduler 共 7 处),
  漏了会像 DRAFT_DIR 那样把合同写进真实 data/
- test_rules:列数 31→32,并断言「合同原件」排最后、AJ 紧随其后
parent 7a75747e
......@@ -50,6 +50,7 @@ def sandbox():
"TABLES_DIR": os.path.join(data, "tables"),
"ATTACH_DIR": os.path.join(data, "attachments"),
"DRAFT_DIR": os.path.join(data, "drafts"),
"CONTRACT_DIR": os.path.join(data, "contracts"),
"CONFIG_FILE": os.path.join(data, "config.json"),
"SEND_LOG_FILE": os.path.join(data, "send_log.json"),
"SESSION_FILE": os.path.join(data, "sessions.json"),
......
# -*- coding: utf-8 -*-
"""合同原件(台账「合同原件」列)端到端冒烟 —— **必须经 ui_verify_runner 沙箱跑**。
用法:python tools/ui_verify_runner.py tools/smoke_verify_contracts.py
覆盖(2026-10-15 新增功能):
1. 列模型:/api/tables/rent_parking 有 AL「合同原件」,type=files,在最后一列
2. 上传:POST /api/contracts(PDF data URL)→ 拿到 id / name / size
3. 查阅:GET /api/contracts/{id}?token= → 200 + application/pdf + 字节与原文件一致
4. 下载:GET /api/contracts/{id}?dl=1 → Content-Disposition: attachment
5. 落库:PUT /api/tables/rent_parking/cells 写 AL → 重读台账值正确
6. 导出:GET /api/tables/rent_parking/export → xlsx 里 AL 单元格是文件名(不是 JSON)
7. 清空:AL 写空串 → 存成 None
8. 权限:合同接口仅管理员(guest 上传 / 删 一律 403),但 guest 能只读台账看到合同
9. 拒绝:.exe 之类不在白名单 → 400
10. 安全:路径穿越 id → 404;删除后查阅 → 404
"""
import base64
import io
import json
import os
import sys
import urllib.error
import urllib.request
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")
BASE = os.environ["RENTOS_BASE"]
TOKEN = ""
FAILED = []
PDF = b"%PDF-1.4\n1 0 obj\n<< /Type /Catalog >>\nendobj\ntrailer\n%%EOF\n"
class ApiError(Exception):
def __init__(self, code, payload):
super().__init__("HTTP %s %s" % (code, payload))
self.code = code
self.payload = payload
def call(path, method="GET", body=None, token=None, raw=False):
req = urllib.request.Request(BASE + path, method=method)
req.add_header("Authorization", "Bearer " + (token or TOKEN))
data = None
if body is not None:
req.add_header("Content-Type", "application/json")
data = json.dumps(body).encode("utf-8")
try:
with urllib.request.urlopen(req, data, timeout=60) as r:
blob = r.read()
if raw:
return r.status, dict(r.headers), blob
return json.loads(blob.decode("utf-8"))
except urllib.error.HTTPError as e:
if raw:
return e.code, dict(e.headers), e.read()
raise ApiError(e.code, e.read().decode("utf-8")) from None
def check(name, cond, detail=""):
print((" PASS " if cond else " FAIL ") + name + (" " + str(detail) if detail else ""))
if not cond:
FAILED.append(name)
def login(username="admin", password="1989@rent05"):
req = urllib.request.Request(BASE + "/api/login", method="POST")
req.add_header("Content-Type", "application/json")
d = json.dumps({"username": username, "password": password}).encode("utf-8")
with urllib.request.urlopen(req, d, timeout=20) as r:
return json.loads(r.read().decode("utf-8"))["token"]
def main():
global TOKEN
TOKEN = login()
print("\n-- 1. 列模型 --")
t = call("/api/tables/rent_parking")
cols = t["table"]["columns"]
al = [c for c in cols if c["id"] == "AL"]
check("AL 列存在", len(al) == 1)
check("标签=合同原件 / type=files", al and al[0]["label"] == "合同原件" and al[0]["type"] == "files", al)
check("合同列排最后", cols[-1]["id"] == "AL", [c["id"] for c in cols][-3:])
rows = t["table"]["rows"]
check("行数 > 0", len(rows) > 0, len(rows))
row_id = rows[1]["_id"] # 用第 2 行(302室)测
check("该行原本没有合同", not rows[1].get("AL"), rows[1].get("AL"))
print("\n-- 2. 上传合同 --")
up = call("/api/contracts", "POST", {
"filename": "302租赁合同.pdf",
"data": "data:application/pdf;base64," + base64.b64encode(PDF).decode(),
})
cid = up["contract"]["id"]
check("上传成功并返回 id", bool(cid) and len(cid) == 32, cid)
check("保留原始文件名", up["contract"]["name"] == "302租赁合同.pdf", up["contract"]["name"])
check("大小正确", up["contract"]["size"] == len(PDF), up["contract"]["size"])
print("\n-- 3. 查阅 / 下载 --")
st, hd, blob = call("/api/contracts/%s?token=%s" % (cid, TOKEN), raw=True)
check("查阅 200", st == 200, st)
check("Content-Type=application/pdf", "application/pdf" in (hd.get("Content-Type") or ""),
hd.get("Content-Type"))
check("字节与上传一致", blob == PDF, len(blob))
check("内联预览(非 attachment)",
"attachment" not in (hd.get("Content-Disposition") or ""), hd.get("Content-Disposition"))
st2, hd2, _ = call("/api/contracts/%s?dl=1&token=%s" % (cid, TOKEN), raw=True)
check("下载 200", st2 == 200, st2)
check("下载是 attachment",
"attachment" in (hd2.get("Content-Disposition") or ""), hd2.get("Content-Disposition"))
print("\n-- 4. 挂到台账 AL 列 --")
val = json.dumps([{"id": cid, "name": "302租赁合同.pdf"}], ensure_ascii=False)
r = call("/api/tables/rent_parking/cells", "PUT",
{"changes": [{"row_id": row_id, "col": "AL", "value": val}],
"confirm": True, "who": "冒烟测试"})
check("写入返回 200", r.get("ok"), r)
t2 = call("/api/tables/rent_parking")
row2 = [x for x in t2["table"]["rows"] if x["_id"] == row_id][0]
got = json.loads(row2["AL"])
check("台账读回 id 正确", got[0]["id"] == cid, got)
check("台账读回 name 正确", got[0]["name"] == "302租赁合同.pdf", got)
print("\n-- 5. 导出 xlsx 里是文件名 --")
st3, _, xb = call("/api/tables/rent_parking/export", raw=True)
check("导出 200", st3 == 200, st3)
try:
import openpyxl
ws = openpyxl.load_workbook(io.BytesIO(xb)).active
col_i = [c["id"] for c in cols if not c.get("hide")].index("AL") + 1
r_i = [x["_id"] for x in t2["table"]["rows"]].index(row_id) + 5
v = ws.cell(r_i, col_i).value
check("AL 单元格是文件名", v == "302租赁合同.pdf", v)
except ImportError:
check("openpyxl 可用(跳过 xlsx 校验)", False, "venv 里应有 openpyxl")
print("\n-- 6. 权限:合同仅管理员 --")
gtoken = login("guest", "guest@2026")
try:
call("/api/contracts", "POST",
{"filename": "x.pdf", "data": base64.b64encode(PDF).decode()}, token=gtoken)
check("访客上传被拒 403", False, "竟然成功了")
except ApiError as e:
check("访客上传被拒 403", e.code == 403, e.code)
try:
call("/api/contracts/" + cid, "DELETE", token=gtoken)
check("访客删合同被拒 403", False, "竟然成功了")
except ApiError as e:
check("访客删合同被拒 403", e.code == 403, e.code)
gt = call("/api/tables/rent_parking", token=gtoken)
grow = [x for x in gt["table"]["rows"] if x["_id"] == row_id][0]
check("访客只读也能看到合同", json.loads(grow["AL"])[0]["id"] == cid, grow.get("AL"))
st4, _, _ = call("/api/contracts/%s?token=%s" % (cid, gtoken), raw=True)
check("访客能查阅合同", st4 == 200, st4)
print("\n-- 7. 拒绝非法类型 / 路径穿越 --")
try:
call("/api/contracts", "POST",
{"filename": "virus.exe", "data": base64.b64encode(b"MZ\x90\x00").decode()})
check("不支持的类型被拒", False, "竟然成功了")
except ApiError as e:
check("不支持的类型被拒 400", e.code == 400, e.code)
st5, _, _ = call("/api/contracts/../../etc/passwd?token=%s" % TOKEN, raw=True)
check("路径穿越 404", st5 == 404, st5)
print("\n-- 8. 清空 AL / 删除合同 --")
r2 = call("/api/tables/rent_parking/cells", "PUT",
{"changes": [{"row_id": row_id, "col": "AL", "value": ""}],
"confirm": True, "who": "冒烟测试"})
check("清空写入 200", r2.get("ok"), r2)
t3 = call("/api/tables/rent_parking")
row3 = [x for x in t3["table"]["rows"] if x["_id"] == row_id][0]
check("清空后为 None", row3.get("AL") is None, row3.get("AL"))
d = call("/api/contracts/" + cid, "DELETE")
check("删除合同 ok", d.get("ok"), d)
st6, _, _ = call("/api/contracts/%s?token=%s" % (cid, TOKEN), raw=True)
check("删除后查阅 404", st6 == 404, st6)
print("\n" + "=" * 60)
if FAILED:
print("失败 %d 项:%s" % (len(FAILED), "; ".join(FAILED)))
return 1
print("全部通过")
return 0
if __name__ == "__main__":
sys.exit(main())
# -*- coding: utf-8 -*-
"""
合同原件单元测试(2026-10-15)
==============================
对应需求(巴特菲):
房租车位费数据表需要支持上传合同留档、可查阅合同,增加「合同原件」列。
覆盖面:
· contracts._decode:PDF / Word / 图片按 data URL 或裸 base64 都能收;
不支持的扩展名(.exe)、空内容、超限、非法 base64 一律报错
· contracts 存取:save → find / path_of / list_all / remove;
clean_id 挡路径穿越;索引落盘为 JSON
· contracts.parse / dump / describe:台账列值的容错解析(JSON 数组 / 纯 id 串 / 脏数据 /
去重),dump 空列表返回空串,describe 补 url 与 exists
· tables:AL「合同原件」列存在且 type=files、不参与合计、coerce 把 list 转紧凑 JSON、
files_text 导出成文件名文本
· 台账端到端:update_cells 写 AL → load_table 读回 → export_xlsx 单元格是文件名
· 权限口径:contracts 与 attachments 目录互不影响
运行:python -m unittest tools.test_contracts -v
"""
import base64
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 config as C # noqa: E402
from rentos import contracts as CT # noqa: E402
from rentos import tables # noqa: E402
PDF = b"%PDF-1.4\n1 0 obj\n<< /Type /Catalog >>\nendobj\ntrailer\n%%EOF\n"
def b64(raw):
return base64.b64encode(raw).decode()
class Base(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.mkdtemp(prefix="rentos_ct_")
self.cdir = os.path.join(self.tmp, "contracts")
self.tdir = os.path.join(self.tmp, "tables")
os.makedirs(self.cdir, exist_ok=True)
os.makedirs(self.tdir, exist_ok=True)
self.p1 = mock.patch.object(C, "CONTRACT_DIR", self.cdir)
self.p2 = mock.patch.object(C, "TABLES_DIR", self.tdir)
self.p1.start()
self.p2.start()
def tearDown(self):
self.p1.stop()
self.p2.stop()
shutil.rmtree(self.tmp, ignore_errors=True)
class TestDecode(Base):
def test_data_url_pdf(self):
raw, mime = CT._decode("data:application/pdf;base64," + b64(PDF), "302租赁合同.pdf")
self.assertEqual(mime, "application/pdf")
self.assertEqual(raw, PDF)
def test_bare_base64_by_filename(self):
raw, mime = CT._decode(b64(PDF), "302租赁合同.pdf")
self.assertEqual(mime, "application/pdf")
self.assertEqual(raw, PDF)
def test_docx_and_image(self):
DOCX = ("application/vnd.openxmlformats-officedocument"
".wordprocessingml.document")
_, m = CT._decode(b64(b"PK\x03\x04xx"), "307&603房屋租赁合同.docx")
self.assertEqual(m, DOCX)
_, m2 = CT._decode(b64(b"\xff\xd8\xff\xe0xx"), "合同照片.jpg")
self.assertEqual(m2, "image/jpeg")
def test_reject_unknown_ext(self):
with self.assertRaises(ValueError):
CT._decode(b64(b"MZ\x90\x00"), "virus.exe")
def test_reject_empty_and_bad_base64(self):
with self.assertRaises(ValueError):
CT._decode(b64(b""), "302.pdf")
with self.assertRaises(ValueError):
CT._decode("!!!not-base64!!!", "302.pdf")
def test_reject_oversize(self):
with mock.patch.object(CT, "MAX_BYTES", 16):
with self.assertRaises(ValueError):
CT._decode(b64(PDF), "302.pdf")
class TestStore(Base):
def test_save_find_path_list_remove(self):
m = CT.save("data:application/pdf;base64," + b64(PDF), "302租赁合同.pdf", "admin")
self.assertEqual(len(m["id"]), 32)
self.assertEqual(m["name"], "302租赁合同.pdf")
self.assertEqual(m["size"], len(PDF))
self.assertEqual(m["by"], "admin")
self.assertTrue(os.path.isfile(os.path.join(self.cdir, m["id"] + ".pdf")))
# 索引落盘
with open(os.path.join(self.cdir, "index.json"), encoding="utf-8") as f:
self.assertEqual(json.load(f)[0]["id"], m["id"])
self.assertEqual(CT.find(m["id"])["name"], "302租赁合同.pdf")
self.assertTrue(CT.path_of(m["id"]).endswith(m["id"] + ".pdf"))
self.assertEqual(len(CT.list_all()), 1)
self.assertTrue(CT.remove(m["id"]))
self.assertEqual(CT.list_all(), [])
self.assertFalse(os.path.exists(os.path.join(self.cdir, m["id"] + ".pdf")))
self.assertFalse(CT.remove(m["id"]))
def test_clean_id_blocks_traversal(self):
for bad in ("../../etc/passwd", "..", "ZZZZ", "", None, "a" * 31):
self.assertEqual(CT.clean_id(bad), "")
self.assertIsNone(CT.find("../x"))
self.assertIsNone(CT.path_of("../../etc/passwd"))
self.assertFalse(CT.remove("../../etc/passwd"))
def test_safe_name_strips_path(self):
m = CT.save(b64(PDF), "C:\\Users\\x\\302租赁合同.pdf")
self.assertEqual(m["name"], "302租赁合同.pdf")
def test_missing_file_returns_none(self):
m = CT.save(b64(PDF), "302.pdf")
os.remove(os.path.join(self.cdir, m["id"] + ".pdf"))
self.assertIsNone(CT.path_of(m["id"]))
self.assertTrue(CT.find(m["id"])) # 索引里还在
class TestColumnValue(Base):
def setUp(self):
super().setUp()
self.a = CT.save(b64(PDF), "302租赁合同.pdf")
self.b = CT.save(b64(PDF), "302补充协议.pdf")
def test_dump_and_parse_roundtrip(self):
v = CT.dump([{"id": self.a["id"]}, {"id": self.b["id"], "name": "改名.pdf"}])
items = json.loads(v)
self.assertEqual(len(items), 2)
self.assertEqual(items[0]["name"], "302租赁合同.pdf") # 名字从索引补全
self.assertEqual(items[1]["name"], "改名.pdf")
back = CT.parse(v)
self.assertEqual([x["id"] for x in back], [self.a["id"], self.b["id"]])
def test_parse_tolerant(self):
self.assertEqual(CT.parse(None), [])
self.assertEqual(CT.parse(""), [])
self.assertEqual(CT.parse("[]"), [])
self.assertEqual(CT.parse("不是 JSON 也不算 id"), [])
self.assertEqual(CT.parse([{"id": "bad"}, {"nope": 1}, "也bad"]), [])
# 纯 id 串(手填)也认
self.assertEqual([x["id"] for x in CT.parse("%s, %s" % (self.a["id"], self.b["id"]))],
[self.a["id"], self.b["id"]])
# 去重
self.assertEqual(len(CT.parse([{"id": self.a["id"]}, {"id": self.a["id"]}])), 1)
def test_dump_empty(self):
self.assertEqual(CT.dump([]), "")
self.assertEqual(CT.dump([{"id": "xx"}]), "")
def test_describe(self):
d = CT.describe(CT.dump([{"id": self.a["id"]}, {"id": "0" * 32}]))
self.assertEqual(len(d), 2)
self.assertTrue(d[0]["exists"])
self.assertTrue(d[0]["url"].startswith("/api/contracts/"))
self.assertFalse(d[1]["exists"]) # 文件被删了也留个位置提示
class TestTableColumn(Base):
def test_al_column_registered(self):
col = [c for c in tables._RENT_PARKING_COLS if c["id"] == "AL"]
self.assertEqual(len(col), 1)
c = col[0]
self.assertEqual(c["label"], "合同原件")
self.assertEqual(c["type"], "files")
self.assertFalse(c["hide"])
self.assertFalse(c["sum"])
self.assertIn("files", tables.TYPES)
# 排在最右边(AJ「下一账单日」之后)
ids = [x["id"] for x in tables._RENT_PARKING_COLS]
self.assertEqual(ids[-1], "AL")
self.assertEqual(ids[-2], "AJ")
def test_coerce_files(self):
v = tables.coerce([{"id": "abc", "name": "x.pdf"}], "files")
self.assertEqual(json.loads(v), [{"id": "abc", "name": "x.pdf"}])
self.assertEqual(tables.coerce([], "files"), None)
self.assertEqual(tables.coerce("", "files"), None)
self.assertIsNone(tables.coerce(None, "files"))
# 字符串原样保留(前端已 JSON.stringify)
s = '[{"id":"a","name":"b"}]'
self.assertEqual(tables.coerce(s, "files"), s)
self.assertEqual(tables.coerce("[]", "files"), None)
# 非法项被剔除,id 转字符串
self.assertEqual(json.loads(tables.coerce([{"id": 5}, "x", 7], "files")),
[{"id": "5", "name": ""}, {"id": "x", "name": ""}])
def test_files_text(self):
v = '[{"id":"a","name":"302租赁合同.pdf"},{"id":"b","name":"补充协议.pdf"}]'
self.assertEqual(tables.files_text(v), "302租赁合同.pdf\n补充协议.pdf")
self.assertEqual(tables.files_text(None), "")
self.assertEqual(tables.files_text(""), "")
self.assertEqual(tables.files_text("手写的文本"), "手写的文本")
self.assertEqual(tables.files_text([{"id": "a"}]), "a")
class TestLedgerEndToEnd(Base):
"""写进台账 → 读回 → 导出 xlsx:整条链路。"""
def setUp(self):
super().setUp()
self.a = CT.save(b64(PDF), "302租赁合同.pdf")
obj = {"columns": [], "rows": [{"_id": "r1", "A": "精锐", "C": "302室"},
{"_id": "r2", "A": "科博", "C": "303室"}],
"next_seq": 3, "name": "房租车位费", "updated_at": "", "updated_by": ""}
with open(os.path.join(self.tdir, "rent_parking.json"), "w", encoding="utf-8") as f:
json.dump(obj, f, ensure_ascii=False)
def test_write_read_export(self):
val = CT.dump([{"id": self.a["id"], "name": self.a["name"]}])
res = tables.update_cells("rent_parking", [("r1", "AL", val)], who="测试")
self.assertEqual(res["updated"], 1)
t = tables.load_table("rent_parking")
row = [r for r in t["rows"] if r["_id"] == "r1"][0]
self.assertEqual(json.loads(row["AL"])[0]["id"], self.a["id"])
# 列定义以代码为准 → AL 自动出现在表头
self.assertIn("AL", [c["id"] for c in t["columns"]])
# 导出:单元格是文件名,不是 JSON
out = os.path.join(self.tmp, "out.xlsx")
tables.export_xlsx("rent_parking", out)
import openpyxl
ws = openpyxl.load_workbook(out).active
cols = [c["id"] for c in t["columns"] if not c.get("hide")]
col_i = cols.index("AL") + 1
self.assertEqual(ws.cell(5, col_i).value, "302租赁合同.pdf")
self.assertIsNone(ws.cell(6, col_i).value)
def test_clear_cell(self):
tables.update_cells("rent_parking",
[("r1", "AL", CT.dump([{"id": self.a["id"]}]))])
tables.update_cells("rent_parking", [("r1", "AL", "")])
t = tables.load_table("rent_parking")
self.assertIsNone([r for r in t["rows"] if r["_id"] == "r1"][0]["AL"])
class TestIsolation(Base):
def test_contract_dir_separate_from_attach(self):
"""合同与备注图片各用各的目录,互不干扰。"""
adir = os.path.join(self.tmp, "attachments")
os.makedirs(adir, exist_ok=True)
with mock.patch.object(C, "ATTACH_DIR", adir):
from rentos import attachments
attachments.save("data:image/png;base64," + b64(b"\x89PNG\r\n\x1a\nxx"), "a.png")
self.assertTrue(os.listdir(adir))
self.assertEqual(CT.list_all(), []) # 合同索引没被图片写脏
CT.save(b64(PDF), "302.pdf")
self.assertEqual(len(os.listdir(self.cdir)), 2) # index.json + 1 份合同
self.assertNotEqual(os.path.abspath(self.cdir), os.path.abspath(adir))
if __name__ == "__main__":
unittest.main(verbosity=2)
......@@ -403,12 +403,12 @@ class TestRentParkingTable(unittest.TestCase):
spec = tables.list_tables()[0]
self.assertEqual(spec["name"], "房租车位费")
self.assertTrue(spec["ready"])
self.assertEqual(spec["col_count"], 31) # 29 出单列 + 1 免租类型 + 1 推送排期
self.assertEqual(spec["col_count"], 32) # 29 出单列 + 1 免租类型 + 1 推送排期 + 1 合同原件
def test_no_hidden_column(self):
t = self._t()
self.assertEqual([c["id"] for c in t["columns"] if c.get("hide")], [])
self.assertEqual(len(t["columns"]), 31)
self.assertEqual(len(t["columns"]), 32)
self.assertNotIn("序号", [c["label"] for c in t["columns"]])
def test_column_layout(self):
......@@ -472,7 +472,8 @@ class TestRentParkingTable(unittest.TestCase):
self.assertEqual(cmap["AJ"]["type"], "date")
self.assertEqual(cmap["AJ"]["group"], "推送")
self.assertEqual(tables.BILL_COLS["next_bill"], "AJ")
self.assertEqual([c["id"] for c in t["columns"]][-1], "AJ", "推送列应是最后一列")
self.assertEqual([c["id"] for c in t["columns"]][-2], "AJ", "推送列紧邻最右的合同列")
self.assertEqual([c["id"] for c in t["columns"]][-1], "AL", "合同原件列排最后")
def test_retired_cols_not_used_by_billing(self):
"""已删除的房产/房产税列不得再被出单逻辑(BILL_COLS)引用。"""
......
......@@ -63,6 +63,7 @@ def start_sandbox():
"TABLES_DIR": os.path.join(data, "tables"),
"ATTACH_DIR": os.path.join(data, "attachments"),
"DRAFT_DIR": os.path.join(data, "drafts"),
"CONTRACT_DIR": os.path.join(data, "contracts"),
"CONFIG_FILE": os.path.join(data, "config.json"),
"SEND_LOG_FILE": os.path.join(data, "send_log.json"),
"SESSION_FILE": os.path.join(data, "sessions.json"),
......
......@@ -57,6 +57,7 @@ def start_sandbox():
"TABLES_DIR": os.path.join(data, "tables"),
"ATTACH_DIR": os.path.join(data, "attachments"),
"DRAFT_DIR": os.path.join(data, "drafts"),
"CONTRACT_DIR": os.path.join(data, "contracts"),
"CONFIG_FILE": os.path.join(data, "config.json"),
"SEND_LOG_FILE": os.path.join(data, "send_log.json"),
"SESSION_FILE": os.path.join(data, "sessions.json"),
......
# -*- coding: utf-8 -*-
"""
「合同原件」列(上传 / 查阅 / 移除 + 台账联动)端到端 UI 验证(2026-10-15)
=========================================================================
对应需求(巴特菲):
房租车位费数据表需要支持上传合同留档、可查阅合同,增加合同原件的列。
做法:**自带沙箱**(复制 data/ + 改写路径常量 + 另起端口),真浏览器跑一遍。
真实 data/ 零改动,沙箱里上传的合同也落在沙箱 data/contracts/。
覆盖:
1. 表头出现「合同原件」且在最后一列;只读态不显示上传入口
2. 编辑模式 → 点「+ 合同」弹出浮层
3. 上传 PDF(走真实文件选择器)→ 浮层里出现文件与「查阅 / 下载 / 移除」
4. 查阅链接带 token、指向 /api/contracts/{id},请求返回 200 + application/pdf
5. 关闭浮层:格子变「📎 1 份」并进入「待保存」高亮
6. 保存 → 台账 AL 落盘、待保存条消失
7. 只读态:按钮仍可点开查阅,但浮层里没有「上传合同 / 移除」
8. 移除 → 只解除关联(台账 AL 清空),服务器上文件仍在
9. 全程无 console / pageerror
运行:python tools/ui_verify12.py
"""
import io
import os
import shutil
import sys
import tempfile
import threading
from http.server import ThreadingHTTPServer
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, ROOT)
os.environ["no_proxy"] = "*"
os.environ["NO_PROXY"] = "*"
from rentos import config as C # noqa: E402
from rentos import server as S # noqa: E402
PORT = int(os.environ.get("RENTOS_VERIFY_PORT", "8908"))
BASE = "http://127.0.0.1:%d" % PORT
OUT = os.path.join(ROOT, "output", "ui_shots")
PDF_NAME = "302租赁合同.pdf"
PDF = b"%PDF-1.4\n1 0 obj\n<< /Type /Catalog >>\nendobj\ntrailer\n%%EOF\n"
OK, FAIL = [], []
def ck(name, cond, extra=""):
(OK if cond else FAIL).append(name)
print(" [%s] %s%s" % ("ok" if cond else "FAIL", name,
(" → %s" % extra) if extra != "" else ""))
def start_sandbox():
tmp = tempfile.mkdtemp(prefix="rentos-verify12-")
shutil.copytree(os.path.join(ROOT, "data"), os.path.join(tmp, "data"))
data = os.path.join(tmp, "data")
for name, val in {
"DATA_DIR": data,
"BATCH_DIR": os.path.join(data, "batches"),
"TABLES_DIR": os.path.join(data, "tables"),
"ATTACH_DIR": os.path.join(data, "attachments"),
"DRAFT_DIR": os.path.join(data, "drafts"),
"CONTRACT_DIR": os.path.join(data, "contracts"),
"CONFIG_FILE": os.path.join(data, "config.json"),
"SEND_LOG_FILE": os.path.join(data, "send_log.json"),
"SESSION_FILE": os.path.join(data, "sessions.json"),
"UNIT_MAP_FILE": os.path.join(data, "unit_map.json"),
"UPLOAD_DIR": os.path.join(tmp, "uploads"),
"OUTPUT_DIR": os.path.join(tmp, "output", "elec"),
"OUTPUT_RENT_DIR": os.path.join(tmp, "output", "rent"),
"TABLE_EXPORT_DIR": os.path.join(tmp, "output", "tables"),
"BATCH_EXPORT_DIR": os.path.join(tmp, "output", "exports"),
"LOG_DIR": os.path.join(tmp, "logs"),
}.items():
setattr(C, name, val)
C.ensure_dirs()
srv = ThreadingHTTPServer(("127.0.0.1", PORT), S.Handler)
srv.daemon_threads = True
threading.Thread(target=srv.serve_forever, daemon=True).start()
return srv, tmp
CELL = "#tbWrap tbody tr:nth-child(1) td.files"
def main():
os.makedirs(OUT, exist_ok=True)
srv, tmp = start_sandbox()
pdf = os.path.join(tmp, PDF_NAME)
with open(pdf, "wb") as f:
f.write(PDF)
print("沙箱数据目录:%s" % tmp)
print("沙箱服务:%s(真实 data/ 未被触碰)" % BASE)
from playwright.sync_api import sync_playwright
errors = []
try:
with sync_playwright() as pw:
b = pw.chromium.launch()
p = b.new_page(viewport={"width": 1760, "height": 1000})
p.on("console", lambda m: errors.append("console:" + m.text)
if (m.type == "error" and "favicon" not in m.text.lower()
and "Failed to load resource" not in m.text) else None)
p.on("pageerror", lambda e: errors.append("pageerror:" + str(e)))
p.on("dialog", lambda d: d.accept())
def js(fn, *a):
return p.evaluate(fn, *a) if a else p.evaluate(fn)
def goto(h):
p.evaluate("(h) => { location.hash = '#/' + h; }", h)
p.wait_for_timeout(1300)
# ---- 登录
p.goto(BASE + "/login.html")
p.fill("#u", "admin")
p.fill("#p", "1989@rent05")
p.click("button[type=submit]")
p.wait_for_url("**/admin.html")
p.wait_for_function("() => location.hash.length > 2", timeout=20000)
print("\n-- 1. 列表头 / 只读态 --")
goto("table/rent_parking")
p.wait_for_selector("#tbWrap td.files", timeout=20000)
heads = js("""() => [...document.querySelectorAll('#tbWrap thead tr:nth-child(1) th')]
.map(e => e.textContent.trim()).filter(Boolean)""")
ck("表头出现「合同原件」", "合同原件" in heads, heads[-4:])
ck("合同原件紧邻最右(其右只余「操作」列)",
heads[-2] == "合同原件" and heads[-1] == "操作", heads[-2:])
unit0 = js("() => document.querySelector('#tbWrap tbody tr:nth-child(1) td:nth-child(4)').textContent.trim()")
ck("第一行取到单元号", unit0.endswith("室") or unit0.endswith("车位"), unit0)
ck("只读态不显示上传入口",
js("() => !document.querySelector('%s').querySelector('button')" % CELL))
print("\n-- 2. 编辑模式 · 上传合同 --")
p.check("#tbEdit")
p.wait_for_timeout(400)
ck("编辑态出现「+ 合同」按钮",
"合同" in js("() => (document.querySelector('%s button.fbtn')||{}).textContent || ''" % CELL))
p.click(CELL + " button.fbtn")
p.wait_for_selector("#modal.show", timeout=8000)
title = p.inner_text("#modalTitle")
ck("浮层标题带单元", "合同原件" in title and unit0[:3] in title, title)
ck("空态提示", "还没有挂合同原件" in p.inner_text("#modalBody"))
ck("有「上传合同」按钮", p.query_selector("#fdlg-up") is not None)
with p.expect_file_chooser() as fc:
p.click("#fdlg-up")
fc.value.set_files(pdf)
p.wait_for_function(
"() => [...document.querySelectorAll('#modalBody tbody tr')]"
".some(tr => tr.textContent.includes('%s'))" % PDF_NAME, timeout=20000)
ck("浮层出现刚上传的文件", True)
body_txt = p.inner_text("#modalBody")
ck("浮层提供查阅 / 下载 / 移除",
all(t in body_txt for t in ("查阅", "下载", "移除")))
href = js("() => document.querySelector('#modalBody tbody tr td a').href")
ck("查阅链接指向 /api/contracts/{id} 且带 token",
"/api/contracts/" in href and "token=" in href, href[:80])
r = p.request.get(href)
ck("查阅请求 200 + application/pdf",
r.status == 200 and "application/pdf" in (r.headers.get("content-type") or ""),
"%s %s" % (r.status, r.headers.get("content-type")))
ck("查阅返回的就是原文件字节", r.body() == PDF, len(r.body()))
p.screenshot(path=os.path.join(OUT, "verify12_1_dialog.png"))
print("\n-- 3. 关浮层 · 待保存 · 保存 --")
p.evaluate("closeModal()")
p.wait_for_timeout(300)
cell_txt = p.inner_text(CELL)
ck("格子显示「1 份」", "1 份" in cell_txt, cell_txt)
ck("格子进入待保存高亮",
"pending" in (p.get_attribute(CELL, "class") or ""),
p.get_attribute(CELL, "class"))
p.click("#btn-tb-save-all")
p.wait_for_selector("#cfm.show", timeout=8000)
ck("确认框列出「合同原件」改动", "合同原件" in p.inner_text("#cfmBody"))
p.click("#cfmOk")
p.wait_for_function(
"() => !(document.querySelector('#tbPendBar')||{}).textContent"
".includes('待保存')", timeout=20000)
ck("保存后待保存条消失", True)
ck("保存后高亮清除", "pending" not in (p.get_attribute(CELL, "class") or ""))
print("\n-- 4. 服务端落库 --")
st = js("""async () => {
const t = localStorage.getItem('rentos_token');
const r = await fetch('/api/tables/rent_parking',
{headers: {Authorization: 'Bearer ' + t}});
const d = await r.json();
return d.table.rows[0].AL || '';
}""")
ck("台账 AL 已落盘", PDF_NAME in str(st), str(st)[:90])
ctx = js("""async () => {
const t = localStorage.getItem('rentos_token');
const r = await fetch('/api/contracts', {headers: {Authorization: 'Bearer ' + t}});
const d = await r.json();
return d.contracts.length;
}""")
ck("合同索引里有 1 份", ctx == 1, ctx)
print("\n-- 5. 只读态查阅 --")
p.uncheck("#tbEdit")
p.wait_for_timeout(400)
ck("只读态仍显示「1 份」按钮", "1 份" in p.inner_text(CELL))
p.click(CELL + " button.fbtn")
p.wait_for_selector("#modal.show", timeout=8000)
ro = p.inner_text("#modalBody")
ck("只读浮层能查阅 / 下载", "查阅" in ro and "下载" in ro)
ck("只读浮层无「上传合同」", "上传合同" not in ro)
ck("只读浮层无移除按钮", p.query_selector("#modalBody [data-fdel]") is None)
ck("只读浮层无「移除」字样", "移除" not in ro, ro[:60])
p.evaluate("closeModal()")
print("\n-- 6. 移除只解关联 --")
p.check("#tbEdit")
p.wait_for_timeout(400)
p.click(CELL + " button.fbtn")
p.wait_for_selector("#modal.show", timeout=8000)
p.click("#modalBody [data-fdel]") # 原生 confirm 已被 accept
p.wait_for_timeout(500)
ck("移除后浮层回到空态", "还没有挂合同原件" in p.inner_text("#modalBody"))
p.evaluate("closeModal()")
p.wait_for_timeout(200)
ck("格子回到「+ 合同」", "合同" in p.inner_text(CELL) and "1 份" not in p.inner_text(CELL))
p.click("#btn-tb-save-all")
p.wait_for_selector("#cfm.show", timeout=8000)
p.click("#cfmOk")
p.wait_for_function(
"() => !(document.querySelector('#tbPendBar')||{}).textContent"
".includes('待保存')", timeout=20000)
st2 = js("""async () => {
const t = localStorage.getItem('rentos_token');
const r = await fetch('/api/tables/rent_parking',
{headers: {Authorization: 'Bearer ' + t}});
const d = await r.json();
return d.table.rows[0].AL || '';
}""")
ck("台账 AL 已清空", not st2, repr(st2))
left = [f for f in os.listdir(os.path.join(tmp, "data", "contracts"))
if f != "index.json"]
ck("服务器上文件仍保留(移除只解关联)", len(left) == 1, left)
print("\n-- 7. 控制台 --")
ck("无 console / pageerror", not errors, errors[:3])
p.screenshot(path=os.path.join(OUT, "verify12_2_grid.png"))
b.close()
finally:
srv.shutdown()
shutil.rmtree(tmp, ignore_errors=True)
print("\n" + "=" * 68)
print("通过 %d 项 / 失败 %d 项" % (len(OK), len(FAIL)))
if FAIL:
print("失败:%s" % "; ".join(FAIL))
return 1 if FAIL else 0
if __name__ == "__main__":
sys.exit(main())
......@@ -57,6 +57,7 @@ def start_sandbox():
"TABLES_DIR": os.path.join(data, "tables"),
"ATTACH_DIR": os.path.join(data, "attachments"),
"DRAFT_DIR": os.path.join(data, "drafts"),
"CONTRACT_DIR": os.path.join(data, "contracts"),
"CONFIG_FILE": os.path.join(data, "config.json"),
"SEND_LOG_FILE": os.path.join(data, "send_log.json"),
"SESSION_FILE": os.path.join(data, "sessions.json"),
......
......@@ -59,6 +59,7 @@ def start_sandbox():
"TABLES_DIR": os.path.join(data, "tables"),
"ATTACH_DIR": os.path.join(data, "attachments"),
"DRAFT_DIR": os.path.join(data, "drafts"),
"CONTRACT_DIR": os.path.join(data, "contracts"),
"CONFIG_FILE": os.path.join(data, "config.json"),
"SEND_LOG_FILE": os.path.join(data, "send_log.json"),
"SESSION_FILE": os.path.join(data, "sessions.json"),
......
......@@ -55,6 +55,7 @@ def start_sandbox():
"TABLES_DIR": os.path.join(data, "tables"),
"ATTACH_DIR": os.path.join(data, "attachments"),
"DRAFT_DIR": os.path.join(data, "drafts"),
"CONTRACT_DIR": os.path.join(data, "contracts"),
"CONFIG_FILE": os.path.join(data, "config.json"),
"SEND_LOG_FILE": os.path.join(data, "send_log.json"),
"SESSION_FILE": os.path.join(data, "sessions.json"),
......
......@@ -46,6 +46,7 @@ def start_sandbox():
"TABLES_DIR": os.path.join(data, "tables"),
"ATTACH_DIR": os.path.join(data, "attachments"),
"DRAFT_DIR": os.path.join(data, "drafts"), # 电费录入草稿(临时保存)
"CONTRACT_DIR": os.path.join(data, "contracts"), # 台账「合同原件」
"CONFIG_FILE": os.path.join(data, "config.json"),
"SEND_LOG_FILE": os.path.join(data, "send_log.json"),
"SESSION_FILE": os.path.join(data, "sessions.json"),
......
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