HEL-176: 公司端转账往来明细筛选与导出
新增 events 组合筛选 + keyset 分页，以及仅导出已确认明细的 CSV（含审计）； 与 B-44 ledger /events 按查询参数分发，不改库表与确认语义。 Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: multica-agent <github@multica.ai>
This commit is contained in:
总工
co-authored by
Cursor
multica-agent
Cursor
multica-agent
parent
cad12b3d28
commit
40f7a91a0f
@@ -155,10 +155,13 @@ class AppHandler(SimpleHTTPRequestHandler):
|
||||
self._handle_admin_intercompany_evidence(int(admin_evidence.group(1)))
|
||||
return
|
||||
|
||||
# Company transfer-summary (HEL-169/HEL-175, eligible_intercompany_events)
|
||||
# Company transfer-summary (HEL-169/HEL-175/HEL-176)
|
||||
if path == "/api/company/intercompany/summary":
|
||||
self._handle_company_intercompany_summary(query)
|
||||
return
|
||||
if path == "/api/company/intercompany/export.csv":
|
||||
self._handle_company_intercompany_export_csv(query)
|
||||
return
|
||||
|
||||
# B-44 intercompany positions (company, own-company scope)
|
||||
if path == "/api/company/intercompany/balances":
|
||||
@@ -169,6 +172,11 @@ class AppHandler(SimpleHTTPRequestHandler):
|
||||
self._handle_company_intercompany_pair(int(company_pair.group(1)), query)
|
||||
return
|
||||
if path == "/api/company/intercompany/events":
|
||||
# HEL-176 transfer-detail list shares this path with B-44 ledger
|
||||
# events; dispatch by distinctive query params (to/direction/…).
|
||||
if company_transfers.is_transfer_summary_events_query(query):
|
||||
self._handle_company_transfer_summary_events(query)
|
||||
else:
|
||||
self._handle_company_intercompany_events(query)
|
||||
return
|
||||
if path == "/api/company/manual-records":
|
||||
@@ -2889,6 +2897,111 @@ class AppHandler(SimpleHTTPRequestHandler):
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
def _reject_forged_company_id(self, query: dict[str, list[str]]) -> bool:
|
||||
"""Return True when the handler already sent a 400 for company_id."""
|
||||
if (query.get("company_id") or [None])[0] is not None:
|
||||
self._send_json(
|
||||
400,
|
||||
{"status": "error", "message": "不允许传入 company_id 参数。"},
|
||||
)
|
||||
return True
|
||||
return False
|
||||
|
||||
def _handle_company_transfer_summary_events(
|
||||
self, query: dict[str, list[str]]
|
||||
) -> None:
|
||||
"""HEL-176 filtered transfer-detail list (keyset pagination)."""
|
||||
connection = connect(DB_PATH)
|
||||
try:
|
||||
user, company_id = self._company_intercompany_scope(connection)
|
||||
if company_id is None:
|
||||
return
|
||||
if self._reject_forged_company_id(query):
|
||||
return
|
||||
try:
|
||||
payload = company_transfers.company_intercompany_events(
|
||||
connection,
|
||||
company_id=int(company_id),
|
||||
from_=(query.get("from") or [None])[0],
|
||||
to=(query.get("to") or [None])[0],
|
||||
counterparty_id=(query.get("counterparty_id") or [None])[0],
|
||||
direction=(query.get("direction") or [None])[0],
|
||||
state=(query.get("state") or [None])[0],
|
||||
limit=(query.get("limit") or [None])[0],
|
||||
cursor=(query.get("cursor") or [None])[0],
|
||||
)
|
||||
except company_transfers.TransferSummaryInputError as exc:
|
||||
self._send_json(400, {"status": "error", "message": str(exc)})
|
||||
return
|
||||
self._send_json(200, {"status": "ok", **payload})
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
def _handle_company_intercompany_export_csv(
|
||||
self, query: dict[str, list[str]]
|
||||
) -> None:
|
||||
"""Confirmed-only CSV export; session company scope + audit (HEL-176)."""
|
||||
connection = connect(DB_PATH)
|
||||
try:
|
||||
user, company_id = self._company_intercompany_scope(connection)
|
||||
if company_id is None:
|
||||
return
|
||||
if self._reject_forged_company_id(query):
|
||||
return
|
||||
# Pending must never leave via export, even if a client sends state=.
|
||||
if (query.get("state") or [None])[0] not in (None, "", "confirmed"):
|
||||
self._send_json(
|
||||
400,
|
||||
{
|
||||
"status": "error",
|
||||
"message": "导出仅支持已确认明细，不允许导出待确认。",
|
||||
},
|
||||
)
|
||||
return
|
||||
try:
|
||||
items, meta = company_transfers.company_intercompany_export_rows(
|
||||
connection,
|
||||
company_id=int(company_id),
|
||||
from_=(query.get("from") or [None])[0],
|
||||
to=(query.get("to") or [None])[0],
|
||||
counterparty_id=(query.get("counterparty_id") or [None])[0],
|
||||
direction=(query.get("direction") or [None])[0],
|
||||
)
|
||||
except company_transfers.TransferSummaryInputError as exc:
|
||||
self._send_json(400, {"status": "error", "message": str(exc)})
|
||||
return
|
||||
content = company_transfers.render_intercompany_export_csv(items)
|
||||
detail_parts = [
|
||||
f"rows:{meta['row_count']}",
|
||||
f"from:{meta['start']}",
|
||||
f"to:{meta['end']}",
|
||||
"state:confirmed",
|
||||
]
|
||||
if meta["counterparty_id"] is not None:
|
||||
detail_parts.append(f"counterparty_id:{meta['counterparty_id']}")
|
||||
if meta["direction"] is not None:
|
||||
detail_parts.append(f"direction:{meta['direction']}")
|
||||
auth.audit(
|
||||
connection,
|
||||
"export_intercompany_csv",
|
||||
actor=user,
|
||||
target=f"company:{company_id}",
|
||||
detail=";".join(detail_parts),
|
||||
ip=self._client_ip,
|
||||
)
|
||||
self.send_response(200)
|
||||
self.send_header("Content-Type", "text/csv; charset=utf-8")
|
||||
self.send_header(
|
||||
"Content-Disposition",
|
||||
'attachment; filename="intercompany-export.csv"',
|
||||
)
|
||||
self.send_header("Content-Length", str(len(content)))
|
||||
self.send_header("Cache-Control", "no-store")
|
||||
self.end_headers()
|
||||
self.wfile.write(content)
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
def _handle_company_intercompany_balances(self, query: dict[str, list[str]]) -> None:
|
||||
connection = connect(DB_PATH)
|
||||
try:
|
||||
|
||||
@@ -1,22 +1,32 @@
|
||||
"""Company-portal intercompany transfer summary (HEL-175 / HEL-169).
|
||||
"""Company-portal intercompany transfer summary / detail / export (HEL-175/176).
|
||||
|
||||
Confirmed totals reuse the authoritative ``eligible_intercompany_events``
|
||||
view (intercompany + paired or locked). Pending counts/amounts are listed
|
||||
separately and never enter outflow, inflow or net. All money math uses
|
||||
``Decimal`` on stored TEXT amounts — never float or SQLite SUM.
|
||||
|
||||
Detail listing (HEL-176) applies counterparty / date / direction / state
|
||||
filters inside SQL before LIMIT, and uses keyset pagination on
|
||||
(effective_at, event_id) descending. Export only ships confirmed rows.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import csv
|
||||
import io
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from decimal import Decimal, InvalidOperation
|
||||
import re
|
||||
import sqlite3
|
||||
|
||||
from . import settings
|
||||
from . import matching, settings
|
||||
|
||||
_DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
|
||||
_ZERO = Decimal("0.00")
|
||||
_DEFAULT_PAGE = 50
|
||||
_MAX_PAGE = 200
|
||||
_MAX_EXPORT_ROWS = 20000
|
||||
|
||||
|
||||
class TransferSummaryInputError(ValueError):
|
||||
@@ -296,3 +306,435 @@ def company_intercompany_summary(
|
||||
},
|
||||
"counterparties": counterparties,
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Detail list + CSV export (HEL-176)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def encode_cursor(effective_at: str, event_id: int) -> str:
|
||||
raw = f"{effective_at}|{event_id}"
|
||||
return base64.urlsafe_b64encode(raw.encode("utf-8")).decode("ascii")
|
||||
|
||||
|
||||
def decode_cursor(cursor: str | None) -> tuple[str, int] | None:
|
||||
if not cursor:
|
||||
return None
|
||||
try:
|
||||
raw = base64.urlsafe_b64decode(cursor.encode("ascii")).decode("utf-8")
|
||||
effective_at, event_id_text = raw.split("|", 1)
|
||||
return effective_at, int(event_id_text)
|
||||
except Exception as exc:
|
||||
raise TransferSummaryInputError("分页游标无效。") from exc
|
||||
|
||||
|
||||
def _parse_limit(raw: object | None, default: int = _DEFAULT_PAGE) -> int:
|
||||
if raw is None or str(raw).strip() == "":
|
||||
return default
|
||||
try:
|
||||
value = int(str(raw).strip())
|
||||
except (TypeError, ValueError) as exc:
|
||||
raise TransferSummaryInputError("limit 必须是正整数。") from exc
|
||||
if value < 1:
|
||||
raise TransferSummaryInputError("limit 必须是正整数。")
|
||||
return min(value, _MAX_PAGE)
|
||||
|
||||
|
||||
def _parse_optional_date(raw: object | None, label: str) -> str | None:
|
||||
if raw is None or str(raw).strip() == "":
|
||||
return None
|
||||
return _validate_date(raw, label)
|
||||
|
||||
|
||||
def _parse_direction(raw: object | None) -> str | None:
|
||||
if raw is None or str(raw).strip() == "":
|
||||
return None
|
||||
value = str(raw).strip().lower()
|
||||
if value not in ("out", "in"):
|
||||
raise TransferSummaryInputError("direction 只能是 out 或 in。")
|
||||
return value
|
||||
|
||||
|
||||
def _parse_state(raw: object | None) -> str | None:
|
||||
if raw is None or str(raw).strip() == "":
|
||||
return None
|
||||
value = str(raw).strip().lower()
|
||||
if value not in ("confirmed", "pending"):
|
||||
raise TransferSummaryInputError("state 只能是 confirmed 或 pending。")
|
||||
return value
|
||||
|
||||
|
||||
def _parse_counterparty_id(raw: object | None) -> int | None:
|
||||
if raw is None or str(raw).strip() == "":
|
||||
return None
|
||||
try:
|
||||
value = int(str(raw).strip())
|
||||
except (TypeError, ValueError) as exc:
|
||||
raise TransferSummaryInputError("counterparty_id 参数无效。") from exc
|
||||
if value < 1:
|
||||
raise TransferSummaryInputError("counterparty_id 参数无效。")
|
||||
return value
|
||||
|
||||
|
||||
def _event_window(
|
||||
connection: sqlite3.Connection,
|
||||
*,
|
||||
from_: str | None,
|
||||
to: str | None,
|
||||
) -> tuple[str, str]:
|
||||
end = _parse_optional_date(to, "to") or today_shanghai()
|
||||
start_default = settings.get_settings(connection).get("start_date") or "2026-01-01"
|
||||
start = _parse_optional_date(from_, "from") or _validate_date(
|
||||
start_default, "start_date"
|
||||
)
|
||||
if start > end:
|
||||
raise TransferSummaryInputError("from 不能晚于 to。")
|
||||
return start, end
|
||||
|
||||
|
||||
# Confirmed = eligible_intercompany_events. Pending matches summary tip set.
|
||||
_CONFIRMED_PREDICATE = """
|
||||
d.classification = 'intercompany'
|
||||
AND (d.pairing = 'paired' OR d.locked = 1)
|
||||
"""
|
||||
|
||||
_PENDING_PREDICATE = """
|
||||
(
|
||||
d.classification IN ('unresolved', 'needs_review', 'internal_single')
|
||||
OR (
|
||||
d.classification = 'intercompany'
|
||||
AND d.pairing != 'paired'
|
||||
AND d.locked = 0
|
||||
)
|
||||
)
|
||||
"""
|
||||
|
||||
_BOTH_PREDICATE = f"""
|
||||
(
|
||||
({_CONFIRMED_PREDICATE})
|
||||
OR ({_PENDING_PREDICATE})
|
||||
)
|
||||
"""
|
||||
|
||||
|
||||
def _state_predicate(state: str | None) -> str:
|
||||
if state == "confirmed":
|
||||
return f"({_CONFIRMED_PREDICATE})"
|
||||
if state == "pending":
|
||||
return f"({_PENDING_PREDICATE})"
|
||||
return _BOTH_PREDICATE
|
||||
|
||||
|
||||
def _list_select_sql() -> str:
|
||||
return """
|
||||
SELECT c.event_id, d.id AS decision_id, d.revision, d.classification,
|
||||
d.pairing, d.amount, d.currency, d.effective_at, d.mode,
|
||||
d.locked, d.rule_version, d.created_at, d.reason,
|
||||
payer.company_id AS payer_company_id,
|
||||
payee.company_id AS payee_company_id,
|
||||
payer.bank_account_id AS payer_account_id,
|
||||
payee.bank_account_id AS payee_account_id,
|
||||
cpayer.name AS payer_company_name,
|
||||
cpayee.name AS payee_company_name,
|
||||
(SELECT COUNT(*) FROM transfer_decision_observations o
|
||||
WHERE o.decision_id = d.id) AS evidence_count,
|
||||
(SELECT r.summary
|
||||
FROM transfer_decision_observations o
|
||||
JOIN source_rows r ON r.id = o.source_row_id
|
||||
JOIN sheet_batches s ON s.id = r.sheet_batch_id
|
||||
JOIN import_batches b ON b.id = s.import_batch_id
|
||||
WHERE o.decision_id = d.id AND b.company_id = ?
|
||||
ORDER BY o.id
|
||||
LIMIT 1) AS summary
|
||||
FROM current_transfer_decisions c
|
||||
JOIN transfer_match_decisions d ON d.id = c.decision_id
|
||||
JOIN canonical_transfer_events e ON e.id = c.event_id
|
||||
LEFT JOIN transfer_decision_participants payer
|
||||
ON payer.decision_id = d.id AND payer.role = 'payer'
|
||||
LEFT JOIN transfer_decision_participants payee
|
||||
ON payee.decision_id = d.id AND payee.role = 'payee'
|
||||
LEFT JOIN companies cpayer ON cpayer.id = payer.company_id
|
||||
LEFT JOIN companies cpayee ON cpayee.id = payee.company_id
|
||||
"""
|
||||
|
||||
|
||||
def _build_event_filters(
|
||||
*,
|
||||
company_id: int,
|
||||
start: str,
|
||||
end: str,
|
||||
counterparty_id: int | None,
|
||||
direction: str | None,
|
||||
state: str | None,
|
||||
cursor: tuple[str, int] | None,
|
||||
) -> tuple[str, list[object]]:
|
||||
conditions = [
|
||||
"e.lifecycle = 'active'",
|
||||
"(payer.company_id = ? OR payee.company_id = ?)",
|
||||
"d.effective_at >= ?",
|
||||
"d.effective_at <= ?",
|
||||
_state_predicate(state),
|
||||
]
|
||||
params: list[object] = [
|
||||
company_id,
|
||||
company_id,
|
||||
start,
|
||||
end + "T23:59:59",
|
||||
]
|
||||
|
||||
if counterparty_id is not None:
|
||||
# Counterparty is the other participant; own company stays forced above.
|
||||
conditions.append(
|
||||
"""
|
||||
(
|
||||
(payer.company_id = ? AND payee.company_id = ?)
|
||||
OR (payee.company_id = ? AND payer.company_id = ?)
|
||||
)
|
||||
"""
|
||||
)
|
||||
params.extend([company_id, counterparty_id, company_id, counterparty_id])
|
||||
|
||||
if direction == "out":
|
||||
conditions.append("payer.company_id = ?")
|
||||
params.append(company_id)
|
||||
elif direction == "in":
|
||||
conditions.append("payee.company_id = ?")
|
||||
params.append(company_id)
|
||||
|
||||
if cursor is not None:
|
||||
cursor_at, cursor_id = cursor
|
||||
conditions.append(
|
||||
"""
|
||||
(
|
||||
d.effective_at < ?
|
||||
OR (d.effective_at = ? AND c.event_id < ?)
|
||||
)
|
||||
"""
|
||||
)
|
||||
params.extend([cursor_at, cursor_at, cursor_id])
|
||||
|
||||
where = " WHERE " + " AND ".join(conditions)
|
||||
return where, params
|
||||
|
||||
|
||||
def _row_state(row: sqlite3.Row) -> str:
|
||||
classification = row["classification"]
|
||||
pairing = row["pairing"]
|
||||
locked = bool(row["locked"])
|
||||
if classification == "intercompany" and (pairing == "paired" or locked):
|
||||
return "confirmed"
|
||||
return "pending"
|
||||
|
||||
|
||||
def _row_direction(row: sqlite3.Row, company_id: int) -> str | None:
|
||||
if row["payer_company_id"] is not None and int(row["payer_company_id"]) == int(
|
||||
company_id
|
||||
):
|
||||
return "out"
|
||||
if row["payee_company_id"] is not None and int(row["payee_company_id"]) == int(
|
||||
company_id
|
||||
):
|
||||
return "in"
|
||||
return None
|
||||
|
||||
|
||||
def _event_list_item(row: sqlite3.Row, company_id: int) -> dict[str, object]:
|
||||
direction = _row_direction(row, company_id)
|
||||
state = _row_state(row)
|
||||
if direction == "out":
|
||||
counterparty_company_id = row["payee_company_id"]
|
||||
counterparty_company_name = row["payee_company_name"]
|
||||
else:
|
||||
counterparty_company_id = row["payer_company_id"]
|
||||
counterparty_company_name = row["payer_company_name"]
|
||||
summary = row["summary"] or row["reason"] or ""
|
||||
return {
|
||||
"event_id": int(row["event_id"]),
|
||||
"decision_id": int(row["decision_id"]),
|
||||
"revision": row["revision"],
|
||||
"classification": row["classification"],
|
||||
"pairing": row["pairing"],
|
||||
"status": matching.exposed_status(row),
|
||||
"state": state,
|
||||
"direction": direction,
|
||||
"amount": row["amount"],
|
||||
"currency": row["currency"],
|
||||
"effective_at": row["effective_at"],
|
||||
"mode": row["mode"],
|
||||
"locked": bool(row["locked"]),
|
||||
"rule_version": row["rule_version"],
|
||||
"summary": summary,
|
||||
"own_company_id": company_id,
|
||||
"counterparty_company_id": (
|
||||
int(counterparty_company_id) if counterparty_company_id is not None else None
|
||||
),
|
||||
"counterparty_company_name": counterparty_company_name,
|
||||
"evidence_count": row["evidence_count"],
|
||||
}
|
||||
|
||||
|
||||
def company_intercompany_events(
|
||||
connection: sqlite3.Connection,
|
||||
*,
|
||||
company_id: int,
|
||||
from_: str | None = None,
|
||||
to: str | None = None,
|
||||
counterparty_id: int | None | object = None,
|
||||
direction: str | None | object = None,
|
||||
state: str | None | object = None,
|
||||
limit: object | None = None,
|
||||
cursor: str | None = None,
|
||||
) -> dict[str, object]:
|
||||
"""Filtered keyset page of company-visible transfer events (HEL-176)."""
|
||||
start, end = _event_window(connection, from_=from_, to=to)
|
||||
cp_id = _parse_counterparty_id(counterparty_id)
|
||||
direction_value = _parse_direction(direction)
|
||||
state_value = _parse_state(state)
|
||||
page_size = _parse_limit(limit)
|
||||
cursor_tuple = decode_cursor(cursor)
|
||||
|
||||
where, params = _build_event_filters(
|
||||
company_id=company_id,
|
||||
start=start,
|
||||
end=end,
|
||||
counterparty_id=cp_id,
|
||||
direction=direction_value,
|
||||
state=state_value,
|
||||
cursor=cursor_tuple,
|
||||
)
|
||||
# summary subquery binds own company_id first.
|
||||
sql = (
|
||||
_list_select_sql()
|
||||
+ where
|
||||
+ " ORDER BY d.effective_at DESC, c.event_id DESC LIMIT ?"
|
||||
)
|
||||
rows = connection.execute(
|
||||
sql, (company_id, *params, page_size + 1)
|
||||
).fetchall()
|
||||
|
||||
has_more = len(rows) > page_size
|
||||
page = rows[:page_size]
|
||||
items = [_event_list_item(row, company_id) for row in page]
|
||||
next_cursor = None
|
||||
if has_more and page:
|
||||
last = page[-1]
|
||||
next_cursor = encode_cursor(str(last["effective_at"]), int(last["event_id"]))
|
||||
|
||||
return {
|
||||
"window": {"start": start, "end": end},
|
||||
"events": items,
|
||||
"next_cursor": next_cursor,
|
||||
"has_more": has_more,
|
||||
}
|
||||
|
||||
|
||||
def company_intercompany_export_rows(
|
||||
connection: sqlite3.Connection,
|
||||
*,
|
||||
company_id: int,
|
||||
from_: str | None = None,
|
||||
to: str | None = None,
|
||||
counterparty_id: int | None | object = None,
|
||||
direction: str | None | object = None,
|
||||
) -> tuple[list[dict[str, object]], dict[str, object]]:
|
||||
"""Confirmed-only rows for CSV export; pending never included."""
|
||||
start, end = _event_window(connection, from_=from_, to=to)
|
||||
cp_id = _parse_counterparty_id(counterparty_id)
|
||||
direction_value = _parse_direction(direction)
|
||||
|
||||
where, params = _build_event_filters(
|
||||
company_id=company_id,
|
||||
start=start,
|
||||
end=end,
|
||||
counterparty_id=cp_id,
|
||||
direction=direction_value,
|
||||
state="confirmed",
|
||||
cursor=None,
|
||||
)
|
||||
sql = (
|
||||
_list_select_sql()
|
||||
+ where
|
||||
+ " ORDER BY d.effective_at DESC, c.event_id DESC LIMIT ?"
|
||||
)
|
||||
rows = connection.execute(
|
||||
sql, (company_id, *params, _MAX_EXPORT_ROWS + 1)
|
||||
).fetchall()
|
||||
if len(rows) > _MAX_EXPORT_ROWS:
|
||||
raise TransferSummaryInputError(
|
||||
f"导出行数超过上限 {_MAX_EXPORT_ROWS}，请缩小筛选范围。"
|
||||
)
|
||||
items = [_event_list_item(row, company_id) for row in rows]
|
||||
meta = {
|
||||
"start": start,
|
||||
"end": end,
|
||||
"counterparty_id": cp_id,
|
||||
"direction": direction_value,
|
||||
"state": "confirmed",
|
||||
"row_count": len(items),
|
||||
}
|
||||
return items, meta
|
||||
|
||||
|
||||
def render_intercompany_export_csv(items: list[dict[str, object]]) -> bytes:
|
||||
buffer = io.StringIO()
|
||||
writer = csv.writer(buffer)
|
||||
writer.writerow(
|
||||
[
|
||||
"日期",
|
||||
"方向",
|
||||
"对方公司",
|
||||
"金额",
|
||||
"币种",
|
||||
"摘要",
|
||||
"状态",
|
||||
"配对",
|
||||
"事件ID",
|
||||
"决策ID",
|
||||
]
|
||||
)
|
||||
direction_label = {"out": "转出", "in": "转入"}
|
||||
for item in items:
|
||||
writer.writerow(
|
||||
[
|
||||
item.get("effective_at") or "",
|
||||
direction_label.get(str(item.get("direction") or ""), ""),
|
||||
item.get("counterparty_company_name") or "",
|
||||
item.get("amount") or "",
|
||||
item.get("currency") or "",
|
||||
item.get("summary") or "",
|
||||
"已确认",
|
||||
item.get("pairing") or "",
|
||||
item.get("event_id") or "",
|
||||
item.get("decision_id") or "",
|
||||
]
|
||||
)
|
||||
# UTF-8 BOM so Excel opens the CSV with the right encoding.
|
||||
return (chr(0xFEFF) + buffer.getvalue()).encode("utf-8")
|
||||
|
||||
|
||||
def is_transfer_summary_events_query(query: dict[str, list[str]]) -> bool:
|
||||
"""Discriminate HEL-176 transfer list from B-44 ledger ``/events``.
|
||||
|
||||
B-44 uses ``cutoff`` / subject / posting_kind / source_kind / pending_subject.
|
||||
HEL-176 uses ``to`` / direction / counterparty_id / state=pending|confirmed
|
||||
(without ledger-only knobs).
|
||||
"""
|
||||
if (query.get("direction") or [None])[0] is not None:
|
||||
return True
|
||||
if (query.get("counterparty_id") or [None])[0] is not None:
|
||||
return True
|
||||
if (query.get("to") or [None])[0] is not None:
|
||||
return True
|
||||
state = (query.get("state") or [None])[0]
|
||||
if state in ("pending", "confirmed") and (query.get("cutoff") or [None])[0] is None:
|
||||
# Bare state=confirmed without cutoff is the transfer-summary list;
|
||||
# B-44 confirmed always pairs with cutoff in existing callers/tests.
|
||||
if (query.get("subject") or [None])[0] is not None:
|
||||
return False
|
||||
if (query.get("posting_kind") or [None])[0] is not None:
|
||||
return False
|
||||
if (query.get("source_kind") or [None])[0] is not None:
|
||||
return False
|
||||
return True
|
||||
return False
|
||||
|
||||
@@ -0,0 +1,500 @@
|
||||
"""HTTP tests for company intercompany events list + CSV export (HEL-176).
|
||||
|
||||
Covers combined filters, keyset pagination, empty state, lateral access,
|
||||
forged company_id, ID guessing, export isolation and audit logging.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
import threading
|
||||
import unittest
|
||||
|
||||
from openpyxl import Workbook
|
||||
|
||||
from bank_importer.db import connect, migrate
|
||||
|
||||
import server
|
||||
from test_server_auth import Client, as_json
|
||||
|
||||
BOOTSTRAP_PASSWORD = "BootAdmin123"
|
||||
ADMIN_PASSWORD = "AdminPass123"
|
||||
CASHIER_PASSWORD = "Cashier123"
|
||||
|
||||
CCB_HEADER = [
|
||||
"客户账号", "账户名称", "交易时间", "借方发生额（支取）", "贷方发生额（收入）",
|
||||
"余额", "币种", "对方户名", "对方账号", "对方开户机构", "摘要", "备注",
|
||||
]
|
||||
|
||||
ACCOUNT_A = "6222000000000001"
|
||||
ACCOUNT_B = "6222000000000002"
|
||||
ACCOUNT_C = "6222000000000003"
|
||||
|
||||
|
||||
def workbook_bytes(rows) -> bytes:
|
||||
workbook = Workbook()
|
||||
sheet = workbook.active
|
||||
sheet.title = "正常流水"
|
||||
sheet.append(CCB_HEADER)
|
||||
for row in rows:
|
||||
sheet.append(row)
|
||||
buffer = io.BytesIO()
|
||||
workbook.save(buffer)
|
||||
return buffer.getvalue()
|
||||
|
||||
|
||||
def outgoing(own: str, cp: str, amount: str, at: str = "2026-01-05 10:00:00"):
|
||||
return [own, "测试公司", at, amount, "", "50000.00", "RMB", "对方", cp, "某银行", "货款", ""]
|
||||
|
||||
|
||||
def incoming(own: str, cp: str, amount: str, at: str = "2026-01-05 11:00:00"):
|
||||
return [own, "测试公司", at, "", amount, "50000.00", "RMB", "对方", cp, "某银行", "收款", ""]
|
||||
|
||||
|
||||
class CompanyIntercompanyEventsTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.temp_dir = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.temp_dir.cleanup)
|
||||
root = Path(self.temp_dir.name)
|
||||
self.db_path = root / "app.db"
|
||||
self.storage = root / "files"
|
||||
|
||||
self._old_db_path = server.DB_PATH
|
||||
self._old_storage = server.STORAGE_DIR
|
||||
server.DB_PATH = self.db_path
|
||||
server.STORAGE_DIR = self.storage
|
||||
|
||||
os.environ["APP_BOOTSTRAP_ADMIN_PASSWORD"] = BOOTSTRAP_PASSWORD
|
||||
connection = connect(self.db_path)
|
||||
migrate(connection)
|
||||
assert server.ensure_bootstrap_admin(connection) is None
|
||||
connection.close()
|
||||
|
||||
class QuietHandler(server.AppHandler):
|
||||
def log_message(self, *args) -> None:
|
||||
pass
|
||||
|
||||
self.httpd = server.ThreadingHTTPServer(("127.0.0.1", 0), QuietHandler)
|
||||
self.port = self.httpd.server_address[1]
|
||||
self.thread = threading.Thread(target=self.httpd.serve_forever, daemon=True)
|
||||
self.thread.start()
|
||||
|
||||
self.admin = Client("127.0.0.1", self.port)
|
||||
status, _, data = self.admin.post_json(
|
||||
"/api/login",
|
||||
{"username": "group-admin", "password": BOOTSTRAP_PASSWORD, "portal": "admin"},
|
||||
)
|
||||
assert status == 200, data
|
||||
status, _, data = self.admin.post_json(
|
||||
"/api/password/change",
|
||||
{"old_password": BOOTSTRAP_PASSWORD, "new_password": ADMIN_PASSWORD},
|
||||
)
|
||||
assert status == 200, data
|
||||
|
||||
self.initial_passwords: dict[str, str] = {}
|
||||
self.company_a = self._create_company("甲公司", "cashier-a")
|
||||
self.company_b = self._create_company("乙公司", "cashier-b")
|
||||
self.company_c = self._create_company("丙公司", "cashier-c")
|
||||
self.cashier_a = self._login_company("cashier-a")
|
||||
self.cashier_b = self._login_company("cashier-b")
|
||||
self.cashier_c = self._login_company("cashier-c")
|
||||
|
||||
self._approve_account(self.company_a, ACCOUNT_A, self.cashier_a)
|
||||
self._approve_account(self.company_b, ACCOUNT_B, self.cashier_b)
|
||||
self._approve_account(self.company_c, ACCOUNT_C, self.cashier_c)
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self.httpd.shutdown()
|
||||
self.httpd.server_close()
|
||||
server.DB_PATH = self._old_db_path
|
||||
server.STORAGE_DIR = self._old_storage
|
||||
os.environ.pop("APP_BOOTSTRAP_ADMIN_PASSWORD", None)
|
||||
|
||||
def _create_company(self, name: str, username: str) -> int:
|
||||
status, _, data = self.admin.post_json(
|
||||
"/api/admin/companies", {"name": name, "username": username}
|
||||
)
|
||||
assert status == 200, data
|
||||
self.initial_passwords[username] = as_json(data)["initial_password"]
|
||||
return as_json(data)["company_id"]
|
||||
|
||||
def _login_company(self, username: str) -> Client:
|
||||
client = Client("127.0.0.1", self.port)
|
||||
initial = self.initial_passwords[username]
|
||||
status, _, data = client.post_json(
|
||||
"/api/login", {"username": username, "password": initial, "portal": "company"}
|
||||
)
|
||||
assert status == 200, data
|
||||
status, _, data = client.post_json(
|
||||
"/api/password/change",
|
||||
{"old_password": initial, "new_password": CASHIER_PASSWORD},
|
||||
)
|
||||
assert status == 200, data
|
||||
return client
|
||||
|
||||
def _approve_account(self, company_id: int, number: str, client: Client) -> int:
|
||||
status, _, data = client.post_json(
|
||||
"/api/company/accounts",
|
||||
{
|
||||
"bank_name": "中信银行",
|
||||
"account_type": "基本户",
|
||||
"account_number": number,
|
||||
"start_date": "2026-01-01",
|
||||
},
|
||||
)
|
||||
assert status == 200, data
|
||||
account_id = as_json(data)["account"]["id"]
|
||||
status, _, data = self.admin.post_json(
|
||||
f"/api/admin/accounts/{account_id}/review",
|
||||
{
|
||||
"decision": "approve",
|
||||
"reason": "测试启用",
|
||||
"effective_from": "2026-01-01",
|
||||
},
|
||||
)
|
||||
assert status == 200, data
|
||||
return account_id
|
||||
|
||||
def _upload_and_confirm(self, client: Client, company_id: int, rows) -> int:
|
||||
content = workbook_bytes(rows)
|
||||
status, _, data = self.admin.post_multipart(
|
||||
"/api/parse", {"company_id": str(company_id)}, "账单.xlsx", content
|
||||
)
|
||||
assert status == 200, data
|
||||
batch_id = as_json(data)["batch_id"]
|
||||
status, _, data = client.get(f"/api/batches/{batch_id}/sheets")
|
||||
names = [s["sheet_name"] for s in as_json(data)["sheets"] if s["outcome"] == "parsed"]
|
||||
status, _, data = client.post_json(
|
||||
f"/api/batches/{batch_id}/confirm", {"sheets": names}
|
||||
)
|
||||
assert status == 200, data
|
||||
return batch_id
|
||||
|
||||
def _lock_single(self, amount: str, at: str = "2026-03-01 10:00:00") -> dict:
|
||||
self._upload_and_confirm(
|
||||
self.cashier_a,
|
||||
self.company_a,
|
||||
[outgoing(ACCOUNT_A, ACCOUNT_B, amount, at)],
|
||||
)
|
||||
status, _, data = self.admin.get("/api/admin/transfer-events")
|
||||
self.assertEqual(200, status, data)
|
||||
single = next(
|
||||
e
|
||||
for e in as_json(data)["events"]
|
||||
if e["status"] == "internal_single" and e["amount"] == amount
|
||||
)
|
||||
status, _, data = self.admin.get(f"/api/admin/transfer-events/{single['event_id']}")
|
||||
self.assertEqual(200, status, data)
|
||||
revision = as_json(data)["event"]["revision"]
|
||||
status, _, data = self.admin.post_json(
|
||||
f"/api/admin/transfer-events/{single['event_id']}/decisions",
|
||||
{
|
||||
"action": "assign_participant",
|
||||
"reason": "函证确认",
|
||||
"expected_revision": revision,
|
||||
"request_key": f"lock-{amount}-{at}",
|
||||
"participant": {"role": "payee", "company_id": self.company_b},
|
||||
},
|
||||
)
|
||||
self.assertEqual(200, status, data)
|
||||
return as_json(data)["decision"]
|
||||
|
||||
def _seed_mixed(self) -> None:
|
||||
"""Paired A→B 100, B→A 40, locked A→B 25, pending A→B 7, B→C 200."""
|
||||
self._upload_and_confirm(
|
||||
self.cashier_a, self.company_a,
|
||||
[outgoing(ACCOUNT_A, ACCOUNT_B, "100.00", "2026-01-05 10:00:00")],
|
||||
)
|
||||
self._upload_and_confirm(
|
||||
self.cashier_b, self.company_b,
|
||||
[incoming(ACCOUNT_B, ACCOUNT_A, "100.00", "2026-01-05 11:00:00")],
|
||||
)
|
||||
self._upload_and_confirm(
|
||||
self.cashier_b, self.company_b,
|
||||
[outgoing(ACCOUNT_B, ACCOUNT_A, "40.00", "2026-01-10 10:00:00")],
|
||||
)
|
||||
self._upload_and_confirm(
|
||||
self.cashier_a, self.company_a,
|
||||
[incoming(ACCOUNT_A, ACCOUNT_B, "40.00", "2026-01-10 11:00:00")],
|
||||
)
|
||||
self._lock_single("25.00", "2026-02-01 10:00:00")
|
||||
self._upload_and_confirm(
|
||||
self.cashier_a, self.company_a,
|
||||
[outgoing(ACCOUNT_A, ACCOUNT_B, "7.00", "2026-02-15 10:00:00")],
|
||||
)
|
||||
self._upload_and_confirm(
|
||||
self.cashier_b, self.company_b,
|
||||
[outgoing(ACCOUNT_B, ACCOUNT_C, "200.00", "2026-01-20 10:00:00")],
|
||||
)
|
||||
self._upload_and_confirm(
|
||||
self.cashier_c, self.company_c,
|
||||
[incoming(ACCOUNT_C, ACCOUNT_B, "200.00", "2026-01-20 11:00:00")],
|
||||
)
|
||||
|
||||
def _events(self, client: Client, query: str = "from=2026-01-01&to=2026-12-31"):
|
||||
status, _, data = client.get(f"/api/company/intercompany/events?{query}")
|
||||
return status, as_json(data) if data else {}
|
||||
|
||||
def _export(self, client: Client, query: str = "from=2026-01-01&to=2026-12-31"):
|
||||
return client.get(f"/api/company/intercompany/export.csv?{query}")
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Auth / parameter guards
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def test_requires_company_role(self) -> None:
|
||||
status, payload = self._events(self.admin)
|
||||
self.assertEqual(403, status, payload)
|
||||
|
||||
def test_rejects_forged_company_id_on_events_and_export(self) -> None:
|
||||
status, payload = self._events(
|
||||
self.cashier_a,
|
||||
f"from=2026-01-01&to=2026-12-31&company_id={self.company_b}",
|
||||
)
|
||||
self.assertEqual(400, status, payload)
|
||||
|
||||
status, _, data = self._export(
|
||||
self.cashier_a,
|
||||
f"from=2026-01-01&to=2026-12-31&company_id={self.company_a}",
|
||||
)
|
||||
self.assertEqual(400, status, data)
|
||||
|
||||
def test_rejects_bad_filters(self) -> None:
|
||||
status, payload = self._events(
|
||||
self.cashier_a, "from=2026-01-01&to=2026-12-31&direction=sideways"
|
||||
)
|
||||
self.assertEqual(400, status, payload)
|
||||
status, payload = self._events(
|
||||
self.cashier_a, "from=2026-01-01&to=2026-12-31&state=maybe"
|
||||
)
|
||||
self.assertEqual(400, status, payload)
|
||||
status, payload = self._events(
|
||||
self.cashier_a, "from=2026-13-40&to=2026-12-31"
|
||||
)
|
||||
self.assertEqual(400, status, payload)
|
||||
|
||||
def test_empty_window(self) -> None:
|
||||
status, payload = self._events(self.cashier_a)
|
||||
self.assertEqual(200, status, payload)
|
||||
self.assertEqual([], payload["events"])
|
||||
self.assertFalse(payload["has_more"])
|
||||
self.assertIsNone(payload["next_cursor"])
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Combined filters + confirmed/pending separation
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def test_combined_filters_and_state_split(self) -> None:
|
||||
self._seed_mixed()
|
||||
|
||||
status, payload = self._events(self.cashier_a)
|
||||
self.assertEqual(200, status, payload)
|
||||
events = payload["events"]
|
||||
# A sees: out 100, in 40, locked out 25, pending out 7 — not B↔C 200
|
||||
self.assertEqual(4, len(events))
|
||||
amounts = {e["amount"] for e in events}
|
||||
self.assertEqual({"100.00", "40.00", "25.00", "7.00"}, amounts)
|
||||
for event in events:
|
||||
self.assertNotEqual(self.company_c, event["counterparty_company_id"])
|
||||
self.assertIn(event["state"], ("confirmed", "pending"))
|
||||
self.assertIn(event["direction"], ("out", "in"))
|
||||
|
||||
status, confirmed = self._events(
|
||||
self.cashier_a, "from=2026-01-01&to=2026-12-31&state=confirmed"
|
||||
)
|
||||
self.assertEqual(200, status, confirmed)
|
||||
self.assertEqual(3, len(confirmed["events"]))
|
||||
self.assertTrue(all(e["state"] == "confirmed" for e in confirmed["events"]))
|
||||
self.assertNotIn("7.00", {e["amount"] for e in confirmed["events"]})
|
||||
|
||||
status, pending = self._events(
|
||||
self.cashier_a, "from=2026-01-01&to=2026-12-31&state=pending"
|
||||
)
|
||||
self.assertEqual(200, status, pending)
|
||||
self.assertEqual(1, len(pending["events"]))
|
||||
self.assertEqual("7.00", pending["events"][0]["amount"])
|
||||
self.assertEqual("pending", pending["events"][0]["state"])
|
||||
|
||||
status, outs = self._events(
|
||||
self.cashier_a, "from=2026-01-01&to=2026-12-31&direction=out&state=confirmed"
|
||||
)
|
||||
self.assertEqual(200, status, outs)
|
||||
self.assertEqual({"100.00", "25.00"}, {e["amount"] for e in outs["events"]})
|
||||
self.assertTrue(all(e["direction"] == "out" for e in outs["events"]))
|
||||
|
||||
status, by_cp = self._events(
|
||||
self.cashier_a,
|
||||
f"from=2026-01-01&to=2026-12-31&counterparty_id={self.company_b}&state=confirmed",
|
||||
)
|
||||
self.assertEqual(200, status, by_cp)
|
||||
self.assertEqual(3, len(by_cp["events"]))
|
||||
|
||||
# Date window excludes Feb locked/pending
|
||||
status, jan = self._events(
|
||||
self.cashier_a, "from=2026-01-01&to=2026-01-31&state=confirmed"
|
||||
)
|
||||
self.assertEqual(200, status, jan)
|
||||
self.assertEqual({"100.00", "40.00"}, {e["amount"] for e in jan["events"]})
|
||||
|
||||
def test_keyset_pagination_no_dup_no_gap(self) -> None:
|
||||
# Three confirmed A→B outs on distinct days
|
||||
for i, amount in enumerate(("11.00", "12.00", "13.00", "14.00", "15.00")):
|
||||
day = 5 + i
|
||||
self._upload_and_confirm(
|
||||
self.cashier_a, self.company_a,
|
||||
[outgoing(ACCOUNT_A, ACCOUNT_B, amount, f"2026-01-{day:02d} 10:00:00")],
|
||||
)
|
||||
self._upload_and_confirm(
|
||||
self.cashier_b, self.company_b,
|
||||
[incoming(ACCOUNT_B, ACCOUNT_A, amount, f"2026-01-{day:02d} 11:00:00")],
|
||||
)
|
||||
|
||||
status, page1 = self._events(
|
||||
self.cashier_a,
|
||||
"from=2026-01-01&to=2026-12-31&state=confirmed&direction=out&limit=2",
|
||||
)
|
||||
self.assertEqual(200, status, page1)
|
||||
self.assertEqual(2, len(page1["events"]))
|
||||
self.assertTrue(page1["has_more"])
|
||||
self.assertIsNotNone(page1["next_cursor"])
|
||||
|
||||
status, page2 = self._events(
|
||||
self.cashier_a,
|
||||
"from=2026-01-01&to=2026-12-31&state=confirmed&direction=out"
|
||||
f"&limit=2&cursor={page1['next_cursor']}",
|
||||
)
|
||||
self.assertEqual(200, status, page2)
|
||||
self.assertEqual(2, len(page2["events"]))
|
||||
self.assertTrue(page2["has_more"])
|
||||
|
||||
status, page3 = self._events(
|
||||
self.cashier_a,
|
||||
"from=2026-01-01&to=2026-12-31&state=confirmed&direction=out"
|
||||
f"&limit=2&cursor={page2['next_cursor']}",
|
||||
)
|
||||
self.assertEqual(200, status, page3)
|
||||
self.assertEqual(1, len(page3["events"]))
|
||||
self.assertFalse(page3["has_more"])
|
||||
self.assertIsNone(page3["next_cursor"])
|
||||
|
||||
ids = [e["event_id"] for e in page1["events"] + page2["events"] + page3["events"]]
|
||||
self.assertEqual(5, len(ids))
|
||||
self.assertEqual(len(ids), len(set(ids)))
|
||||
# Descending by effective_at then event_id
|
||||
amounts = [e["amount"] for e in page1["events"] + page2["events"] + page3["events"]]
|
||||
self.assertEqual(["15.00", "14.00", "13.00", "12.00", "11.00"], amounts)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Isolation / ID guess / detail reuse
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def test_lateral_isolation_and_id_guess_404(self) -> None:
|
||||
self._seed_mixed()
|
||||
|
||||
status, payload_a = self._events(self.cashier_a)
|
||||
self.assertEqual(200, status, payload_a)
|
||||
a_ids = {e["event_id"] for e in payload_a["events"]}
|
||||
|
||||
status, payload_c = self._events(self.cashier_c)
|
||||
self.assertEqual(200, status, payload_c)
|
||||
# C only participates in B↔C 200
|
||||
self.assertTrue(payload_c["events"])
|
||||
for event in payload_c["events"]:
|
||||
self.assertEqual("200.00", event["amount"])
|
||||
self.assertNotIn(event["event_id"], a_ids)
|
||||
|
||||
# C guessing A's event id via transfer-events detail → 404
|
||||
a_event_id = next(iter(a_ids))
|
||||
status, _, data = self.cashier_c.get(
|
||||
f"/api/company/transfer-events/{a_event_id}"
|
||||
)
|
||||
self.assertEqual(404, status, data)
|
||||
|
||||
# A can open own event; counterparty account masked; only own observations
|
||||
status, _, data = self.cashier_a.get(
|
||||
f"/api/company/transfer-events/{a_event_id}"
|
||||
)
|
||||
self.assertEqual(200, status, data)
|
||||
detail = as_json(data)["event"]
|
||||
if detail.get("counterparty") and "account_number_masked" in detail["counterparty"]:
|
||||
masked = detail["counterparty"]["account_number_masked"]
|
||||
self.assertTrue(str(masked).startswith("****"))
|
||||
self.assertNotIn(ACCOUNT_B, masked)
|
||||
for obs in detail["observations"]:
|
||||
self.assertEqual(self.company_a, obs["batch_company_id"])
|
||||
self.assertNotIn(ACCOUNT_B, json.dumps(obs, ensure_ascii=False))
|
||||
|
||||
# Filtering by counterparty C still cannot leak B↔C into A's list
|
||||
status, filtered = self._events(
|
||||
self.cashier_a,
|
||||
f"from=2026-01-01&to=2026-12-31&counterparty_id={self.company_c}",
|
||||
)
|
||||
self.assertEqual(200, status, filtered)
|
||||
self.assertEqual([], filtered["events"])
|
||||
|
||||
def test_b44_events_path_still_works_with_cutoff(self) -> None:
|
||||
# Without HEL-176 discriminators, /events stays on B-44 ledger list.
|
||||
status, _, data = self.cashier_a.get(
|
||||
"/api/company/intercompany/events?from=2026-01-01&cutoff=2026-12-31"
|
||||
)
|
||||
self.assertEqual(200, status, data)
|
||||
payload = as_json(data)
|
||||
self.assertIn("items", payload)
|
||||
self.assertNotIn("events", payload)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Export: confirmed only + audit + isolation
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def test_export_confirmed_only_isolated_and_audited(self) -> None:
|
||||
self._seed_mixed()
|
||||
|
||||
status, headers, data = self._export(self.cashier_a)
|
||||
self.assertEqual(200, status, data)
|
||||
self.assertEqual("text/csv; charset=utf-8", headers.get("content-type"))
|
||||
text = data.decode("utf-8-sig")
|
||||
lines = [line for line in text.splitlines() if line]
|
||||
self.assertGreaterEqual(len(lines), 2)
|
||||
body = "\n".join(lines[1:])
|
||||
self.assertIn("100.00", body)
|
||||
self.assertIn("40.00", body)
|
||||
self.assertIn("25.00", body)
|
||||
self.assertNotIn("7.00", body) # pending excluded
|
||||
self.assertNotIn("200.00", body) # B↔C excluded
|
||||
|
||||
# Explicit pending state rejected
|
||||
status, _, data = self._export(
|
||||
self.cashier_a, "from=2026-01-01&to=2026-12-31&state=pending"
|
||||
)
|
||||
self.assertEqual(400, status, data)
|
||||
|
||||
# C export must not contain A's amounts
|
||||
status, _, data = self._export(self.cashier_c)
|
||||
self.assertEqual(200, status, data)
|
||||
text_c = data.decode("utf-8-sig")
|
||||
self.assertNotIn("100.00", text_c)
|
||||
self.assertNotIn("25.00", text_c)
|
||||
self.assertIn("200.00", text_c)
|
||||
|
||||
status, _, data = self.admin.get("/api/admin/audit-log?limit=50")
|
||||
self.assertEqual(200, status, data)
|
||||
actions = [row["action"] for row in as_json(data)["entries"]]
|
||||
self.assertIn("export_intercompany_csv", actions)
|
||||
export_rows = [
|
||||
row
|
||||
for row in as_json(data)["entries"]
|
||||
if row["action"] == "export_intercompany_csv"
|
||||
]
|
||||
self.assertTrue(export_rows)
|
||||
self.assertTrue(
|
||||
any(f"company:{self.company_a}" == row.get("target") for row in export_rows)
|
||||
)
|
||||
self.assertTrue(any("state:confirmed" in (row.get("detail") or "") for row in export_rows))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user