217 lines
9.0 KiB
Python
217 lines
9.0 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import sqlite3
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from runtime_config import DB_PATH
|
|
|
|
|
|
CASES = (
|
|
("01-01", "50004", "11-20250131-020161-3", {"DIRECT_MATCH_CANDIDATE": 3, "SETTLEMENT_BRIDGE_CANDIDATE": 1}),
|
|
("01-05", "50003", "11-20250109-B0100-5", {"DIRECT_MATCH_CANDIDATE": 1, "SHARED_ALLOCATION_CANDIDATE": 2}),
|
|
("01-05", "50004", "11-20250109-B0100-5", {"DIRECT_MATCH_CANDIDATE": 1, "SHARED_ALLOCATION_CANDIDATE": 2}),
|
|
("01-07", "50013", "11-20250131-017128-1", {"DIRECT_MATCH_CANDIDATE": 2, "SETTLEMENT_BRIDGE_CANDIDATE": 1}),
|
|
("01-10", "00036", "11-20250110-B0100-15", {"DIRECT_MATCH_CANDIDATE": 1}),
|
|
("01-10", "00048", "11-20250110-B0100-13", {"DIRECT_MATCH_CANDIDATE": 1}),
|
|
("01-10", "00056", "11-20250110-B0100-1", {"DIRECT_MATCH_CANDIDATE": 1, "SHARED_BUNDLE_PAYMENT_CANDIDATE": 1}),
|
|
("01-10", "00058", "11-20250110-B0100-1", {"DIRECT_MATCH_CANDIDATE": 1, "SHARED_BUNDLE_PAYMENT_CANDIDATE": 1}),
|
|
("01-10", "00059", "11-20250110-B0100-1", {"DIRECT_MATCH_CANDIDATE": 1, "SHARED_BUNDLE_PAYMENT_CANDIDATE": 1}),
|
|
("01-10", "00061", "11-20241223-J0100-3", {"DIRECT_MATCH_CANDIDATE": 1, "SETTLEMENT_BRIDGE_CANDIDATE": 1}),
|
|
)
|
|
|
|
|
|
def latest_signature(conn: sqlite3.Connection, start_year: int, end_year: int) -> str:
|
|
row = conn.execute(
|
|
"""
|
|
SELECT signature
|
|
FROM wehago_compare_query_groups
|
|
WHERE start_year = ?
|
|
AND end_year = ?
|
|
AND status_key = 'voucher_recheck'
|
|
GROUP BY signature
|
|
ORDER BY MAX(updated_at) DESC
|
|
LIMIT 1
|
|
""",
|
|
(start_year, end_year),
|
|
).fetchone()
|
|
return str(row[0] or "") if row else ""
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Validate fixed WEHAGO Recheck regression cases and quality gates.")
|
|
parser.add_argument("--start-year", type=int, default=2025)
|
|
parser.add_argument("--end-year", type=int, default=2025)
|
|
parser.add_argument("--signature", default="")
|
|
args = parser.parse_args()
|
|
|
|
conn = sqlite3.connect(DB_PATH)
|
|
conn.row_factory = sqlite3.Row
|
|
signature = args.signature or latest_signature(conn, args.start_year, args.end_year)
|
|
if not signature:
|
|
raise SystemExit("No voucher_recheck projection signature found.")
|
|
|
|
failures: list[str] = []
|
|
results: list[dict[str, object]] = []
|
|
for ledger_date, voucher_no, expected_base, expected_roles in CASES:
|
|
group = conn.execute(
|
|
"""
|
|
SELECT group_index, draft_no, review_reason
|
|
FROM wehago_compare_query_groups
|
|
WHERE start_year = ?
|
|
AND end_year = ?
|
|
AND status_key = 'voucher_recheck'
|
|
AND signature = ?
|
|
AND ledger_date = ?
|
|
AND voucher_no = ?
|
|
LIMIT 1
|
|
""",
|
|
(args.start_year, args.end_year, signature, ledger_date, voucher_no),
|
|
).fetchone()
|
|
case_key = f"{args.start_year}-{ledger_date} {voucher_no}"
|
|
if not group:
|
|
failures.append(f"{case_key}: projection group missing")
|
|
continue
|
|
shown_bases = {part.strip() for part in str(group["draft_no"] or "").split(",") if part.strip()}
|
|
if shown_bases != {expected_base}:
|
|
failures.append(f"{case_key}: draft bases {sorted(shown_bases)} != [{expected_base}]")
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT draft_no, ledger_account_name, voucher_account_name,
|
|
ledger_debit, ledger_credit, voucher_debit, voucher_credit, matched_case
|
|
FROM wehago_compare_query_rows
|
|
WHERE start_year = ?
|
|
AND end_year = ?
|
|
AND status_key = 'voucher_recheck'
|
|
AND signature = ?
|
|
AND group_index = ?
|
|
ORDER BY row_index
|
|
""",
|
|
(args.start_year, args.end_year, signature, int(group["group_index"])),
|
|
).fetchall()
|
|
role_counts: dict[str, int] = {}
|
|
displayed_erp_rows = []
|
|
for row in rows:
|
|
role = str(row["matched_case"] or "")
|
|
role_counts[role] = role_counts.get(role, 0) + 1
|
|
has_erp = bool(
|
|
str(row["voucher_account_name"] or "")
|
|
or abs(float(row["voucher_debit"] or 0)) >= 0.5
|
|
or abs(float(row["voucher_credit"] or 0)) >= 0.5
|
|
)
|
|
if has_erp:
|
|
displayed_erp_rows.append(row)
|
|
if role == "DIRECT_MATCH_CANDIDATE":
|
|
ledger_amount = max(abs(float(row["ledger_debit"] or 0)), abs(float(row["ledger_credit"] or 0)))
|
|
voucher_amount = max(abs(float(row["voucher_debit"] or 0)), abs(float(row["voucher_credit"] or 0)))
|
|
if ledger_amount < 0.5 or voucher_amount < 0.5:
|
|
failures.append(f"{case_key}: zero/one-sided direct match at {row['draft_no']}")
|
|
for role, minimum in expected_roles.items():
|
|
if role_counts.get(role, 0) < minimum:
|
|
failures.append(f"{case_key}: {role}={role_counts.get(role, 0)} < {minimum}")
|
|
|
|
raw = conn.execute(
|
|
"""
|
|
SELECT COUNT(*) AS row_count,
|
|
COALESCE(SUM(debit_supply), 0) AS debit_sum,
|
|
COALESCE(SUM(credit_supply), 0) AS credit_sum
|
|
FROM wehago_voucher_rows
|
|
WHERE draft_no = ?
|
|
OR draft_no LIKE ?
|
|
""",
|
|
(expected_base, f"{expected_base}-%"),
|
|
).fetchone()
|
|
displayed_debit = sum(float(row["voucher_debit"] or 0) for row in displayed_erp_rows)
|
|
displayed_credit = sum(float(row["voucher_credit"] or 0) for row in displayed_erp_rows)
|
|
if len(displayed_erp_rows) != int(raw["row_count"] or 0):
|
|
failures.append(
|
|
f"{case_key}: ERP rows displayed={len(displayed_erp_rows)} raw={int(raw['row_count'] or 0)}"
|
|
)
|
|
if abs(displayed_debit - float(raw["debit_sum"] or 0)) >= 0.5:
|
|
failures.append(f"{case_key}: ERP debit displayed={displayed_debit} raw={raw['debit_sum']}")
|
|
if abs(displayed_credit - float(raw["credit_sum"] or 0)) >= 0.5:
|
|
failures.append(f"{case_key}: ERP credit displayed={displayed_credit} raw={raw['credit_sum']}")
|
|
results.append(
|
|
{
|
|
"case": case_key,
|
|
"draft_base": expected_base,
|
|
"erp_rows": len(displayed_erp_rows),
|
|
"roles": role_counts,
|
|
"internal_status": str(group["review_reason"] or "").split("INTERNAL_STATUS=")[-1],
|
|
}
|
|
)
|
|
|
|
mixed_count = conn.execute(
|
|
"""
|
|
WITH per_group AS (
|
|
SELECT group_index,
|
|
COUNT(DISTINCT CASE
|
|
WHEN COALESCE(ledger_account_name, '') <> ''
|
|
OR ABS(COALESCE(ledger_debit, 0)) >= 0.5
|
|
OR ABS(COALESCE(ledger_credit, 0)) >= 0.5
|
|
THEN COALESCE(ledger_date, '') || '|' || COALESCE(voucher_no, '')
|
|
END) AS wehago_keys
|
|
FROM wehago_compare_query_rows
|
|
WHERE start_year = ?
|
|
AND end_year = ?
|
|
AND status_key = 'voucher_recheck'
|
|
AND signature = ?
|
|
GROUP BY group_index
|
|
)
|
|
SELECT COUNT(*) FROM per_group WHERE wehago_keys > 1
|
|
""",
|
|
(args.start_year, args.end_year, signature),
|
|
).fetchone()[0]
|
|
if int(mixed_count or 0) != 0:
|
|
failures.append(f"mixed WEHAGO groups={mixed_count}")
|
|
|
|
displayed_draft_nos = {
|
|
str(row[0] or "")
|
|
for row in conn.execute(
|
|
"""
|
|
SELECT DISTINCT draft_no
|
|
FROM wehago_compare_query_rows
|
|
WHERE start_year = ?
|
|
AND end_year = ?
|
|
AND status_key = 'voucher_recheck'
|
|
AND signature = ?
|
|
AND (
|
|
COALESCE(voucher_account_name, '') <> ''
|
|
OR ABS(COALESCE(voucher_debit, 0)) >= 0.5
|
|
OR ABS(COALESCE(voucher_credit, 0)) >= 0.5
|
|
)
|
|
""",
|
|
(args.start_year, args.end_year, signature),
|
|
)
|
|
if str(row[0] or "")
|
|
}
|
|
raw_draft_nos = {
|
|
str(row[0] or "")
|
|
for row in conn.execute(
|
|
"""
|
|
SELECT DISTINCT draft_no
|
|
FROM wehago_voucher_rows
|
|
WHERE fiscal_year BETWEEN ? AND ?
|
|
AND COALESCE(draft_no, '') <> ''
|
|
""",
|
|
(args.start_year - 1, args.end_year + 1),
|
|
)
|
|
}
|
|
confirmed_only_count = len(displayed_draft_nos - raw_draft_nos)
|
|
if int(confirmed_only_count or 0) != 0:
|
|
failures.append(f"ERP rows not backed by raw draft_no={confirmed_only_count}")
|
|
|
|
print({"db": str(DB_PATH), "signature": signature, "cases": results})
|
|
print({"quality_gates": {"mixed_wehago_groups": mixed_count, "confirmed_only_rows": confirmed_only_count}})
|
|
if failures:
|
|
for failure in failures:
|
|
print({"failure": failure})
|
|
raise SystemExit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|