280 lines
9.5 KiB
Python
280 lines
9.5 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import unittest
|
|
|
|
from openpyxl import Workbook
|
|
|
|
from arr_ingestion.contracts import ArtifactRef
|
|
from arr_web.booking_uploads import ProgramBookingSourceCoordinator
|
|
from arr_web.contracts import PortalError
|
|
from booking_ingestion.excel_postgres import BookingSourceSummary
|
|
from booking_ingestion.excel_review import (
|
|
BookingDraftItem,
|
|
BookingDraftPage,
|
|
BookingDraftSummary,
|
|
)
|
|
|
|
|
|
DRAFT_ID = "bookingdraft-" + "a" * 32
|
|
|
|
|
|
def valid_workbook() -> bytes:
|
|
workbook = Workbook()
|
|
worksheet = workbook.active
|
|
worksheet.title = "Booking"
|
|
worksheet.append(["NO", "Tour Code", "Tour Days", "โรงแรม"])
|
|
worksheet.append([1, "SYN-G1", "", "酒店 (【U-TWN8.5】 2,【U-เตียงเสริม13】 1)"])
|
|
output = io.BytesIO()
|
|
workbook.save(output)
|
|
workbook.close()
|
|
return output.getvalue()
|
|
|
|
|
|
def summary(disposition: str = "imported_and_activated") -> BookingSourceSummary:
|
|
return BookingSourceSummary(
|
|
source_batch_id=7,
|
|
source_kind="booking_excel",
|
|
filename="Booking 报表.xlsx",
|
|
source_rows=1,
|
|
worksheet_count=1,
|
|
distinct_group_codes=1,
|
|
room_quantity=3,
|
|
activated_at=None,
|
|
disposition=disposition,
|
|
)
|
|
|
|
|
|
def draft_item(
|
|
*,
|
|
item_id: int = 1,
|
|
room_type: str | None = "U-TWN",
|
|
status: str = "confirmed",
|
|
) -> BookingDraftItem:
|
|
return BookingDraftItem(
|
|
item_id=item_id,
|
|
draft_id=DRAFT_ID,
|
|
worksheet="Booking",
|
|
row_no=2,
|
|
item_no=item_id,
|
|
tour_code="SYN-G1",
|
|
room_type_raw="U-TWN8.5" if item_id == 1 else "U-เตียงเสริม13",
|
|
room_type_code=room_type,
|
|
quantity=2 if item_id == 1 else 1,
|
|
review_status=status,
|
|
automatic=status == "confirmed",
|
|
source_fragment="【U-TWN8.5】 2",
|
|
updated_at=None,
|
|
)
|
|
|
|
|
|
def draft_page() -> BookingDraftPage:
|
|
return BookingDraftPage(
|
|
summary=BookingDraftSummary(
|
|
draft_id=DRAFT_ID,
|
|
filename="Booking 报表.xlsx",
|
|
status="reviewing",
|
|
source_rows=1,
|
|
worksheet_count=1,
|
|
distinct_group_codes=1,
|
|
extracted_items=2,
|
|
confirmed_items=1,
|
|
pending_items=1,
|
|
deleted_items=0,
|
|
confirmed_room_quantity=2,
|
|
created_at=None,
|
|
updated_at=None,
|
|
),
|
|
items=(
|
|
draft_item(item_id=1),
|
|
draft_item(item_id=2, room_type=None, status="pending"),
|
|
),
|
|
total=2,
|
|
limit=50,
|
|
offset=0,
|
|
)
|
|
|
|
|
|
class FakeStored:
|
|
def __init__(self, reference: ArtifactRef) -> None:
|
|
self.reference = reference
|
|
|
|
def to_artifact_ref(self) -> ArtifactRef:
|
|
return self.reference
|
|
|
|
|
|
class FakeStore:
|
|
def __init__(self) -> None:
|
|
self.calls: list[dict[str, object]] = []
|
|
|
|
def upload_committed(self, **values: object) -> FakeStored:
|
|
self.calls.append(values)
|
|
return FakeStored(
|
|
ArtifactRef(
|
|
role="booking_source",
|
|
file_kind="booking_excel",
|
|
object_key=(
|
|
f"arr/jobs/{values['job_id']}/attempts/0001/committed/"
|
|
"booking_source/booking-source.xlsx"
|
|
),
|
|
original_filename=str(values["original_filename"]),
|
|
sha256=str(values["expected_sha256"]),
|
|
byte_size=int(values["expected_byte_size"]),
|
|
mime_type=(
|
|
"application/vnd.openxmlformats-officedocument."
|
|
"spreadsheetml.sheet"
|
|
),
|
|
)
|
|
)
|
|
|
|
|
|
class FakeRepository:
|
|
def __init__(self, existing: BookingSourceSummary | None = None) -> None:
|
|
self.existing = existing
|
|
self.page: BookingDraftPage | None = None
|
|
self.created: list[tuple[str, ArtifactRef, object]] = []
|
|
self.updated: list[tuple[str, int, str, int]] = []
|
|
self.deleted: list[tuple[str, int]] = []
|
|
self.deleted_batches: list[tuple[str, tuple[int, ...]]] = []
|
|
self.discarded: list[str] = []
|
|
self.activated: list[str] = []
|
|
|
|
def current_source(self) -> BookingSourceSummary | None:
|
|
return self.existing
|
|
|
|
def current_draft(self, *, limit: int = 50, offset: int = 0) -> BookingDraftPage | None:
|
|
_ = (limit, offset)
|
|
return self.page
|
|
|
|
def create_draft(
|
|
self,
|
|
draft_id: str,
|
|
source: ArtifactRef,
|
|
document: object,
|
|
*,
|
|
limit: int = 50,
|
|
) -> BookingDraftPage:
|
|
_ = limit
|
|
self.created.append((draft_id, source, document))
|
|
return draft_page()
|
|
|
|
def update_item(
|
|
self,
|
|
draft_id: str,
|
|
item_id: int,
|
|
room_type: str,
|
|
quantity: int,
|
|
) -> BookingDraftItem:
|
|
self.updated.append((draft_id, item_id, room_type, quantity))
|
|
return draft_item(item_id=item_id, room_type=room_type)
|
|
|
|
def delete_item(self, draft_id: str, item_id: int) -> None:
|
|
self.deleted.append((draft_id, item_id))
|
|
|
|
def delete_items(self, draft_id: str, item_ids: object) -> None:
|
|
assert isinstance(item_ids, tuple)
|
|
self.deleted_batches.append((draft_id, item_ids))
|
|
|
|
def discard_draft(self, draft_id: str) -> None:
|
|
self.discarded.append(draft_id)
|
|
|
|
def activate_draft(self, draft_id: str) -> BookingSourceSummary:
|
|
self.activated.append(draft_id)
|
|
return summary()
|
|
|
|
|
|
class ProgramBookingSourceCoordinatorTests(unittest.TestCase):
|
|
def test_valid_upload_is_stored_privately_then_becomes_review_draft(self) -> None:
|
|
store = FakeStore()
|
|
repository = FakeRepository()
|
|
coordinator = ProgramBookingSourceCoordinator(store, repository)
|
|
|
|
result = coordinator.submit("Booking 报表.xlsx", valid_workbook())
|
|
|
|
self.assertEqual(result["summary"]["filename"], "Booking 报表.xlsx")
|
|
self.assertEqual(result["summary"]["pending_items"], 1)
|
|
self.assertEqual(len(store.calls), 1)
|
|
self.assertEqual(store.calls[0]["role"], "booking_source")
|
|
self.assertTrue(str(store.calls[0]["job_id"]).startswith("bookingdraft-"))
|
|
self.assertEqual(len(repository.created), 1)
|
|
self.assertEqual(repository.created[0][1].original_filename, "Booking 报表.xlsx")
|
|
|
|
def test_reextracting_same_content_still_creates_a_fresh_draft(self) -> None:
|
|
store = FakeStore()
|
|
repository = FakeRepository(summary())
|
|
coordinator = ProgramBookingSourceCoordinator(store, repository)
|
|
|
|
coordinator.submit("again.xlsx", valid_workbook())
|
|
coordinator.submit("again.xlsx", valid_workbook())
|
|
|
|
self.assertEqual(len(store.calls), 2)
|
|
self.assertEqual(len(repository.created), 2)
|
|
self.assertNotEqual(repository.created[0][0], repository.created[1][0])
|
|
|
|
def test_invalid_workbook_never_reaches_storage_or_database(self) -> None:
|
|
store = FakeStore()
|
|
repository = FakeRepository()
|
|
coordinator = ProgramBookingSourceCoordinator(store, repository)
|
|
|
|
with self.assertRaises(PortalError) as raised:
|
|
coordinator.submit("not.xlsx", b"invalid")
|
|
|
|
self.assertEqual(raised.exception.code, "BOOKING_EXCEL_INVALID")
|
|
self.assertEqual(store.calls, [])
|
|
self.assertEqual(repository.created, [])
|
|
|
|
def test_review_actions_delegate_and_return_safe_results(self) -> None:
|
|
repository = FakeRepository()
|
|
repository.page = draft_page()
|
|
coordinator = ProgramBookingSourceCoordinator(FakeStore(), repository)
|
|
|
|
self.assertEqual(coordinator.draft()["summary"]["draft_id"], DRAFT_ID)
|
|
updated = coordinator.update_item(DRAFT_ID, 2, "EXTRA BED", 1)
|
|
self.assertEqual(updated["room_type"], "EXTRA BED")
|
|
self.assertTrue(coordinator.delete_item(DRAFT_ID, 2)["deleted"])
|
|
batch = coordinator.delete_items(DRAFT_ID, [3, 4])
|
|
self.assertEqual(batch["deleted_count"], 2)
|
|
self.assertEqual(batch["item_ids"], [3, 4])
|
|
self.assertTrue(coordinator.discard(DRAFT_ID)["discarded"])
|
|
self.assertEqual(coordinator.activate(DRAFT_ID)["source_batch_id"], 7)
|
|
self.assertEqual(repository.updated, [(DRAFT_ID, 2, "EXTRA BED", 1)])
|
|
self.assertEqual(
|
|
repository.deleted_batches,
|
|
[(DRAFT_ID, (2,)), (DRAFT_ID, (3, 4))],
|
|
)
|
|
self.assertEqual(repository.discarded, [DRAFT_ID])
|
|
self.assertEqual(repository.activated, [DRAFT_ID])
|
|
|
|
def test_current_hides_historical_internal_filename(self) -> None:
|
|
legacy = BookingSourceSummary(
|
|
source_batch_id=1,
|
|
source_kind="expected_fixture",
|
|
filename="internal-fixture.md",
|
|
source_rows=867,
|
|
worksheet_count=6,
|
|
distinct_group_codes=348,
|
|
room_quantity=867,
|
|
activated_at=None,
|
|
)
|
|
coordinator = ProgramBookingSourceCoordinator(FakeStore(), FakeRepository(legacy))
|
|
|
|
result = coordinator.current()
|
|
|
|
self.assertIsNotNone(result)
|
|
self.assertEqual(result["source_type"], "historical")
|
|
self.assertIsNone(result["filename"])
|
|
|
|
def test_bulk_delete_rejects_duplicate_items_before_repository_call(self) -> None:
|
|
repository = FakeRepository()
|
|
coordinator = ProgramBookingSourceCoordinator(FakeStore(), repository)
|
|
|
|
with self.assertRaises(PortalError) as raised:
|
|
coordinator.delete_items(DRAFT_ID, [2, 2])
|
|
|
|
self.assertEqual(raised.exception.code, "BOOKING_EXCEL_REVIEW_ITEMS_INVALID")
|
|
self.assertEqual(repository.deleted_batches, [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|