zkxw/backend/tests/test_biz_no_generator.py

"""biz_no_generator 单元测试"""
import unittest
from unittest.mock import MagicMock, patch

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.models.base import Base
from app.models.models import BizBizNoSequence

from app.crud.biz_no_generator import (
    MAX_SEQ,
    PREFIX_COLLECTION,
    PREFIX_DONATION,
    _format_biz_no,
    _prefix_for,
)


class TestBizNoGenerator(unittest.TestCase):
    def test_prefix_for(self):
        self.assertEqual(_prefix_for("collection"), PREFIX_COLLECTION)
        self.assertEqual(_prefix_for("donation"), PREFIX_DONATION)

    def test_format_biz_no(self):
        self.assertEqual(_format_biz_no("XBCC", "20260527", 1), "XBCC20260527000001")
        self.assertEqual(_format_biz_no("JZCC", "20260527", 42), "JZCC20260527000042")

    def test_max_seq(self):
        self.assertEqual(MAX_SEQ, 999999)


class TestAllocateNextBizNo(unittest.TestCase):
    def test_sqlite_fallback_allocates_incrementing_numbers(self):
        from app.crud import biz_no_generator as gen

        engine = create_engine('sqlite:///:memory:')
        Base.metadata.create_all(engine, tables=[BizBizNoSequence.__table__])
        db = sessionmaker(bind=engine)()
        with patch.object(gen, 'assert_biz_no_available'):
            first = gen.allocate_next_biz_no(
                db,
                tenant_no='zkxw',
                kind='collection',
                biz_date=gen.datetime(2026, 10, 13),
            )
            second = gen.allocate_next_biz_no(
                db,
                tenant_no='zkxw',
                kind='collection',
                biz_date=gen.datetime(2026, 10, 13),
            )

        self.assertEqual(first, 'XBCC20261013000001')
        self.assertEqual(second, 'XBCC20261013000002')

    def test_allocate_raises_when_seq_exhausted(self):
        from app.crud import biz_no_generator as gen

        db = MagicMock()
        db.execute.side_effect = [
            None,
            MagicMock(scalar=lambda: MAX_SEQ + 1),
        ]
        with self.assertRaises(ValueError) as ctx:
            gen.allocate_next_biz_no(db, tenant_no="zkxw", kind="collection")
        self.assertIn("序号已用尽", str(ctx.exception))


if __name__ == "__main__":
    unittest.main()