"""Document isolation regressions; no GUI, DB connection or real movements.""" import unittest from unittest.mock import AsyncMock, patch from barcode_repository import BarcodeRepository, LegacyMoveResult from barcode_service import BarcodeService def row(doc, pallet, priority, order): return dict(Documento=doc, Pallet=pallet, IDStato=priority, Ordinamento=order, Cella=1101, Ubicazione="1A - 1 - a") class PickingRepositoryFake: def __init__(self): # Without a document filter the sequence would be 172, 171, 172. self.rows = [row("173", "600001", 1, 5), row("172", "600002", 0, 10), row("171", "600003", 0, 20), row("172", "600004", 0, 30)] self.skipped = set() self.locations = {r["Pallet"]: r["Cella"] for r in self.rows} self.released = [] async def fetch_next_picking(self, id_stato, documento=None): candidates = [r for r in self.rows if r["IDStato"] == id_stato and (documento is None or r["Documento"] == documento) and r["Pallet"] not in self.skipped] return min(candidates, key=lambda r: r["Ordinamento"]) if candidates else None async def fetch_active_skipped_document(self, id_stato, documento=None): candidates = [r for r in self.rows if r["IDStato"] == id_stato and (documento is None or r["Documento"] == documento) and r["Pallet"] in self.skipped] return dict(Documento=candidates[0]["Documento"], SkippedCount=len(candidates)) if candidates else None async def count_picking_document_remaining(self, documento): return len({r["Pallet"] for r in self.rows if r["Documento"] == documento}) async def release_picking_document(self, *, documento, operator_id): self.released.append(documento) return 0 async def skip_picking_pallet(self, *, documento, pallet, **kwargs): self.skipped.add(pallet) async def fetch_picking_by_pallet(self, pallet, documento=None): return next((r for r in self.rows if r["Pallet"] == pallet and (documento is None or r["Documento"] == documento)), None) async def fetch_trace_by_pallet(self, pallet): return dict(Lotto="LOT", Prodotto="ART", Descrizione="Test") async def fetch_current_location_by_pallet(self, pallet): return dict(IDCella=self.locations[pallet]) async def fetch_open_locations_by_pallet(self, pallet): return [await self.fetch_current_location_by_pallet(pallet)] async def fetch_closed_picking_by_pallet(self, pallet): return None async def execute_legacy_move(self, *, barcode_pallet, barcode_cella, **kwargs): target = 9999 if barcode_cella == "9000000" else 1000 self.locations[barcode_pallet] = target if target == 9999: self.rows = [r for r in self.rows if r["Pallet"] != barcode_pallet] return LegacyMoveResult(0, barcode_cella, barcode_pallet, int(barcode_cella)) class DocumentBindingTests(unittest.IsolatedAsyncioTestCase): def setUp(self): logger = patch("barcode_service.log_runtime_event") logger.start() self.addCleanup(logger.stop) self.repo = PickingRepositoryFake() self.service = BarcodeService(self.repo, operator_id=1) async def ship_current(self): result = await self.service.submit( scanned_pallet=self.service.state.expected_pallet, destination_barcode="9000000") self.assertTrue(result.ok) self.assertEqual(result.state.auto_advance_delay_ms, 3000) async def test_auto_advance_never_interleaves_low_priority_documents(self): await self.service.start_priority_queue(0) await self.ship_current() result = await self.service.start_priority_queue(0) self.assertEqual((result.state.document, result.state.expected_pallet), ("172", "600004")) async def test_f2_f1_f2_stays_bound_even_if_global_order_changes(self): await self.service.start_priority_queue(0) self.repo.rows[2]["Ordinamento"] = 1 for _ in range(5): high = await self.service.start_priority_queue(1) self.assertEqual(high.state.document, "173") low = await self.service.start_priority_queue(0) self.assertEqual((low.state.document, low.state.expected_pallet), ("172", "600002")) async def test_pause_free_move_resume_and_switch_preserve_document(self): await self.service.start_priority_queue(0) paused = self.service.state self.service.begin_priority_pause(0) free = await self.service.submit(scanned_pallet="600003", destination_barcode="9001000") self.assertTrue(free.ok) self.service.resume_priority_state(paused) await self.service.start_priority_queue(1) low = await self.service.start_priority_queue(0) self.assertEqual(low.state.document, "172") self.assertEqual(low.state.expected_pallet, "600002") async def test_skip_advances_only_within_active_document(self): await self.service.start_priority_queue(0) await self.service.skip_current_picking_pallet() low = await self.service.start_priority_queue(0) self.assertEqual(low.state.expected_pallet, "600004") async def test_only_skipped_residuals_stop_even_with_other_lists_available(self): await self.service.start_priority_queue(0) await self.service.skip_current_picking_pallet() await self.service.start_priority_queue(0) await self.service.skip_current_picking_pallet() result = await self.service.start_priority_queue(0) self.assertEqual(result.state.mode, "manual_unload") self.assertIn("PL 172 sospesa", result.state.status_text) self.assertEqual(self.repo.released, ["172"]) async def test_completion_stops_before_next_document(self): await self.service.start_priority_queue(0) await self.ship_current() await self.service.start_priority_queue(0) await self.ship_current() stopped = await self.service.start_priority_queue(0) self.assertEqual(stopped.state.mode, "manual_unload") self.assertIn("PL 172 terminata", stopped.state.status_text) self.assertEqual(stopped.state.auto_advance_delay_ms, 0) # A later explicit F2 can start another list, after the stop message. next_list = await self.service.start_priority_queue(0) self.assertEqual(next_list.state.document, "171") async def test_priority_change_blocks_without_forgetting_previous_document(self): await self.service.start_priority_queue(0) for r in self.repo.rows: if r["Documento"] == "172": r["IDStato"] = 1 for _ in range(2): stopped = await self.service.start_priority_queue(0) self.assertIn("PL 172", stopped.state.status_text) self.assertIn("Verifica la prenotazione", stopped.state.status_text) for r in self.repo.rows: if r["Documento"] == "172": r["IDStato"] = 0 resumed = await self.service.start_priority_queue(0) self.assertEqual(resumed.state.document, "172") async def test_query_exception_keeps_document_for_retry(self): await self.service.start_priority_queue(0) with patch.object(self.repo, "fetch_next_picking", side_effect=TimeoutError): with self.assertRaises(TimeoutError): await self.service.start_priority_queue(0) self.repo.rows[2]["Ordinamento"] = 1 result = await self.service.start_priority_queue(0) self.assertEqual(result.state.document, "172") async def test_wrong_document_from_repository_is_rejected(self): await self.service.start_priority_queue(0) with patch.object(self.repo, "fetch_next_picking", return_value=self.repo.rows[2]): with self.assertRaisesRegex(RuntimeError, "Cambio inatteso documento"): await self.service.start_priority_queue(0) result = await self.service.start_priority_queue(0) self.assertEqual(result.state.document, "172") async def test_confirmation_cannot_display_another_document_for_same_pallet(self): await self.service.start_priority_queue(0) self.repo.rows.append(row("171", "600002", 0, 1)) self.repo.fetch_picking_by_pallet = AsyncMock(wraps=self.repo.fetch_picking_by_pallet) state = await self.service._build_post_move_state( barcode_pallet="600002", destination_barcode="9000000", destination_display="7G.1.1", last_priority_state=0, auto_advance_delay_ms=3000) self.assertEqual(state.document, "172") self.repo.fetch_picking_by_pallet.assert_awaited_once_with("600002", documento="172") async def test_repository_passes_document_to_all_related_queries(self): db = type("DB", (), {"query_json": AsyncMock(return_value={"rows": []})})() repo = BarcodeRepository(db) await repo.fetch_next_picking(0, documento="172") await repo.fetch_active_skipped_document(0, documento="172") await repo.fetch_picking_by_pallet("600002", documento="172") for call in db.query_json.await_args_list: self.assertEqual(call.args[1]["documento"], "172") self.assertIn(":documento", call.args[0]) if __name__ == "__main__": unittest.main()