fe6b4e59ec
将 openModal/closeModal 提升为顶层唯一定义，公司端启动即加载断档提醒， 并在 await 前提取表单引用；补充 Playwright 两端冒烟与源码契约测试。 Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: multica-agent <github@multica.ai>
519 lines
20 KiB
Python
519 lines
20 KiB
Python
"""HEL-203: 真实浏览器冒烟——原因弹窗、首屏断档、表单 reset。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import re
|
||
import tempfile
|
||
import threading
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
from bank_importer import auth, calculation, matching, master_data
|
||
from bank_importer.db import connect, migrate, utc_now
|
||
|
||
import server
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
WEB = ROOT / "web"
|
||
ADMIN_PASSWORD = "AdminPass123"
|
||
CASHIER_PASSWORD = "CashierA123"
|
||
|
||
try:
|
||
from playwright.sync_api import sync_playwright
|
||
except ImportError: # pragma: no cover
|
||
sync_playwright = None
|
||
|
||
|
||
def _prepare_chrome_libs() -> Path | None:
|
||
"""本机缺系统 atk 时，复用仓库旁的本地 chromium 依赖目录。"""
|
||
candidates = [ROOT / ".chrome-libs" / "lib"]
|
||
for lib_dir in candidates:
|
||
if (lib_dir / "libatk-1.0.so.0").exists():
|
||
current = os.environ.get("LD_LIBRARY_PATH", "")
|
||
prefix = str(lib_dir)
|
||
if prefix not in current.split(":"):
|
||
os.environ["LD_LIBRARY_PATH"] = (
|
||
f"{prefix}:{current}" if current else prefix
|
||
)
|
||
return lib_dir
|
||
return None
|
||
|
||
|
||
def _chromium_available() -> bool:
|
||
if not sync_playwright:
|
||
return False
|
||
_prepare_chrome_libs()
|
||
try:
|
||
with sync_playwright() as p:
|
||
browser = p.chromium.launch(headless=True, args=["--no-sandbox"])
|
||
browser.close()
|
||
return True
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
class Hel203SourceContractTests(unittest.TestCase):
|
||
"""不依赖浏览器：锁住 N1～N3 的源码契约。"""
|
||
|
||
def test_open_modal_is_single_top_level(self) -> None:
|
||
js = (WEB / "app.js").read_text(encoding="utf-8")
|
||
defs = list(re.finditer(r"(?m)^function openModal\(", js))
|
||
self.assertEqual(1, len(defs), "openModal 必须只有一处顶层定义")
|
||
# 不得再出现在 initAdmin / initCompany 函数体内的局部副本
|
||
self.assertNotRegex(
|
||
js,
|
||
r"function initAdmin\(\)[\s\S]*?function openModal\(",
|
||
)
|
||
self.assertNotRegex(
|
||
js,
|
||
r"function initCompany\(\)[\s\S]*?function openModal\(",
|
||
)
|
||
self.assertIn("function askReason(", js)
|
||
ask_pos = js.index("function askReason(")
|
||
open_pos = defs[0].start()
|
||
self.assertLess(open_pos, ask_pos, "openModal 须在 askReason 之前定义")
|
||
|
||
def test_init_company_boots_coverage_gaps(self) -> None:
|
||
js = (WEB / "app.js").read_text(encoding="utf-8")
|
||
company_fn = js[js.index("function initCompany(") :]
|
||
boot = company_fn[: company_fn.index("\nif (portal ===")]
|
||
self.assertIn("await loadCompanyWorkspace()", boot)
|
||
self.assertIn("await loadCompanyCoverageGaps()", boot)
|
||
|
||
def test_async_forms_capture_form_before_await(self) -> None:
|
||
js = (WEB / "app.js").read_text(encoding="utf-8")
|
||
self.assertNotIn("event.currentTarget.reset()", js)
|
||
for marker in ("#companyForm", "#openingForm", "#accountForm"):
|
||
idx = js.index(marker)
|
||
chunk = js[idx : idx + 2500]
|
||
self.assertIn("const form = event.currentTarget", chunk)
|
||
self.assertIn("form.reset()", chunk)
|
||
|
||
|
||
@unittest.skipUnless(sync_playwright, "playwright 未安装，跳过浏览器冒烟")
|
||
@unittest.skipUnless(_chromium_available(), "chromium 无法启动，跳过浏览器冒烟")
|
||
class Hel203BrowserSmokeTests(unittest.TestCase):
|
||
"""真实 Chromium：改起算日→期初→公司端期末；断档首屏→说明→审核。"""
|
||
|
||
@classmethod
|
||
def setUpClass(cls) -> None:
|
||
_prepare_chrome_libs()
|
||
cls.temp_dir = tempfile.TemporaryDirectory()
|
||
root = Path(cls.temp_dir.name)
|
||
cls.db_path = root / "app.db"
|
||
cls.storage = root / "files"
|
||
cls.storage.mkdir()
|
||
|
||
cls._old_db = server.DB_PATH
|
||
cls._old_storage = server.STORAGE_DIR
|
||
server.DB_PATH = cls.db_path
|
||
server.STORAGE_DIR = cls.storage
|
||
|
||
connection = connect(cls.db_path)
|
||
migrate(connection)
|
||
auth.create_user(
|
||
connection,
|
||
"group-admin",
|
||
ADMIN_PASSWORD,
|
||
"admin",
|
||
must_change_password=False,
|
||
)
|
||
cls.company_a = master_data.create_company(
|
||
connection, "甲公司", None, None, None
|
||
)
|
||
cls.company_b = master_data.create_company(
|
||
connection, "乙公司", None, None, None
|
||
)
|
||
auth.create_user(
|
||
connection,
|
||
"cashier-a",
|
||
CASHIER_PASSWORD,
|
||
"company",
|
||
cls.company_a,
|
||
must_change_password=False,
|
||
)
|
||
admin = connection.execute(
|
||
"SELECT * FROM users WHERE username = 'group-admin'"
|
||
).fetchone()
|
||
account = master_data.submit_bank_account(
|
||
connection,
|
||
company_id=cls.company_a,
|
||
bank_name="中信银行",
|
||
account_type="基本户",
|
||
account_number="6222000000000001",
|
||
start_date="2026-06-01",
|
||
actor=None,
|
||
)
|
||
cls.account_a = master_data.review_bank_account(
|
||
connection,
|
||
account["id"],
|
||
"approve",
|
||
None,
|
||
admin,
|
||
effective_from="2026-06-01",
|
||
)
|
||
account_b = master_data.submit_bank_account(
|
||
connection,
|
||
company_id=cls.company_b,
|
||
bank_name="中信银行",
|
||
account_type="基本户",
|
||
account_number="6222000000000002",
|
||
start_date="2026-06-01",
|
||
actor=None,
|
||
)
|
||
master_data.review_bank_account(
|
||
connection,
|
||
account_b["id"],
|
||
"approve",
|
||
None,
|
||
admin,
|
||
effective_from="2026-06-01",
|
||
)
|
||
# 制造一处 mid 断档，供公司端首屏提醒
|
||
with connection:
|
||
cursor = connection.execute(
|
||
"""
|
||
INSERT INTO source_files
|
||
(sha256, original_filename, size_bytes, storage_path, created_at)
|
||
VALUES (?, 'gap.xlsx', 1, 'data/files/gap.xlsx', ?)
|
||
""",
|
||
("sha-hel203-gap", utc_now()),
|
||
)
|
||
source_file_id = int(cursor.lastrowid)
|
||
cursor = connection.execute(
|
||
"""
|
||
INSERT INTO import_batches (
|
||
source_file_id, status, company_id, upload_bank_account_id,
|
||
created_at, updated_at
|
||
) VALUES (?, 'parsed', ?, ?, ?, ?)
|
||
""",
|
||
(
|
||
source_file_id,
|
||
cls.company_a,
|
||
cls.account_a["id"],
|
||
utc_now(),
|
||
utc_now(),
|
||
),
|
||
)
|
||
batch_id = int(cursor.lastrowid)
|
||
cursor = connection.execute(
|
||
"""
|
||
INSERT INTO sheet_batches (
|
||
import_batch_id, sheet_name, bank_name, template_id,
|
||
template_version, header_row, transaction_count, warnings,
|
||
created_at
|
||
) VALUES (?, '流水', '测试银行', 'test-v1', 1, 1, 1, '[]', ?)
|
||
""",
|
||
(batch_id, utc_now()),
|
||
)
|
||
sheet_batch_id = int(cursor.lastrowid)
|
||
connection.execute(
|
||
"""
|
||
INSERT INTO sheet_reviews (
|
||
import_batch_id, sheet_name, outcome, sheet_batch_id,
|
||
review_status, created_at
|
||
) VALUES (?, '流水', 'parsed', ?, 'confirmed', ?)
|
||
""",
|
||
(batch_id, sheet_batch_id, utc_now()),
|
||
)
|
||
connection.execute(
|
||
"""
|
||
INSERT INTO source_rows (
|
||
sheet_batch_id, source_row, transaction_at, income, expense,
|
||
own_account, created_at
|
||
) VALUES (?, 1, '2026-06-21T10:00:00', '0', '0', ?, ?)
|
||
""",
|
||
(sheet_batch_id, "6222000000000001", utc_now()),
|
||
)
|
||
calculation.set_calculation_start_date(
|
||
connection, "2026-06-01", "初始化起算", admin
|
||
)
|
||
calculation.recalculate_coverage_gaps(connection)
|
||
|
||
# 种一笔已确认往来，确认期初后公司端才能进入完整期末口径
|
||
def _add_row(company_id, account_id, own, cp, *, income, expense, at, ref):
|
||
with connection:
|
||
cursor = connection.execute(
|
||
"""
|
||
INSERT INTO source_files
|
||
(sha256, original_filename, size_bytes, storage_path, created_at)
|
||
VALUES (?, 'xfer.xlsx', 1, 'data/files/xfer.xlsx', ?)
|
||
""",
|
||
(ref, utc_now()),
|
||
)
|
||
source_file_id = int(cursor.lastrowid)
|
||
cursor = connection.execute(
|
||
"""
|
||
INSERT INTO import_batches (
|
||
source_file_id, status, company_id, upload_bank_account_id,
|
||
created_at, updated_at
|
||
) VALUES (?, 'parsed', ?, ?, ?, ?)
|
||
""",
|
||
(source_file_id, company_id, account_id, utc_now(), utc_now()),
|
||
)
|
||
batch_id = int(cursor.lastrowid)
|
||
cursor = connection.execute(
|
||
"""
|
||
INSERT INTO sheet_batches (
|
||
import_batch_id, sheet_name, bank_name, template_id,
|
||
template_version, header_row, transaction_count, warnings,
|
||
created_at
|
||
) VALUES (?, '流水', '测试银行', 'test-v1', 1, 1, 1, '[]', ?)
|
||
""",
|
||
(batch_id, utc_now()),
|
||
)
|
||
sheet_batch_id = int(cursor.lastrowid)
|
||
connection.execute(
|
||
"""
|
||
INSERT INTO sheet_reviews (
|
||
import_batch_id, sheet_name, outcome, sheet_batch_id,
|
||
review_status, created_at
|
||
) VALUES (?, '流水', 'parsed', ?, 'confirmed', ?)
|
||
""",
|
||
(batch_id, sheet_batch_id, utc_now()),
|
||
)
|
||
cursor = connection.execute(
|
||
"""
|
||
INSERT INTO source_rows (
|
||
sheet_batch_id, source_row, transaction_at, income, expense,
|
||
own_account, own_name, counterparty_account, counterparty_name,
|
||
summary, purpose, currency, created_at
|
||
) VALUES (?, 1, ?, ?, ?, ?, '测试', ?, '对方', '往来', '往来款', 'CNY', ?)
|
||
""",
|
||
(
|
||
sheet_batch_id,
|
||
at,
|
||
income,
|
||
expense,
|
||
own,
|
||
cp,
|
||
utc_now(),
|
||
),
|
||
)
|
||
return int(cursor.lastrowid)
|
||
|
||
row_a = _add_row(
|
||
cls.company_a,
|
||
cls.account_a["id"],
|
||
"6222000000000001",
|
||
"6222000000000002",
|
||
income="0",
|
||
expense="100.00",
|
||
at="2026-06-20T10:00:00",
|
||
ref="sha-hel203-a",
|
||
)
|
||
row_b = _add_row(
|
||
cls.company_b,
|
||
account_b["id"],
|
||
"6222000000000002",
|
||
"6222000000000001",
|
||
income="100.00",
|
||
expense="0",
|
||
at="2026-06-20T11:00:00",
|
||
ref="sha-hel203-b",
|
||
)
|
||
matching.reconcile_rows(connection, [row_a, row_b])
|
||
connection.close()
|
||
|
||
class QuietHandler(server.AppHandler):
|
||
def log_message(self, *args) -> None:
|
||
pass
|
||
|
||
cls.httpd = server.ThreadingHTTPServer(("127.0.0.1", 0), QuietHandler)
|
||
cls.port = cls.httpd.server_address[1]
|
||
cls.thread = threading.Thread(target=cls.httpd.serve_forever, daemon=True)
|
||
cls.thread.start()
|
||
cls.base = f"http://127.0.0.1:{cls.port}"
|
||
|
||
@classmethod
|
||
def tearDownClass(cls) -> None:
|
||
cls.httpd.shutdown()
|
||
cls.httpd.server_close()
|
||
server.DB_PATH = cls._old_db
|
||
server.STORAGE_DIR = cls._old_storage
|
||
cls.temp_dir.cleanup()
|
||
|
||
def _new_page(self, playwright):
|
||
browser = playwright.chromium.launch(
|
||
headless=True, args=["--no-sandbox", "--disable-dev-shm-usage"]
|
||
)
|
||
context = browser.new_context(viewport={"width": 1440, "height": 900})
|
||
page = context.new_page()
|
||
errors: list[str] = []
|
||
page.on("pageerror", lambda err: errors.append(str(err)))
|
||
page.on(
|
||
"console",
|
||
lambda msg: errors.append(f"console.{msg.type}: {msg.text}")
|
||
if msg.type == "error"
|
||
else None,
|
||
)
|
||
return browser, page, errors
|
||
|
||
def _login(self, page, *, portal: str, username: str, password: str) -> None:
|
||
login_path = "login-admin.html" if portal == "admin" else "login-company.html"
|
||
page.goto(f"{self.base}/{login_path}", wait_until="domcontentloaded")
|
||
page.fill("#account", username)
|
||
page.fill("#password", password)
|
||
page.click('button[type="submit"]')
|
||
expect = "admin.html" if portal == "admin" else "company.html"
|
||
page.wait_for_url(f"**/{expect}", timeout=15000)
|
||
|
||
def test_02_admin_start_date_opening_company_ending(self) -> None:
|
||
with sync_playwright() as p:
|
||
browser, page, errors = self._new_page(p)
|
||
try:
|
||
self._login(
|
||
page,
|
||
portal="admin",
|
||
username="group-admin",
|
||
password=ADMIN_PASSWORD,
|
||
)
|
||
page.click('a[data-view="settings"]')
|
||
page.wait_for_selector("#cs-start", state="visible")
|
||
|
||
# 修改起算日 → 原因弹窗必须打开且发出 PUT
|
||
page.fill("#cs-start", "2026-06-15")
|
||
with page.expect_request(
|
||
lambda req: req.method == "PUT"
|
||
and "/api/admin/settings/calculation-start" in req.url
|
||
) as start_req:
|
||
page.click('#systemSettings button[type="submit"]')
|
||
page.wait_for_selector("#reasonDialog.open", timeout=5000)
|
||
page.fill("#reasonInput", "调整起算日供冒烟")
|
||
page.click("#reasonSubmit")
|
||
self.assertTrue(start_req.value.post_data)
|
||
page.wait_for_function(
|
||
"() => document.getElementById('cs-start')?.value === '2026-06-15'"
|
||
)
|
||
|
||
# 创建期初并确认
|
||
page.click("#openOpeningDialog")
|
||
page.wait_for_selector("#openingDialog.open")
|
||
page.select_option("#ob-from", label="甲公司")
|
||
page.select_option("#ob-to", label="乙公司")
|
||
page.fill("#ob-amount", "200")
|
||
page.fill("#ob-reason", "冒烟期初录入")
|
||
with page.expect_request(
|
||
lambda req: req.method == "POST"
|
||
and req.url.endswith("/api/admin/opening-balances")
|
||
):
|
||
page.click('#openingForm button[type="submit"]')
|
||
page.wait_for_selector(
|
||
'#openingRows button[data-confirm-opening]',
|
||
timeout=8000,
|
||
)
|
||
|
||
with page.expect_request(
|
||
lambda req: req.method == "POST"
|
||
and "/opening-balances/" in req.url
|
||
and req.url.endswith("/confirm")
|
||
):
|
||
page.click('#openingRows button[data-confirm-opening]')
|
||
page.wait_for_selector("#reasonDialog.open", timeout=5000)
|
||
page.fill("#reasonInput", "确认期初冒烟")
|
||
page.click("#reasonSubmit")
|
||
page.wait_for_selector(
|
||
'#openingRows button[data-void-opening]',
|
||
timeout=8000,
|
||
)
|
||
|
||
# 公司端看到完整期末口径
|
||
self._login(
|
||
page,
|
||
portal="company",
|
||
username="cashier-a",
|
||
password=CASHIER_PASSWORD,
|
||
)
|
||
page.click('a[data-view="transfers"]')
|
||
page.wait_for_function(
|
||
"""() => {
|
||
const data = document.getElementById('transfersData');
|
||
const empty = document.getElementById('transfersEmpty');
|
||
const ready = (data && !data.hidden) || (empty && !empty.hidden);
|
||
const card = document.getElementById('tfStatEndingCard');
|
||
const title = document.getElementById('tfStatNetTitle');
|
||
const emptyHtml = document.getElementById('transfersEmptyStats')?.innerHTML || '';
|
||
return ready && (
|
||
(card && !card.hidden) ||
|
||
(title && title.textContent.includes('期末')) ||
|
||
emptyHtml.includes('期末')
|
||
);
|
||
}""",
|
||
timeout=15000,
|
||
)
|
||
fatal = [e for e in errors if "openModal is not defined" in e
|
||
or "Cannot read properties of null" in e]
|
||
self.assertEqual([], fatal, fatal)
|
||
finally:
|
||
browser.close()
|
||
|
||
def test_01_company_gap_notice_then_admin_approve(self) -> None:
|
||
with sync_playwright() as p:
|
||
browser, page, errors = self._new_page(p)
|
||
try:
|
||
self._login(
|
||
page,
|
||
portal="company",
|
||
username="cashier-a",
|
||
password=CASHIER_PASSWORD,
|
||
)
|
||
# 首屏即可见断档提醒，无需手动刷新
|
||
page.wait_for_selector(
|
||
"#companyCoverageNotice",
|
||
state="visible",
|
||
timeout=10000,
|
||
)
|
||
body = page.locator("#companyCoverageBody").inner_text()
|
||
self.assertTrue(body.strip())
|
||
self.assertNotIn("0002", body)
|
||
|
||
page.click("#openAttestationFromWorkspace")
|
||
page.wait_for_selector("#attestationDialog.open", timeout=5000)
|
||
page.fill("#att-reason", "节假日账户无资金往来")
|
||
with page.expect_request(
|
||
lambda req: req.method == "POST"
|
||
and req.url.endswith("/api/company/no-business-attestations")
|
||
) as att_req:
|
||
page.click('#attestationForm button[type="submit"]')
|
||
self.assertTrue(att_req.value.post_data)
|
||
page.wait_for_function(
|
||
"""() => !document.getElementById('attestationDialog')?.classList.contains('open')""",
|
||
timeout=8000,
|
||
)
|
||
|
||
self._login(
|
||
page,
|
||
portal="admin",
|
||
username="group-admin",
|
||
password=ADMIN_PASSWORD,
|
||
)
|
||
page.click('a[data-view="audit"]')
|
||
page.wait_for_selector(
|
||
'button[data-audit-action="approve-attestation"]',
|
||
timeout=10000,
|
||
)
|
||
with page.expect_request(
|
||
lambda req: req.method == "POST"
|
||
and "/no-business-attestations/" in req.url
|
||
and req.url.endswith("/review")
|
||
):
|
||
page.click('button[data-audit-action="approve-attestation"]')
|
||
page.wait_for_selector("#reasonDialog.open", timeout=5000)
|
||
page.fill("#reasonInput", "审核通过说明")
|
||
page.click("#reasonSubmit")
|
||
page.wait_for_function(
|
||
"""() => !document.querySelector(
|
||
'button[data-audit-action=\"approve-attestation\"]'
|
||
)""",
|
||
timeout=10000,
|
||
)
|
||
fatal = [e for e in errors if "openModal is not defined" in e]
|
||
self.assertEqual([], fatal, fatal)
|
||
finally:
|
||
browser.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|