551 lines
23 KiB
Python
551 lines
23 KiB
Python
from __future__ import annotations
|
|
|
|
from decimal import Decimal
|
|
import hashlib
|
|
import sqlite3
|
|
import tempfile
|
|
from pathlib import Path
|
|
import unittest
|
|
|
|
from openpyxl import Workbook
|
|
|
|
from bank_importer.db import applied_versions, connect, migrate, rollback, utc_now
|
|
from bank_importer.importing import import_statement
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SAMPLES = ROOT / "流水模板"
|
|
SAMPLE_FILE = SAMPLES / "中国建设银行账户流水.xls"
|
|
|
|
|
|
class PersistenceTestCase(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.connection = connect(self.db_path)
|
|
self.addCleanup(self.connection.close)
|
|
migrate(self.connection)
|
|
|
|
def import_sample(self) -> object:
|
|
return import_statement(
|
|
self.connection,
|
|
self.storage,
|
|
SAMPLE_FILE.name,
|
|
SAMPLE_FILE.read_bytes(),
|
|
)
|
|
|
|
|
|
class MigrationTests(PersistenceTestCase):
|
|
def test_migrate_creates_schema_and_is_idempotent(self) -> None:
|
|
first = applied_versions(self.connection)
|
|
self.assertEqual([1, 2, 3, 4, 5, 6, 7], first)
|
|
self.assertEqual([], migrate(self.connection))
|
|
self.assertEqual(first, applied_versions(self.connection))
|
|
tables = {
|
|
row["name"]
|
|
for row in self.connection.execute(
|
|
"SELECT name FROM sqlite_master WHERE type = 'table'"
|
|
)
|
|
}
|
|
for table in (
|
|
"companies",
|
|
"bank_accounts",
|
|
"account_aliases",
|
|
"master_data_changes",
|
|
"source_files",
|
|
"import_batches",
|
|
"sheet_batches",
|
|
"sheet_reviews",
|
|
"source_rows",
|
|
"import_exceptions",
|
|
"users",
|
|
"sessions",
|
|
"login_attempts",
|
|
"audit_log",
|
|
"personal_transit_mappings",
|
|
"canonical_transfer_events",
|
|
"transfer_match_decisions",
|
|
"transfer_decision_observations",
|
|
"transfer_decision_participants",
|
|
"transfer_match_candidates",
|
|
"current_transfer_decisions",
|
|
"transfer_observation_claims",
|
|
"manual_records",
|
|
"manual_record_decisions",
|
|
"current_manual_record_decisions",
|
|
"ledger_events",
|
|
"ledger_event_revisions",
|
|
"current_ledger_event_revisions",
|
|
"ledger_event_bank_sources",
|
|
"ledger_event_manual_sources",
|
|
"ledger_subject_suggestions",
|
|
"system_settings",
|
|
"system_setting_changes",
|
|
"reminders",
|
|
"schema_migrations",
|
|
):
|
|
self.assertIn(table, tables)
|
|
|
|
def test_rollback_removes_schema_and_forward_rebuilds_it(self) -> None:
|
|
self.assertEqual([7, 6, 5, 4, 3, 2, 1], rollback(self.connection, 0))
|
|
self.assertEqual([], applied_versions(self.connection))
|
|
remaining = self.connection.execute(
|
|
"SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'source_rows'"
|
|
).fetchone()
|
|
self.assertIsNone(remaining)
|
|
self.assertEqual([1, 2, 3, 4, 5, 6, 7], migrate(self.connection))
|
|
self.assertEqual([1, 2, 3, 4, 5, 6, 7], applied_versions(self.connection))
|
|
|
|
def test_rollback_to_4_keeps_bank_evidence_and_drops_event_layer(self) -> None:
|
|
self.import_sample()
|
|
row_count = self.connection.execute(
|
|
"SELECT COUNT(*) AS n FROM source_rows"
|
|
).fetchone()["n"]
|
|
self.assertGreater(row_count, 0)
|
|
self.assertEqual([7, 6, 5], rollback(self.connection, 4))
|
|
# The pre-migration evidence and schema are untouched.
|
|
self.assertEqual(
|
|
row_count,
|
|
self.connection.execute("SELECT COUNT(*) AS n FROM source_rows").fetchone()["n"],
|
|
)
|
|
self.assertEqual(
|
|
"parsed",
|
|
self.connection.execute(
|
|
"SELECT status FROM import_batches LIMIT 1"
|
|
).fetchone()["status"],
|
|
)
|
|
remaining = self.connection.execute(
|
|
"SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'transfer_match_decisions'"
|
|
).fetchone()
|
|
self.assertIsNone(remaining)
|
|
ledger_remaining = self.connection.execute(
|
|
"SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'ledger_event_revisions'"
|
|
).fetchone()
|
|
self.assertIsNone(ledger_remaining)
|
|
|
|
def test_event_layer_views_exist_after_migration(self) -> None:
|
|
view = self.connection.execute(
|
|
"SELECT name FROM sqlite_master WHERE type = 'view' AND name = 'eligible_intercompany_events'"
|
|
).fetchone()
|
|
self.assertIsNotNone(view)
|
|
|
|
def test_ledger_layer_views_exist_after_migration(self) -> None:
|
|
view = self.connection.execute(
|
|
"SELECT name FROM sqlite_master WHERE type = 'view' AND name = 'eligible_position_events'"
|
|
).fetchone()
|
|
self.assertIsNotNone(view)
|
|
table = self.connection.execute(
|
|
"SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'ledger_event_revisions'"
|
|
).fetchone()
|
|
self.assertIsNotNone(table)
|
|
|
|
def test_ledger_revision_log_is_immutable(self) -> None:
|
|
with self.connection:
|
|
self.connection.execute(
|
|
"INSERT INTO companies (name, created_at, updated_at) VALUES ('甲公司', ?, ?)",
|
|
(utc_now(), utc_now()),
|
|
)
|
|
self.connection.execute(
|
|
"INSERT INTO companies (name, created_at, updated_at) VALUES ('乙公司', ?, ?)",
|
|
(utc_now(), utc_now()),
|
|
)
|
|
company_a, company_b = [
|
|
row["id"]
|
|
for row in self.connection.execute("SELECT id FROM companies ORDER BY id")
|
|
]
|
|
self.connection.execute(
|
|
"INSERT INTO ledger_events (lifecycle, created_at) VALUES ('active', ?)",
|
|
(utc_now(),),
|
|
)
|
|
event_id = self.connection.execute(
|
|
"SELECT id FROM ledger_events LIMIT 1"
|
|
).fetchone()["id"]
|
|
self.connection.execute(
|
|
"""
|
|
INSERT INTO ledger_event_revisions (
|
|
ledger_event_id, revision, state, effective_at, amount,
|
|
amount_scale, currency, payer_company_id, payee_company_id,
|
|
source_kind, posting_kind, created_at
|
|
) VALUES (?, 1, 'pending_subject', '2026-01-01T00:00:00', '1.00', 2,
|
|
'CNY', ?, ?, 'bank', 'normal', ?)
|
|
""",
|
|
(event_id, company_a, company_b, utc_now()),
|
|
)
|
|
for statement in (
|
|
"UPDATE ledger_event_revisions SET amount = '9.99'",
|
|
"DELETE FROM ledger_event_revisions",
|
|
"UPDATE ledger_events SET created_at = '2020-01-01T00:00:00'",
|
|
"DELETE FROM ledger_events",
|
|
):
|
|
with self.subTest(statement=statement):
|
|
with self.assertRaises(sqlite3.IntegrityError):
|
|
self.connection.execute(statement)
|
|
self.connection.rollback()
|
|
|
|
def test_manual_record_facts_are_immutable(self) -> None:
|
|
with self.connection:
|
|
self.connection.execute(
|
|
"INSERT INTO companies (name, created_at, updated_at) VALUES ('甲公司', ?, ?)",
|
|
(utc_now(), utc_now()),
|
|
)
|
|
self.connection.execute(
|
|
"INSERT INTO companies (name, created_at, updated_at) VALUES ('乙公司', ?, ?)",
|
|
(utc_now(), utc_now()),
|
|
)
|
|
company_a, company_b = [
|
|
row["id"]
|
|
for row in self.connection.execute("SELECT id FROM companies ORDER BY id")
|
|
]
|
|
self.connection.execute(
|
|
"""
|
|
INSERT INTO manual_records (
|
|
company_id, counterparty_company_id, occurred_at, direction,
|
|
amount, amount_scale, currency, funding_source, requested_subject,
|
|
request_key, created_at
|
|
) VALUES (?, ?, '2026-01-01T00:00:00', 'incoming', '1.00', 2,
|
|
'CNY', 'other', 'receivable', 'k1', ?)
|
|
""",
|
|
(company_a, company_b, utc_now()),
|
|
)
|
|
for statement in (
|
|
"UPDATE manual_records SET amount = '9.99'",
|
|
"DELETE FROM manual_records",
|
|
):
|
|
with self.subTest(statement=statement):
|
|
with self.assertRaises(sqlite3.IntegrityError):
|
|
self.connection.execute(statement)
|
|
self.connection.rollback()
|
|
|
|
def test_decision_log_immutability_triggers(self) -> None:
|
|
self.import_sample()
|
|
source_row_id = self.connection.execute(
|
|
"SELECT id FROM source_rows LIMIT 1"
|
|
).fetchone()["id"]
|
|
with self.connection:
|
|
self.connection.execute(
|
|
"INSERT INTO canonical_transfer_events (lifecycle, created_at) VALUES ('active', ?)",
|
|
(utc_now(),),
|
|
)
|
|
event_id = self.connection.execute(
|
|
"SELECT id FROM canonical_transfer_events LIMIT 1"
|
|
).fetchone()["id"]
|
|
self.connection.execute(
|
|
"""
|
|
INSERT INTO transfer_match_decisions (
|
|
event_id, revision, classification, pairing, mode, rule_version,
|
|
locked, created_at
|
|
) VALUES (?, 1, 'unresolved', 'not_applicable', 'auto', 'transfer-match-v1', 0, ?)
|
|
""",
|
|
(event_id, utc_now()),
|
|
)
|
|
decision_id = self.connection.execute(
|
|
"SELECT id FROM transfer_match_decisions LIMIT 1"
|
|
).fetchone()["id"]
|
|
self.connection.execute(
|
|
"""
|
|
INSERT INTO transfer_match_candidates (
|
|
decision_id, source_row_id, rule_tier, rule_version, created_at
|
|
) VALUES (?, ?, 'R1', 'transfer-match-v1', ?)
|
|
""",
|
|
(decision_id, source_row_id, utc_now()),
|
|
)
|
|
for statement in (
|
|
"UPDATE transfer_match_decisions SET classification = 'external'",
|
|
"DELETE FROM transfer_match_decisions",
|
|
"UPDATE transfer_match_candidates SET rule_tier = 'M1'",
|
|
"DELETE FROM transfer_match_candidates",
|
|
"DELETE FROM canonical_transfer_events",
|
|
):
|
|
with self.subTest(statement=statement):
|
|
with self.assertRaises(sqlite3.IntegrityError):
|
|
self.connection.execute(statement)
|
|
self.connection.rollback()
|
|
|
|
def test_observation_claims_source_row_is_unique(self) -> None:
|
|
self.import_sample()
|
|
source_row_id = self.connection.execute(
|
|
"SELECT id FROM source_rows LIMIT 1"
|
|
).fetchone()["id"]
|
|
with self.connection:
|
|
self.connection.execute(
|
|
"INSERT INTO canonical_transfer_events (lifecycle, created_at) VALUES ('active', ?)",
|
|
(utc_now(),),
|
|
)
|
|
event_id = self.connection.execute(
|
|
"SELECT id FROM canonical_transfer_events LIMIT 1"
|
|
).fetchone()["id"]
|
|
self.connection.execute(
|
|
"""
|
|
INSERT INTO transfer_match_decisions (
|
|
event_id, revision, classification, pairing, mode, locked, created_at
|
|
) VALUES (?, 1, 'unresolved', 'not_applicable', 'auto', 0, ?)
|
|
""",
|
|
(event_id, utc_now()),
|
|
)
|
|
decision_id = self.connection.execute(
|
|
"SELECT id FROM transfer_match_decisions LIMIT 1"
|
|
).fetchone()["id"]
|
|
self.connection.execute(
|
|
"""
|
|
INSERT INTO transfer_observation_claims (source_row_id, event_id, decision_id)
|
|
VALUES (?, ?, ?)
|
|
""",
|
|
(source_row_id, event_id, decision_id),
|
|
)
|
|
with self.assertRaises(sqlite3.IntegrityError):
|
|
self.connection.execute(
|
|
"""
|
|
INSERT INTO transfer_observation_claims (source_row_id, event_id, decision_id)
|
|
VALUES (?, ?, ?)
|
|
""",
|
|
(source_row_id, event_id, decision_id),
|
|
)
|
|
self.connection.rollback()
|
|
|
|
|
|
class ImportPersistenceTests(PersistenceTestCase):
|
|
def test_import_persists_batch_sheets_and_rows(self) -> None:
|
|
result = self.import_sample()
|
|
self.assertEqual("parsed", result.status)
|
|
|
|
batch = self.connection.execute(
|
|
"SELECT status FROM import_batches WHERE id = ?", (result.batch_id,)
|
|
).fetchone()
|
|
self.assertEqual("parsed", batch["status"])
|
|
|
|
sheets = self.connection.execute(
|
|
"SELECT * FROM sheet_batches WHERE import_batch_id = ?", (result.batch_id,)
|
|
).fetchall()
|
|
self.assertEqual(1, len(sheets))
|
|
self.assertEqual("中国建设银行", sheets[0]["bank_name"])
|
|
self.assertEqual("ccb-account-detail-v1", sheets[0]["template_id"])
|
|
self.assertEqual(1, sheets[0]["template_version"])
|
|
|
|
rows = self.connection.execute(
|
|
"SELECT * FROM source_rows WHERE sheet_batch_id = ?", (sheets[0]["id"],)
|
|
).fetchall()
|
|
self.assertEqual(sheets[0]["transaction_count"], len(rows))
|
|
|
|
def test_amounts_keep_decimal_precision(self) -> None:
|
|
result = self.import_sample()
|
|
rows = self.connection.execute(
|
|
"""
|
|
SELECT income, expense FROM source_rows
|
|
WHERE sheet_batch_id IN (
|
|
SELECT id FROM sheet_batches WHERE import_batch_id = ?
|
|
)
|
|
""",
|
|
(result.batch_id,),
|
|
).fetchall()
|
|
self.assertTrue(rows)
|
|
for row in rows:
|
|
self.assertEqual(row["income"], str(Decimal(row["income"])))
|
|
self.assertEqual(row["expense"], str(Decimal(row["expense"])))
|
|
|
|
def test_every_row_traces_back_to_file_sheet_and_template(self) -> None:
|
|
result = self.import_sample()
|
|
row = self.connection.execute(
|
|
"""
|
|
SELECT r.source_row, s.sheet_name, s.template_id, s.template_version, f.sha256
|
|
FROM source_rows r
|
|
JOIN sheet_batches s ON s.id = r.sheet_batch_id
|
|
JOIN import_batches b ON b.id = s.import_batch_id
|
|
JOIN source_files f ON f.id = b.source_file_id
|
|
WHERE b.id = ?
|
|
LIMIT 1
|
|
""",
|
|
(result.batch_id,),
|
|
).fetchone()
|
|
self.assertIsNotNone(row)
|
|
self.assertEqual(
|
|
hashlib.sha256(SAMPLE_FILE.read_bytes()).hexdigest(), row["sha256"]
|
|
)
|
|
self.assertEqual("ccb-account-detail-v1", row["template_id"])
|
|
self.assertEqual(1, row["template_version"])
|
|
self.assertGreater(row["source_row"], 0)
|
|
self.assertTrue(row["sheet_name"])
|
|
|
|
def test_source_file_is_stored_immutably_by_content_hash(self) -> None:
|
|
self.import_sample()
|
|
stored = self.connection.execute(
|
|
"SELECT storage_path FROM source_files"
|
|
).fetchone()
|
|
stored_path = Path(stored["storage_path"])
|
|
self.assertTrue(stored_path.is_file())
|
|
self.assertEqual(SAMPLE_FILE.read_bytes(), stored_path.read_bytes())
|
|
self.assertIn(hashlib.sha256(SAMPLE_FILE.read_bytes()).hexdigest(), stored_path.name)
|
|
|
|
def test_data_survives_reconnect(self) -> None:
|
|
result = self.import_sample()
|
|
self.connection.close()
|
|
|
|
reopened = connect(self.db_path)
|
|
self.addCleanup(reopened.close)
|
|
batch = reopened.execute(
|
|
"SELECT status FROM import_batches WHERE id = ?", (result.batch_id,)
|
|
).fetchone()
|
|
self.assertEqual("parsed", batch["status"])
|
|
files = reopened.execute("SELECT COUNT(*) AS n FROM source_files").fetchone()
|
|
self.assertEqual(1, files["n"])
|
|
rows = reopened.execute("SELECT COUNT(*) AS n FROM source_rows").fetchone()
|
|
self.assertGreater(rows["n"], 0)
|
|
|
|
|
|
class IdempotencyTests(PersistenceTestCase):
|
|
def test_duplicate_upload_creates_no_second_facts(self) -> None:
|
|
first = self.import_sample()
|
|
second = self.import_sample()
|
|
|
|
self.assertEqual("parsed", first.status)
|
|
self.assertEqual("duplicate", second.status)
|
|
self.assertEqual(first.batch_id, second.batch_id)
|
|
self.assertTrue(second.duplicate_same_company)
|
|
|
|
files = self.connection.execute("SELECT COUNT(*) AS n FROM source_files").fetchone()
|
|
self.assertEqual(1, files["n"])
|
|
sheets = self.connection.execute("SELECT COUNT(*) AS n FROM sheet_batches").fetchone()
|
|
self.assertEqual(1, sheets["n"])
|
|
rows = self.connection.execute("SELECT COUNT(*) AS n FROM source_rows").fetchone()
|
|
self.assertEqual(first.batches[0].transactions.__len__(), rows["n"])
|
|
|
|
duplicate = self.connection.execute(
|
|
"SELECT status, duplicate_of_id FROM import_batches WHERE status = 'duplicate'"
|
|
).fetchone()
|
|
self.assertEqual(first.batch_id, duplicate["duplicate_of_id"])
|
|
|
|
def test_cross_company_duplicate_never_returns_other_companys_batch(self) -> None:
|
|
now = utc_now()
|
|
with self.connection:
|
|
self.connection.execute(
|
|
"INSERT INTO companies (name, created_at, updated_at) VALUES ('甲公司', ?, ?)",
|
|
(now, now),
|
|
)
|
|
self.connection.execute(
|
|
"INSERT INTO companies (name, created_at, updated_at) VALUES ('乙公司', ?, ?)",
|
|
(now, now),
|
|
)
|
|
company_a, company_b = [
|
|
row["id"]
|
|
for row in self.connection.execute("SELECT id FROM companies ORDER BY id").fetchall()
|
|
]
|
|
|
|
first = import_statement(
|
|
self.connection, self.storage, SAMPLE_FILE.name,
|
|
SAMPLE_FILE.read_bytes(), company_id=company_a,
|
|
)
|
|
self.assertEqual("parsed", first.status)
|
|
|
|
second = import_statement(
|
|
self.connection, self.storage, SAMPLE_FILE.name,
|
|
SAMPLE_FILE.read_bytes(), company_id=company_b,
|
|
)
|
|
self.assertEqual("duplicate", second.status)
|
|
self.assertFalse(second.duplicate_same_company)
|
|
# The returned batch id is the uploader's own duplicate batch, never
|
|
# the other company's original batch.
|
|
self.assertNotEqual(first.batch_id, second.batch_id)
|
|
self.assertNotEqual(second.batch_id, first.batch_id)
|
|
own = self.connection.execute(
|
|
"SELECT company_id FROM import_batches WHERE id = ?", (second.batch_id,)
|
|
).fetchone()
|
|
self.assertEqual(company_b, own["company_id"])
|
|
|
|
def test_repeated_upload_under_a_different_filename_is_still_duplicate(self) -> None:
|
|
first = self.import_sample()
|
|
second = import_statement(
|
|
self.connection,
|
|
self.storage,
|
|
"改名后的流水.xls",
|
|
SAMPLE_FILE.read_bytes(),
|
|
)
|
|
self.assertEqual("duplicate", second.status)
|
|
self.assertEqual(first.batch_id, second.batch_id)
|
|
files = self.connection.execute("SELECT COUNT(*) AS n FROM source_files").fetchone()
|
|
self.assertEqual(1, files["n"])
|
|
|
|
def test_immutability_triggers_block_updates_and_deletes(self) -> None:
|
|
self.import_sample()
|
|
for statement in (
|
|
"UPDATE source_rows SET income = '0'",
|
|
"DELETE FROM source_rows",
|
|
"UPDATE source_files SET sha256 = 'x'",
|
|
"DELETE FROM source_files",
|
|
"UPDATE sheet_batches SET bank_name = 'x'",
|
|
"DELETE FROM sheet_batches",
|
|
):
|
|
with self.subTest(statement=statement):
|
|
with self.assertRaises(sqlite3.IntegrityError):
|
|
self.connection.execute(statement)
|
|
self.connection.rollback()
|
|
|
|
def test_row_level_unique_constraint_blocks_duplicate_rows(self) -> None:
|
|
self.import_sample()
|
|
row = self.connection.execute(
|
|
"SELECT sheet_batch_id, source_row FROM source_rows LIMIT 1"
|
|
).fetchone()
|
|
with self.assertRaises(sqlite3.IntegrityError):
|
|
self.connection.execute(
|
|
"""
|
|
INSERT INTO source_rows (
|
|
sheet_batch_id, source_row, transaction_at, income, expense, created_at
|
|
) VALUES (?, ?, '2026-01-01T00:00:00', '1', '0', '2026-01-01T00:00:00Z')
|
|
""",
|
|
(row["sheet_batch_id"], row["source_row"]),
|
|
)
|
|
self.connection.rollback()
|
|
|
|
|
|
class FailedImportTests(PersistenceTestCase):
|
|
def _unknown_template_bytes(self) -> bytes:
|
|
workbook = Workbook()
|
|
sheet = workbook.active
|
|
sheet.title = "流水"
|
|
sheet.append(["日期", "金额", "备注"])
|
|
sheet.append(["2026-01-01", "100.00", "测试"])
|
|
target = Path(self.temp_dir.name) / "unknown.xlsx"
|
|
workbook.save(target)
|
|
return target.read_bytes()
|
|
|
|
def test_parse_failure_keeps_exception_batch_without_rows(self) -> None:
|
|
result = import_statement(
|
|
self.connection, self.storage, "未知银行.xlsx", self._unknown_template_bytes()
|
|
)
|
|
self.assertEqual("exception", result.status)
|
|
self.assertIsNotNone(result.message)
|
|
self.assertIn("未知银行.xlsx", result.message)
|
|
self.assertNotIn(result.sha256, result.message)
|
|
|
|
batch = self.connection.execute(
|
|
"SELECT status FROM import_batches WHERE id = ?", (result.batch_id,)
|
|
).fetchone()
|
|
self.assertEqual("exception", batch["status"])
|
|
|
|
exceptions = self.connection.execute(
|
|
"SELECT * FROM import_exceptions WHERE import_batch_id = ?",
|
|
(result.batch_id,),
|
|
).fetchall()
|
|
self.assertEqual(1, len(exceptions))
|
|
self.assertEqual("parse", exceptions[0]["stage"])
|
|
|
|
rows = self.connection.execute("SELECT COUNT(*) AS n FROM source_rows").fetchone()
|
|
self.assertEqual(0, rows["n"])
|
|
|
|
# The source file is still preserved as evidence for later diagnosis.
|
|
files = self.connection.execute("SELECT COUNT(*) AS n FROM source_files").fetchone()
|
|
self.assertEqual(1, files["n"])
|
|
|
|
def test_reupload_after_failure_is_tracked_as_duplicate(self) -> None:
|
|
content = self._unknown_template_bytes()
|
|
first = import_statement(self.connection, self.storage, "未知银行.xlsx", content)
|
|
second = import_statement(self.connection, self.storage, "未知银行.xlsx", content)
|
|
self.assertEqual("exception", first.status)
|
|
self.assertEqual("duplicate", second.status)
|
|
self.assertEqual(first.batch_id, second.batch_id)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|