2fb1f2b462
翻转 _pair_net_change 符号，与往来层/页面口径一致；同步修正旧断言并新增期初200+转出100=期末300的双边守恒与浏览器方向冒烟。 Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: multica-agent <github@multica.ai>
594 lines
23 KiB
Python
594 lines
23 KiB
Python
"""Tests for calculation window: start date, opening balances, coverage gaps."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
import json
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
|
|
from bank_importer import auth, calculation, matching, master_data
|
|
from bank_importer.db import connect, migrate, utc_now
|
|
|
|
import server
|
|
from test_server_auth import Client, as_json
|
|
|
|
|
|
class CalculationBase(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.temp_dir = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp_dir.cleanup)
|
|
self.db_path = Path(self.temp_dir.name) / "app.db"
|
|
self.connection = connect(self.db_path)
|
|
self.addCleanup(self.connection.close)
|
|
migrate(self.connection)
|
|
self.admin = self._admin()
|
|
self.company_a = self._company("甲公司")
|
|
self.company_b = self._company("乙公司")
|
|
self.account_a = self._approved_account(self.company_a, "6222000000000001")
|
|
self.account_b = self._approved_account(self.company_b, "6222000000000002")
|
|
|
|
def _admin(self):
|
|
auth.create_user(self.connection, "admin-u", "AdminPass123", "admin")
|
|
return self.connection.execute(
|
|
"SELECT * FROM users WHERE username = 'admin-u'"
|
|
).fetchone()
|
|
|
|
def _company(self, name: str) -> int:
|
|
with self.connection:
|
|
cursor = self.connection.execute(
|
|
"INSERT INTO companies (name, created_at, updated_at) VALUES (?, ?, ?)",
|
|
(name, utc_now(), utc_now()),
|
|
)
|
|
return int(cursor.lastrowid)
|
|
|
|
def _approved_account(self, company_id: int, number: str, start: str = "2026-01-01"):
|
|
account = master_data.submit_bank_account(
|
|
self.connection,
|
|
company_id=company_id,
|
|
bank_name="中信银行",
|
|
account_type="基本户",
|
|
account_number=number,
|
|
start_date=start,
|
|
actor=None,
|
|
)
|
|
return master_data.review_bank_account(
|
|
self.connection, account["id"], "approve", None, self.admin,
|
|
effective_from=start,
|
|
)
|
|
|
|
def add_confirmed_row(
|
|
self,
|
|
company_id: int,
|
|
*,
|
|
account_id: int,
|
|
own_account: str,
|
|
at: str,
|
|
income: str = "0",
|
|
expense: str = "0",
|
|
cp_account: str | None = None,
|
|
sheet: str = "流水",
|
|
source_row: int = 1,
|
|
) -> int:
|
|
with self.connection:
|
|
cursor = self.connection.execute(
|
|
"""
|
|
INSERT INTO source_files (sha256, original_filename, size_bytes, storage_path, created_at)
|
|
VALUES (?, '测试.xlsx', 1, 'data/files/测试.xlsx', ?)
|
|
""",
|
|
(f"sha-{at}-{own_account}-{source_row}", utc_now()),
|
|
)
|
|
source_file_id = int(cursor.lastrowid)
|
|
cursor = self.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 = self.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, sheet, utc_now()),
|
|
)
|
|
sheet_batch_id = int(cursor.lastrowid)
|
|
self.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, sheet_batch_id, utc_now()),
|
|
)
|
|
cursor = self.connection.execute(
|
|
"""
|
|
INSERT INTO source_rows (
|
|
sheet_batch_id, source_row, transaction_at, income, expense,
|
|
own_account, counterparty_account, created_at
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(sheet_batch_id, source_row, at, income, expense, own_account, cp_account, utc_now()),
|
|
)
|
|
return int(cursor.lastrowid)
|
|
|
|
def add_confirmed_batch_range(
|
|
self,
|
|
company_id: int,
|
|
*,
|
|
account_id: int,
|
|
own_account: str,
|
|
start: str,
|
|
end: str,
|
|
sheet: str,
|
|
) -> None:
|
|
with self.connection:
|
|
cursor = self.connection.execute(
|
|
"""
|
|
INSERT INTO source_files (sha256, original_filename, size_bytes, storage_path, created_at)
|
|
VALUES (?, '测试.xlsx', 1, 'data/files/测试.xlsx', ?)
|
|
""",
|
|
(f"sha-batch-{sheet}", utc_now()),
|
|
)
|
|
source_file_id = int(cursor.lastrowid)
|
|
cursor = self.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 = self.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, 2, '[]', ?)
|
|
""",
|
|
(batch_id, sheet, utc_now()),
|
|
)
|
|
sheet_batch_id = int(cursor.lastrowid)
|
|
self.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, sheet_batch_id, utc_now()),
|
|
)
|
|
for source_row, day in ((1, start), (2, end)):
|
|
self.connection.execute(
|
|
"""
|
|
INSERT INTO source_rows (
|
|
sheet_batch_id, source_row, transaction_at, income, expense,
|
|
own_account, created_at
|
|
) VALUES (?, ?, ?, '0', '0', ?, ?)
|
|
""",
|
|
(sheet_batch_id, source_row, f"{day}T10:00:00", own_account, utc_now()),
|
|
)
|
|
|
|
|
|
class StartDateTests(CalculationBase):
|
|
def test_set_start_date_records_change(self) -> None:
|
|
result = calculation.set_calculation_start_date(
|
|
self.connection, "2026-01-01", "首次设定", self.admin
|
|
)
|
|
self.assertEqual("2026-01-01", result["calculation_start_date"])
|
|
row = self.connection.execute(
|
|
"SELECT 1 FROM master_data_changes WHERE entity_type = 'system_setting'"
|
|
).fetchone()
|
|
self.assertIsNotNone(row)
|
|
|
|
def test_locked_after_closed_period(self) -> None:
|
|
calculation.set_calculation_start_date(
|
|
self.connection, "2026-01-01", "首次设定", self.admin
|
|
)
|
|
with self.connection:
|
|
self.connection.execute(
|
|
"INSERT INTO closed_periods (year_month, closed_at, closed_by) VALUES ('2026-01', ?, ?)",
|
|
(utc_now(), self.admin["id"]),
|
|
)
|
|
with self.assertRaises(calculation.LockedError):
|
|
calculation.set_calculation_start_date(
|
|
self.connection, "2026-02-01", "尝试修改", self.admin
|
|
)
|
|
|
|
|
|
class OpeningBalanceTests(CalculationBase):
|
|
def setUp(self) -> None:
|
|
super().setUp()
|
|
calculation.set_calculation_start_date(
|
|
self.connection, "2026-01-01", "测试起算日", self.admin
|
|
)
|
|
|
|
def test_bilateral_conservation_on_storage(self) -> None:
|
|
item = calculation.create_opening_balance(
|
|
self.connection,
|
|
self.company_a,
|
|
self.company_b,
|
|
"100.00",
|
|
"期初录入",
|
|
self.admin,
|
|
viewer_company_id=self.company_a,
|
|
)
|
|
low, high = calculation.normalize_pair(self.company_a, self.company_b)
|
|
self.assertEqual(low, item["company_id_low"])
|
|
stored = calculation.confirmed_opening_amount(self.connection, low, high)
|
|
self.assertIsNone(stored)
|
|
calculation.confirm_opening_balance(
|
|
self.connection, item["id"], "确认期初", self.admin
|
|
)
|
|
stored = calculation.confirmed_opening_amount(self.connection, low, high)
|
|
self.assertEqual(Decimal("100.00"), stored)
|
|
from_b = calculation.signed_from_viewer(self.company_b, low, high, stored)
|
|
self.assertEqual(Decimal("-100.00"), from_b)
|
|
|
|
def test_confirmed_requires_revision_not_overwrite(self) -> None:
|
|
item = calculation.create_opening_balance(
|
|
self.connection, self.company_a, self.company_b, "50", "录入", self.admin
|
|
)
|
|
calculation.confirm_opening_balance(
|
|
self.connection, item["id"], "确认", self.admin
|
|
)
|
|
with self.assertRaises(calculation.ConflictError):
|
|
calculation.create_opening_balance(
|
|
self.connection, self.company_a, self.company_b, "80", "重复录入", self.admin
|
|
)
|
|
revised = calculation.revise_opening_balance(
|
|
self.connection, item["id"], "80", "修订", self.admin
|
|
)
|
|
self.assertEqual("draft", revised["status"])
|
|
|
|
|
|
class CoverageGapTests(CalculationBase):
|
|
def setUp(self) -> None:
|
|
super().setUp()
|
|
calculation.set_calculation_start_date(
|
|
self.connection, "2026-06-01", "起算", self.admin
|
|
)
|
|
|
|
def test_adjacent_intervals_no_mid_gap(self) -> None:
|
|
self.add_confirmed_batch_range(
|
|
self.company_a,
|
|
account_id=self.account_a["id"],
|
|
own_account="6222000000000001",
|
|
start="2026-06-21",
|
|
end="2026-07-21",
|
|
sheet="批次A",
|
|
)
|
|
self.add_confirmed_batch_range(
|
|
self.company_a,
|
|
account_id=self.account_a["id"],
|
|
own_account="6222000000000001",
|
|
start="2026-07-22",
|
|
end="2026-08-21",
|
|
sheet="批次B",
|
|
)
|
|
calculation.recalculate_coverage_gaps(self.connection)
|
|
mids = self.connection.execute(
|
|
"SELECT * FROM coverage_gaps WHERE gap_kind = 'mid'"
|
|
).fetchall()
|
|
self.assertEqual([], mids)
|
|
|
|
def test_missing_day_mid_gap(self) -> None:
|
|
self.add_confirmed_batch_range(
|
|
self.company_a,
|
|
account_id=self.account_a["id"],
|
|
own_account="6222000000000001",
|
|
start="2026-06-21",
|
|
end="2026-07-21",
|
|
sheet="批次A",
|
|
)
|
|
self.add_confirmed_batch_range(
|
|
self.company_a,
|
|
account_id=self.account_a["id"],
|
|
own_account="6222000000000001",
|
|
start="2026-07-23",
|
|
end="2026-08-21",
|
|
sheet="批次B",
|
|
)
|
|
calculation.recalculate_coverage_gaps(self.connection)
|
|
gap = self.connection.execute(
|
|
"""
|
|
SELECT gap_start, gap_end FROM coverage_gaps
|
|
WHERE gap_kind = 'mid' AND gap_start = '2026-07-22'
|
|
"""
|
|
).fetchone()
|
|
self.assertIsNotNone(gap)
|
|
self.assertEqual("2026-07-22", gap["gap_end"])
|
|
|
|
def test_attestation_closes_gap_without_bank_row(self) -> None:
|
|
self.add_confirmed_row(
|
|
self.company_a,
|
|
account_id=self.account_a["id"],
|
|
own_account="6222000000000001",
|
|
at="2026-06-21T10:00:00",
|
|
)
|
|
calculation.recalculate_coverage_gaps(self.connection)
|
|
gap = self.connection.execute(
|
|
"SELECT * FROM coverage_gaps WHERE status = 'open'"
|
|
).fetchone()
|
|
self.assertIsNotNone(gap)
|
|
before_rows = self.connection.execute("SELECT COUNT(*) AS n FROM source_rows").fetchone()["n"]
|
|
cashier_id = auth.create_user(
|
|
self.connection, "cashier-a", "CashierA123", "company", self.company_a
|
|
)
|
|
cashier = self.connection.execute(
|
|
"SELECT * FROM users WHERE id = ?", (cashier_id,)
|
|
).fetchone()
|
|
att = calculation.submit_no_business_attestation(
|
|
self.connection,
|
|
company_id=self.company_a,
|
|
bank_account_id=self.account_a["id"],
|
|
gap_start=gap["gap_start"],
|
|
gap_end=gap["gap_end"],
|
|
reason="当日账户无资金往来",
|
|
evidence=None,
|
|
actor=cashier,
|
|
)
|
|
calculation.review_no_business_attestation(
|
|
self.connection, att["id"], "approve", "审核通过", self.admin
|
|
)
|
|
after_rows = self.connection.execute("SELECT COUNT(*) AS n FROM source_rows").fetchone()["n"]
|
|
self.assertEqual(before_rows, after_rows)
|
|
closed = self.connection.execute(
|
|
"SELECT status FROM coverage_gaps WHERE id = ?", (gap["id"],)
|
|
).fetchone()
|
|
self.assertEqual("closed_attested", closed["status"])
|
|
|
|
|
|
class BalanceBasisTests(CalculationBase):
|
|
def setUp(self) -> None:
|
|
super().setUp()
|
|
calculation.set_calculation_start_date(
|
|
self.connection, "2026-01-01", "起算", self.admin
|
|
)
|
|
|
|
def test_without_opening_returns_net_change(self) -> None:
|
|
row_a = self.add_confirmed_row(
|
|
self.company_a,
|
|
account_id=self.account_a["id"],
|
|
own_account="6222000000000001",
|
|
at="2026-01-05T10:00:00",
|
|
expense="100.00",
|
|
cp_account="6222000000000002",
|
|
)
|
|
row_b = self.add_confirmed_row(
|
|
self.company_b,
|
|
account_id=self.account_b["id"],
|
|
own_account="6222000000000002",
|
|
at="2026-01-05T11:00:00",
|
|
income="100.00",
|
|
cp_account="6222000000000001",
|
|
)
|
|
matching.reconcile_rows(self.connection, [row_a, row_b])
|
|
balance = calculation.compute_pair_balance(
|
|
self.connection, self.company_a, self.company_b, cutoff="2026-01-31"
|
|
)
|
|
self.assertEqual("net_change", balance["basis"])
|
|
self.assertNotIn("closing", balance)
|
|
|
|
def test_with_opening_returns_full_basis(self) -> None:
|
|
item = calculation.create_opening_balance(
|
|
self.connection, self.company_a, self.company_b, "200", "录入", self.admin
|
|
)
|
|
calculation.confirm_opening_balance(
|
|
self.connection, item["id"], "确认", self.admin
|
|
)
|
|
balance = calculation.compute_pair_balance(
|
|
self.connection, self.company_a, self.company_b, cutoff="2026-01-31"
|
|
)
|
|
self.assertEqual("full", balance["basis"])
|
|
self.assertEqual("200", balance["opening"])
|
|
self.assertEqual("200", balance["closing"])
|
|
|
|
def test_pre_start_events_excluded(self) -> None:
|
|
item = calculation.create_opening_balance(
|
|
self.connection, self.company_a, self.company_b, "0", "零期初", self.admin
|
|
)
|
|
calculation.confirm_opening_balance(
|
|
self.connection, item["id"], "确认", self.admin
|
|
)
|
|
row_a = self.add_confirmed_row(
|
|
self.company_a,
|
|
account_id=self.account_a["id"],
|
|
own_account="6222000000000001",
|
|
at="2025-12-31T10:00:00",
|
|
expense="50.00",
|
|
cp_account="6222000000000002",
|
|
)
|
|
row_b = self.add_confirmed_row(
|
|
self.company_b,
|
|
account_id=self.account_b["id"],
|
|
own_account="6222000000000002",
|
|
at="2025-12-31T11:00:00",
|
|
income="50.00",
|
|
cp_account="6222000000000001",
|
|
)
|
|
matching.reconcile_rows(self.connection, [row_a, row_b])
|
|
balance = calculation.compute_pair_balance(
|
|
self.connection, self.company_a, self.company_b, cutoff="2026-01-31"
|
|
)
|
|
self.assertEqual("0", balance["net_change"])
|
|
|
|
def test_cutoff_day_event_is_included(self) -> None:
|
|
"""effective_at with time on the cutoff date must still count."""
|
|
from test_matching import MatchingBase
|
|
|
|
item = calculation.create_opening_balance(
|
|
self.connection, self.company_a, self.company_b, "0", "零期初", self.admin
|
|
)
|
|
calculation.confirm_opening_balance(
|
|
self.connection, item["id"], "确认", self.admin
|
|
)
|
|
helper = object.__new__(MatchingBase)
|
|
helper.connection = self.connection
|
|
row_a = helper.add_row(
|
|
self.company_a,
|
|
own_account="6222000000000001",
|
|
cp_account="6222000000000002",
|
|
expense="80.00",
|
|
at="2026-01-31T10:00:00",
|
|
)
|
|
matching.reconcile_rows(self.connection, [row_a])
|
|
row_b = helper.add_row(
|
|
self.company_b,
|
|
own_account="6222000000000002",
|
|
cp_account="6222000000000001",
|
|
income="80.00",
|
|
at="2026-01-31T11:00:00",
|
|
)
|
|
matching.reconcile_rows(self.connection, [row_b])
|
|
eligible = self.connection.execute(
|
|
"SELECT effective_at, amount FROM eligible_intercompany_events"
|
|
).fetchall()
|
|
self.assertEqual(1, len(eligible))
|
|
self.assertEqual("2026-01-31T10:00:00", eligible[0]["effective_at"])
|
|
# Full-timestamp string compare wrongly excludes the cutoff day.
|
|
self.assertEqual(
|
|
[],
|
|
self.connection.execute(
|
|
"""
|
|
SELECT 1 FROM eligible_intercompany_events
|
|
WHERE effective_at <= '2026-01-31'
|
|
"""
|
|
).fetchall(),
|
|
)
|
|
balance = calculation.compute_pair_balance(
|
|
self.connection, self.company_a, self.company_b, cutoff="2026-01-31"
|
|
)
|
|
# 甲转出 80 → 应收方向为正
|
|
self.assertEqual("80.00", balance["net_change"])
|
|
self.assertEqual("80.00", balance["closing"])
|
|
|
|
def test_opening_plus_outflow_increases_receivable(self) -> None:
|
|
"""期初应收 200 + 本期转出垫付 100 → 期末应收 300；乙方对称为应付 300。"""
|
|
from test_matching import MatchingBase
|
|
|
|
item = calculation.create_opening_balance(
|
|
self.connection, self.company_a, self.company_b, "200", "期初应收", self.admin
|
|
)
|
|
calculation.confirm_opening_balance(
|
|
self.connection, item["id"], "确认", self.admin
|
|
)
|
|
helper = object.__new__(MatchingBase)
|
|
helper.connection = self.connection
|
|
row_a = helper.add_row(
|
|
self.company_a,
|
|
own_account="6222000000000001",
|
|
cp_account="6222000000000002",
|
|
expense="100.00",
|
|
at="2026-01-15T10:00:00",
|
|
)
|
|
matching.reconcile_rows(self.connection, [row_a])
|
|
row_b = helper.add_row(
|
|
self.company_b,
|
|
own_account="6222000000000002",
|
|
cp_account="6222000000000001",
|
|
income="100.00",
|
|
at="2026-01-15T11:00:00",
|
|
)
|
|
matching.reconcile_rows(self.connection, [row_b])
|
|
|
|
bal_a = calculation.compute_pair_balance(
|
|
self.connection, self.company_a, self.company_b, cutoff="2026-01-31"
|
|
)
|
|
self.assertEqual("full", bal_a["basis"])
|
|
self.assertEqual("200", bal_a["opening"])
|
|
self.assertEqual("100.00", bal_a["net_change"])
|
|
self.assertEqual("300.00", bal_a["closing"])
|
|
|
|
bal_b = calculation.compute_pair_balance(
|
|
self.connection, self.company_b, self.company_a, cutoff="2026-01-31"
|
|
)
|
|
self.assertEqual("full", bal_b["basis"])
|
|
self.assertEqual("-200", bal_b["opening"])
|
|
self.assertEqual("-100.00", bal_b["net_change"])
|
|
self.assertEqual("-300.00", bal_b["closing"])
|
|
# 双边守恒
|
|
self.assertEqual(
|
|
Decimal(bal_a["closing"]) + Decimal(bal_b["closing"]),
|
|
Decimal("0"),
|
|
)
|
|
|
|
|
|
class CalculationApiTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.temp_dir = tempfile.TemporaryDirectory()
|
|
cls.db_path = Path(cls.temp_dir.name) / "app.db"
|
|
cls.storage_dir = Path(cls.temp_dir.name) / "files"
|
|
cls.storage_dir.mkdir()
|
|
server.DB_PATH = cls.db_path
|
|
server.STORAGE_DIR = cls.storage_dir
|
|
connection = connect(cls.db_path)
|
|
migrate(connection)
|
|
auth.create_user(
|
|
connection, "group-admin", "AdminPass123", "admin",
|
|
must_change_password=False,
|
|
)
|
|
company_id = master_data.create_company(connection, "甲公司", None, None, None)
|
|
auth.create_user(
|
|
connection, "cashier-a", "CashierA123", "company", company_id,
|
|
must_change_password=False,
|
|
)
|
|
connection.close()
|
|
cls.httpd = server.ThreadingHTTPServer(("127.0.0.1", 0), server.AppHandler)
|
|
cls.port = cls.httpd.server_address[1]
|
|
cls.thread = threading.Thread(target=cls.httpd.serve_forever, daemon=True)
|
|
cls.thread.start()
|
|
cls.client = Client("127.0.0.1", cls.port)
|
|
|
|
@classmethod
|
|
def tearDownClass(cls) -> None:
|
|
cls.httpd.shutdown()
|
|
cls.temp_dir.cleanup()
|
|
|
|
def test_company_cannot_call_admin_start_date(self) -> None:
|
|
self.client.post_json("/api/login", {
|
|
"username": "cashier-a", "password": "CashierA123", "portal": "company",
|
|
})
|
|
status, _, _ = self.client.request(
|
|
"PUT",
|
|
"/api/admin/settings/calculation-start",
|
|
body=json.dumps(
|
|
{"calculation_start_date": "2026-01-01", "reason": "越权"}
|
|
).encode("utf-8"),
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
self.assertEqual(403, status)
|
|
|
|
def test_admin_can_set_start_date(self) -> None:
|
|
client = Client("127.0.0.1", self.port)
|
|
client.post_json("/api/login", {
|
|
"username": "group-admin", "password": "AdminPass123", "portal": "admin",
|
|
})
|
|
status, _, body = client.request(
|
|
"PUT",
|
|
"/api/admin/settings/calculation-start",
|
|
body=json.dumps(
|
|
{"calculation_start_date": "2026-01-01", "reason": "初始化"}
|
|
).encode("utf-8"),
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
self.assertEqual(200, status)
|
|
data = as_json(body)
|
|
self.assertEqual("2026-01-01", data["calculation_start_date"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|