"""Unit tests for backend.modules.telegram.handlers.receipt_handlers (US-008). Tests are organized by function: - _format_receipt_preview (3 tests) - _confidence_warning (2 tests) - _build_oracle_config (3 tests) - _format_oracle_error (4 tests) - _save_to_oracle (6 tests) - handle_receipt_callback (7 tests) - handle_document_message (5 tests) - handle_photo_message (3 tests) """ import asyncio import sys import types from datetime import date, datetime, timedelta from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch import pytest # --------------------------------------------------------------------------- # Module mocks (BEFORE importing receipt_handlers). # python-telegram-bot, sqlmodel and the data_entry services package import # heavy dependencies that aren't available in the unit-test environment, so we # stub them out to keep these tests pure (no DB / no Telegram). # --------------------------------------------------------------------------- sys.path.insert(0, str(Path(__file__).resolve().parents[3])) # Telegram package stubs. _tg = types.ModuleType("telegram") _tg.Update = MagicMock _tg.InlineKeyboardButton = MagicMock _tg.InlineKeyboardMarkup = MagicMock sys.modules.setdefault("telegram", _tg) _tg_constants = types.ModuleType("telegram.constants") class _FakeParseMode: MARKDOWN = "Markdown" _tg_constants.ParseMode = _FakeParseMode sys.modules.setdefault("telegram.constants", _tg_constants) _tg_error = types.ModuleType("telegram.error") class _FakeTelegramError(Exception): """Fake replacement for telegram.error.TelegramError.""" _tg_error.TelegramError = _FakeTelegramError sys.modules.setdefault("telegram.error", _tg_error) _tg_ext = types.ModuleType("telegram.ext") class _FakeContextTypes: DEFAULT_TYPE = MagicMock _tg_ext.ContextTypes = _FakeContextTypes sys.modules.setdefault("telegram.ext", _tg_ext) # Stub out backend.modules.data_entry.services package — its real __init__ # imports SQLModel-based models that are heavy and not needed for these tests. _de = types.ModuleType("backend.modules.data_entry") _de.__path__ = [] _de_services = types.ModuleType("backend.modules.data_entry.services") _de_services.__path__ = [] _de_services_ocr = types.ModuleType("backend.modules.data_entry.services.ocr") _de_services_ocr.__path__ = [] _de_queue = types.ModuleType("backend.modules.data_entry.services.ocr.queue_client") _de_queue.submit_ocr_job = AsyncMock(return_value="job-test") _de_queue.wait_for_result = AsyncMock(return_value={"success": True, "result": {}}) _de_writer = types.ModuleType("backend.modules.data_entry.services.oracle_receipt_writer") _de_writer.write_receipt = MagicMock(return_value=(123, "OK")) sys.modules.setdefault("backend.modules.data_entry", _de) sys.modules.setdefault("backend.modules.data_entry.services", _de_services) sys.modules.setdefault("backend.modules.data_entry.services.ocr", _de_services_ocr) sys.modules.setdefault( "backend.modules.data_entry.services.ocr.queue_client", _de_queue ) sys.modules.setdefault( "backend.modules.data_entry.services.oracle_receipt_writer", _de_writer ) # Now safe to import the module under test. import oracledb # noqa: E402 from backend.modules.telegram.handlers import receipt_handlers as rh # noqa: E402 # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture(autouse=True) def _reset_module_state(): """Clear caches between tests so they don't leak.""" rh._pending_receipts.clear() rh._write_pool_registered.clear() yield rh._pending_receipts.clear() rh._write_pool_registered.clear() def _make_server(server_id="srv1", password="readpass"): srv = MagicMock() srv.id = server_id srv.name = "Test Server" srv.host = "localhost" srv.port = 1521 srv.user = "TESTUSER" srv.password = password srv.sid = "ROA" srv.service_name = None srv.get_dsn = MagicMock(return_value="localhost:1521/ROA") return srv # =========================================================================== # _format_receipt_preview (3 tests) # =========================================================================== def test_format_receipt_preview_complete(): receipt = { "partner_name": "LIDL DISCOUNT", "cui": "22891860", "receipt_date": date(2026, 1, 15), "receipt_number": "BON-001", "amount": 123.45, "tva_total": 19.71, } out = rh._format_receipt_preview(receipt) assert "LIDL DISCOUNT" in out assert "22891860" in out assert "15.01.2026" in out assert "BON-001" in out assert "123.45" in out assert "19.71" in out assert out.startswith("📄 *Preview bon fiscal*") def test_format_receipt_preview_with_missing_fields(): out = rh._format_receipt_preview({}) assert "Necunoscut" in out # partner_name fallback assert "—" in out # other fields fallback assert "0.00 RON" in out # amount fallback def test_format_receipt_preview_with_string_date(): """Date as ISO string (not yet normalized) must not crash and is stringified.""" receipt = { "partner_name": "Test", "cui": "1", "receipt_date": "2026-03-04", "receipt_number": "X", "amount": 50, "tva_total": 5, } out = rh._format_receipt_preview(receipt) assert "2026-03-04" in out assert "50.00" in out # =========================================================================== # _confidence_warning (2 tests) # =========================================================================== def test_confidence_warning_below_threshold(): out = rh._confidence_warning(0.45) assert "Atenție" in out assert "45%" in out assert out.startswith("\n⚠") def test_confidence_warning_above_threshold_or_none(): assert rh._confidence_warning(0.99) == "" assert rh._confidence_warning(rh.LOW_CONFIDENCE_THRESHOLD) == "" assert rh._confidence_warning(None) == "" # =========================================================================== # _build_oracle_config (3 tests) # =========================================================================== def test_build_oracle_config_returns_read_credentials(): """Single Oracle user: config returns the same user/password used for read.""" srv = _make_server(server_id="srv1", password="readpass") fake_settings = MagicMock() fake_settings.get_oracle_server.return_value = srv with patch.object(rh, "settings", fake_settings): cfg = rh._build_oracle_config("srv1") assert cfg == { "user": "TESTUSER", "password": "readpass", "dsn": "localhost:1521/ROA", } def test_build_oracle_config_no_server_id_uses_default(): srv = _make_server(server_id="default") fake_settings = MagicMock() fake_settings.get_default_oracle_server.return_value = srv with patch.object(rh, "settings", fake_settings): cfg = rh._build_oracle_config(None) assert cfg is not None assert cfg["user"] == "TESTUSER" fake_settings.get_default_oracle_server.assert_called_once() def test_build_oracle_config_no_server_returns_none(): fake_settings = MagicMock() fake_settings.get_default_oracle_server.return_value = None fake_settings.get_oracle_server.return_value = None with patch.object(rh, "settings", fake_settings): assert rh._build_oracle_config(None) is None assert rh._build_oracle_config("does-not-exist") is None # =========================================================================== # _format_oracle_error (4 tests) # =========================================================================== def test_format_oracle_error_ora_01017(): err = oracledb.DatabaseError("ORA-01017: invalid username/password") msg = rh._format_oracle_error(err) assert "Credențiale Oracle incorecte" in msg assert msg.startswith("❌") def test_format_oracle_error_ora_00001(): err = oracledb.DatabaseError("ORA-00001: unique constraint violated") msg = rh._format_oracle_error(err) assert "duplicat" in msg.lower() def test_format_oracle_error_ora_12541(): err = oracledb.DatabaseError("ORA-12541: TNS no listener") msg = rh._format_oracle_error(err) assert "nu este disponibil" in msg def test_format_oracle_error_unknown_code(): err = oracledb.DatabaseError("ORA-99999: something exotic") msg = rh._format_oracle_error(err) assert "Eroare Oracle" in msg assert "ORA-99999" in msg # =========================================================================== # _save_to_oracle (6 tests) # =========================================================================== @pytest.mark.asyncio async def test_save_to_oracle_registers_pool_first_time(): receipt = {"partner_name": "Test"} cfg = {"user": "READUSER", "password": "P", "dsn": "localhost:1521/ROA"} srv = _make_server() fake_settings = MagicMock() fake_settings.get_oracle_server.return_value = srv fake_pool = MagicMock() fake_conn = MagicMock() fake_pool.acquire.return_value = fake_conn with patch.object(rh, "settings", fake_settings), \ patch.object(rh.oracle_pool, "register_server") as mock_register, \ patch.object(rh.oracle_pool, "is_server_registered", return_value=True), \ patch.object(rh.oracle_pool, "get_pool", new=AsyncMock(return_value=fake_pool)), \ patch.object(rh, "write_receipt", return_value=(7, "OK")) as mock_write: cod, mesaj = await rh._save_to_oracle(receipt, cfg, server_id="srv1", schema="MARIUSM_AUTO") mock_register.assert_called_once() kwargs = mock_register.call_args.kwargs # Connect AS schema owner with the read user's password assert kwargs["user"] == "MARIUSM_AUTO" assert kwargs["password"] == "P" assert "srv1_write_MARIUSM_AUTO" in rh._write_pool_registered mock_write.assert_called_once_with(receipt, fake_conn) assert (cod, mesaj) == (7, "OK") @pytest.mark.asyncio async def test_save_to_oracle_uses_existing_pool_no_reregister(): receipt = {"x": 1} cfg = {"user": "READUSER", "password": "P", "dsn": "d"} rh._write_pool_registered.add("srv1_write_MARIUSM_AUTO") # simulate previous registration fake_pool = MagicMock() fake_pool.acquire.return_value = MagicMock() with patch.object(rh.oracle_pool, "register_server") as mock_register, \ patch.object(rh.oracle_pool, "is_server_registered", return_value=True), \ patch.object(rh.oracle_pool, "get_pool", new=AsyncMock(return_value=fake_pool)), \ patch.object(rh, "write_receipt", return_value=(1, "OK")): await rh._save_to_oracle(receipt, cfg, server_id="srv1", schema="MARIUSM_AUTO") mock_register.assert_not_called() @pytest.mark.asyncio async def test_save_to_oracle_returns_cod_and_message(): fake_pool = MagicMock() fake_pool.acquire.return_value = MagicMock() fake_settings = MagicMock() fake_settings.get_default_oracle_server.return_value = _make_server() with patch.object(rh, "settings", fake_settings), \ patch.object(rh.oracle_pool, "register_server"), \ patch.object(rh.oracle_pool, "is_server_registered", return_value=True), \ patch.object(rh.oracle_pool, "get_pool", new=AsyncMock(return_value=fake_pool)), \ patch.object(rh, "write_receipt", return_value=(42, "Bon salvat")): cod, mesaj = await rh._save_to_oracle({}, {"user": "u", "password": "p", "dsn": "d"}, None) assert cod == 42 assert mesaj == "Bon salvat" @pytest.mark.asyncio async def test_save_to_oracle_falls_back_to_direct_when_no_server_config(): """When server lookup returns None and pool isn't registered, write_receipt is called directly with the dict cfg (no pool indirection).""" cfg = {"user": "u", "password": "p", "dsn": "d"} fake_settings = MagicMock() fake_settings.get_default_oracle_server.return_value = None with patch.object(rh, "settings", fake_settings), \ patch.object(rh.oracle_pool, "is_server_registered", return_value=False), \ patch.object(rh, "write_receipt", return_value=(99, "OK")) as mock_write: cod, mesaj = await rh._save_to_oracle({"a": 1}, cfg, None) mock_write.assert_called_once_with({"a": 1}, cfg) assert (cod, mesaj) == (99, "OK") @pytest.mark.asyncio async def test_save_to_oracle_releases_connection_after_write(): """Connection acquired from pool must be closed in finally clause.""" fake_conn = MagicMock() fake_pool = MagicMock() fake_pool.acquire.return_value = fake_conn rh._write_pool_registered.add("srv1_write_MARIUSM_AUTO") with patch.object(rh.oracle_pool, "is_server_registered", return_value=True), \ patch.object(rh.oracle_pool, "get_pool", new=AsyncMock(return_value=fake_pool)), \ patch.object(rh, "write_receipt", return_value=(1, "OK")): await rh._save_to_oracle({}, {"user": "u", "password": "p", "dsn": "d"}, "srv1", "MARIUSM_AUTO") fake_conn.close.assert_called_once() @pytest.mark.asyncio async def test_save_to_oracle_propagates_oracle_error(): """Errors from write_receipt bubble up so handler can translate them.""" fake_pool = MagicMock() fake_pool.acquire.return_value = MagicMock() rh._write_pool_registered.add("srv1_write") db_err = oracledb.DatabaseError("ORA-00001: unique violation") with patch.object(rh.oracle_pool, "is_server_registered", return_value=True), \ patch.object(rh.oracle_pool, "get_pool", new=AsyncMock(return_value=fake_pool)), \ patch.object(rh, "write_receipt", side_effect=db_err): with pytest.raises(oracledb.DatabaseError): await rh._save_to_oracle({}, {"user": "u", "password": "p", "dsn": "d"}, "srv1") # =========================================================================== # handle_receipt_callback (7 tests) # =========================================================================== def _make_callback_update(data, effective_user_id=42): update = MagicMock() update.effective_user.id = effective_user_id update.callback_query.data = data update.callback_query.answer = AsyncMock() update.callback_query.edit_message_text = AsyncMock() return update @pytest.mark.asyncio async def test_handle_receipt_callback_invalid_data_format(): update = _make_callback_update("garbage") await rh.handle_receipt_callback(update, MagicMock()) update.callback_query.edit_message_text.assert_awaited_once() args, _ = update.callback_query.edit_message_text.call_args assert "invalidă" in args[0].lower() @pytest.mark.asyncio async def test_handle_receipt_callback_invalid_uid_int(): update = _make_callback_update("receipt:confirm:notanumber") await rh.handle_receipt_callback(update, MagicMock()) update.callback_query.edit_message_text.assert_awaited_once() args, _ = update.callback_query.edit_message_text.call_args assert "invalidă" in args[0].lower() @pytest.mark.asyncio async def test_handle_receipt_callback_unauthorized_other_user(): """Caller is user 99, but the callback targets user 42 → forbidden.""" update = _make_callback_update("receipt:confirm:42", effective_user_id=99) await rh.handle_receipt_callback(update, MagicMock()) # second answer() call carries the alert message answer_calls = update.callback_query.answer.await_args_list assert len(answer_calls) == 2 assert "permisiunea" in answer_calls[1].args[0].lower() update.callback_query.edit_message_text.assert_not_called() @pytest.mark.asyncio async def test_handle_receipt_callback_pending_not_found(): update = _make_callback_update("receipt:confirm:42", effective_user_id=42) # Empty _pending_receipts (autouse fixture clears it). await rh.handle_receipt_callback(update, MagicMock()) update.callback_query.edit_message_text.assert_awaited_once() args, _ = update.callback_query.edit_message_text.call_args assert "expirat" in args[0].lower() or "procesat" in args[0].lower() @pytest.mark.asyncio async def test_handle_receipt_callback_pending_expired(): update = _make_callback_update("receipt:confirm:42", effective_user_id=42) rh._pending_receipts[42] = { "receipt_dict": {}, "temp_path": "/tmp/receipt_doesnotexist.pdf", "created_at": datetime.now() - timedelta(seconds=rh.PENDING_TTL_S + 60), } await rh.handle_receipt_callback(update, MagicMock()) assert 42 not in rh._pending_receipts args, _ = update.callback_query.edit_message_text.call_args assert "expirat" in args[0].lower() @pytest.mark.asyncio async def test_handle_receipt_callback_cancel_clears_state(): update = _make_callback_update("receipt:cancel:42", effective_user_id=42) rh._pending_receipts[42] = { "receipt_dict": {"partner_name": "X"}, "temp_path": "/tmp/receipt_xyz.pdf", "created_at": datetime.now(), } with patch("pathlib.Path.unlink") as mock_unlink: await rh.handle_receipt_callback(update, MagicMock()) assert 42 not in rh._pending_receipts mock_unlink.assert_called_once_with(missing_ok=True) args, _ = update.callback_query.edit_message_text.call_args assert "anulat" in args[0].lower() @pytest.mark.asyncio async def test_handle_receipt_callback_confirm_writes_to_oracle(): update = _make_callback_update("receipt:confirm:42", effective_user_id=42) rh._pending_receipts[42] = { "receipt_dict": {"partner_name": "X", "cui": "1", "amount": 50, "tva_total": 5}, "temp_path": "/tmp/receipt_xyz.pdf", "created_at": datetime.now(), } fake_user = {"oracle_server_id": "srv1"} fake_cfg = {"user": "TESTUSER", "password": "P", "dsn": "d"} fake_session = MagicMock() fake_session.get_active_company.return_value = {"id": 110, "name": "MARIUSM AUTO"} fake_sm = MagicMock() fake_sm.get_or_create_session = AsyncMock(return_value=fake_session) with patch("backend.modules.telegram.handlers.receipt_handlers.get_user", new=AsyncMock(return_value=fake_user)), \ patch.object(rh, "_build_oracle_config", return_value=fake_cfg), \ patch.object(rh, "_resolve_schema", new=AsyncMock(return_value="MARIUSM_AUTO")), \ patch("backend.modules.telegram.agent.session.get_session_manager", return_value=fake_sm), \ patch.object(rh, "_save_to_oracle", new=AsyncMock(return_value=(101, "Document salvat"))), \ patch("pathlib.Path.unlink"): await rh.handle_receipt_callback(update, MagicMock()) # Two edit_message_text calls: "Salvez..." then success message. edit_calls = update.callback_query.edit_message_text.await_args_list assert len(edit_calls) >= 2 final_text = edit_calls[-1].args[0] assert "salvat" in final_text.lower() assert "101" in final_text assert 42 not in rh._pending_receipts # =========================================================================== # handle_document_message (5 tests) # =========================================================================== def _make_message_update(user_id=42, document=None, photo=None): update = MagicMock() update.effective_user.id = user_id update.message.reply_text = AsyncMock() if document is not None: update.message.document = document if photo is not None: update.message.photo = photo return update def _make_telegram_doc(file_id="file-1", file_name="bon.pdf"): doc = MagicMock() doc.file_id = file_id doc.file_name = file_name return doc @pytest.mark.asyncio async def test_handle_document_unlinked_user_replies_login_hint(): update = _make_message_update(user_id=1, document=_make_telegram_doc()) context = MagicMock() context.bot.get_file = AsyncMock() with patch("backend.modules.telegram.handlers.receipt_handlers.is_user_linked", new=AsyncMock(return_value=False)): await rh.handle_document_message(update, context) args, _ = update.message.reply_text.call_args assert "/login" in args[0] context.bot.get_file.assert_not_called() @pytest.mark.asyncio async def test_handle_document_pending_exists_blocks_new_submission(): rh._pending_receipts[42] = { "receipt_dict": {}, "temp_path": "/tmp/receipt_x.pdf", "created_at": datetime.now(), } update = _make_message_update(user_id=42, document=_make_telegram_doc()) context = MagicMock() context.bot.get_file = AsyncMock() with patch("backend.modules.telegram.handlers.receipt_handlers.is_user_linked", new=AsyncMock(return_value=True)): await rh.handle_document_message(update, context) args, _ = update.message.reply_text.call_args assert "așteptare" in args[0] context.bot.get_file.assert_not_called() @pytest.mark.asyncio async def test_handle_document_download_failure_replies_with_error(): update = _make_message_update(user_id=42, document=_make_telegram_doc()) context = MagicMock() tg_file = MagicMock() tg_file.download_to_drive = AsyncMock(side_effect=OSError("net down")) context.bot.get_file = AsyncMock(return_value=tg_file) with patch("backend.modules.telegram.handlers.receipt_handlers.is_user_linked", new=AsyncMock(return_value=True)), \ patch.object(rh, "_submit_ocr_and_preview", new=AsyncMock()) as mock_submit: await rh.handle_document_message(update, context) # Last reply must be an error message; submit-ocr must not have run. last_args, _ = update.message.reply_text.call_args assert "❌" in last_args[0] mock_submit.assert_not_awaited() @pytest.mark.asyncio async def test_handle_document_success_invokes_submit_ocr(): update = _make_message_update(user_id=42, document=_make_telegram_doc()) context = MagicMock() tg_file = MagicMock() tg_file.download_to_drive = AsyncMock() context.bot.get_file = AsyncMock(return_value=tg_file) with patch("backend.modules.telegram.handlers.receipt_handlers.is_user_linked", new=AsyncMock(return_value=True)), \ patch.object(rh, "_submit_ocr_and_preview", new=AsyncMock()) as mock_submit: await rh.handle_document_message(update, context) mock_submit.assert_awaited_once() # Third arg is the temp Path; must use the receipt_ prefix. temp_path = mock_submit.await_args.args[2] assert isinstance(temp_path, Path) assert temp_path.name.startswith(rh.TEMP_FILE_PREFIX) @pytest.mark.asyncio async def test_handle_document_unknown_extension_defaults_to_pdf(): """When file_name has no extension, suffix falls back to '.pdf'.""" doc = _make_telegram_doc(file_name="weirdfile") update = _make_message_update(user_id=42, document=doc) context = MagicMock() tg_file = MagicMock() tg_file.download_to_drive = AsyncMock() context.bot.get_file = AsyncMock(return_value=tg_file) with patch("backend.modules.telegram.handlers.receipt_handlers.is_user_linked", new=AsyncMock(return_value=True)), \ patch.object(rh, "_submit_ocr_and_preview", new=AsyncMock()) as mock_submit: await rh.handle_document_message(update, context) temp_path = mock_submit.await_args.args[2] assert temp_path.suffix == ".pdf" # =========================================================================== # handle_photo_message (3 tests) # =========================================================================== def _make_telegram_photo_list(): """PTB returns a list of PhotoSize from smallest to largest; we use [-1].""" small = MagicMock() small.file_id = "small-id" large = MagicMock() large.file_id = "large-id" return [small, large] @pytest.mark.asyncio async def test_handle_photo_unlinked_user_replies_login_hint(): update = _make_message_update(user_id=1, photo=_make_telegram_photo_list()) context = MagicMock() context.bot.get_file = AsyncMock() with patch("backend.modules.telegram.handlers.receipt_handlers.is_user_linked", new=AsyncMock(return_value=False)): await rh.handle_photo_message(update, context) args, _ = update.message.reply_text.call_args assert "/login" in args[0] context.bot.get_file.assert_not_called() @pytest.mark.asyncio async def test_handle_photo_pending_exists_blocks_new_submission(): rh._pending_receipts[42] = { "receipt_dict": {}, "temp_path": "/tmp/receipt_x.jpg", "created_at": datetime.now(), } update = _make_message_update(user_id=42, photo=_make_telegram_photo_list()) context = MagicMock() context.bot.get_file = AsyncMock() with patch("backend.modules.telegram.handlers.receipt_handlers.is_user_linked", new=AsyncMock(return_value=True)): await rh.handle_photo_message(update, context) args, _ = update.message.reply_text.call_args assert "așteptare" in args[0] context.bot.get_file.assert_not_called() @pytest.mark.asyncio async def test_handle_photo_success_uses_largest_resolution(): photos = _make_telegram_photo_list() update = _make_message_update(user_id=42, photo=photos) context = MagicMock() tg_file = MagicMock() tg_file.download_to_drive = AsyncMock() context.bot.get_file = AsyncMock(return_value=tg_file) with patch("backend.modules.telegram.handlers.receipt_handlers.is_user_linked", new=AsyncMock(return_value=True)), \ patch.object(rh, "_submit_ocr_and_preview", new=AsyncMock()) as mock_submit: await rh.handle_photo_message(update, context) # Largest photo is photos[-1], so file_id="large-id". context.bot.get_file.assert_awaited_once_with("large-id") temp_path = mock_submit.await_args.args[2] assert temp_path.suffix == ".jpg" assert temp_path.name.startswith(rh.TEMP_FILE_PREFIX)