"""Integration tests for oracle_receipt_writer.write_receipt. Tests cover: - Mocked Oracle connection (always runs): validates SQL/callproc invocation, ACT_TEMP INSERT lines, commit/rollback flow, exception handling. - Real Oracle dev DB (optional): gated by env var ORACLE_INTEGRATION_DSN. Uses commit=False (dry-run) for cleanup so dev DB stays untouched. Run mocked (default): pytest tests/integration/test_oracle_receipt_writer.py -v Run against real dev DB: ORACLE_INTEGRATION_DSN=10.0.20.121:1521/ROA \\ ORACLE_INTEGRATION_USER=MARIUSM_AUTO \\ ORACLE_INTEGRATION_PASSWORD=ROMFASTSOFT \\ pytest tests/integration/test_oracle_receipt_writer.py -v """ import importlib.util import os import sys from datetime import date, datetime from pathlib import Path from unittest.mock import MagicMock, patch import pytest # Ensure project root on sys.path _PROJECT_ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(_PROJECT_ROOT)) def _load_writer_module(): """Load oracle_receipt_writer.py directly to avoid the heavy services/__init__.py import chain (which pulls sqlmodel/aiosqlite/etc. that aren't needed for this test). """ module_path = _PROJECT_ROOT / "backend/modules/data_entry/services/oracle_receipt_writer.py" spec = importlib.util.spec_from_file_location("oracle_receipt_writer", module_path) module = importlib.util.module_from_spec(spec) # Register in sys.modules so `patch("oracle_receipt_writer.oracledb.connect", ...)` resolves sys.modules["oracle_receipt_writer"] = module spec.loader.exec_module(module) return module _writer = _load_writer_module() write_receipt = _writer.write_receipt SAMPLE_RECEIPT = { "partner_name": "MOL Romania SRL", "cui": "RO14991381", "receipt_date": date(2026, 5, 8), "receipt_number": "12345", "amount": 250.00, "tva_total": 47.50, } SAMPLE_CONFIG = { "user": "TEST_USER", "password": "TEST_PASS", "dsn": "localhost:1521/TESTDB", } def _build_mock_cursor(*, next_cod: int = 5, id_part: int = 123, oracle_msg: str = "OK"): """Build a mock cursor pre-loaded with the queries write_receipt issues. next_cod: value returned by `SELECT seq_cod.nextval FROM DUAL` (globally unique document COD). """ cursor = MagicMock() cursor.fetchone.side_effect = [(next_cod,), (id_part,)] var_obj = MagicMock() var_obj.getvalue.return_value = oracle_msg cursor.var.return_value = var_obj return cursor def _build_mock_connection(cursor): conn = MagicMock() conn.cursor.return_value = cursor return conn # ---------------------------------------------------------------------------- # Mocked-DB integration tests # ---------------------------------------------------------------------------- class TestWriteReceiptMocked: """Validates write_receipt logic against a fully mocked Oracle connection.""" def test_returns_cod_and_message_tuple(self): cursor = _build_mock_cursor(next_cod=43, oracle_msg="Saved OK") mock_conn = _build_mock_connection(cursor) with patch( "oracle_receipt_writer.oracledb.connect", return_value=mock_conn, ): cod, mesaj = write_receipt(SAMPLE_RECEIPT, SAMPLE_CONFIG, commit=True) assert cod == 43 assert mesaj == "Saved OK" def test_inserts_three_act_temp_lines_when_tva_present(self): cursor = _build_mock_cursor() mock_conn = _build_mock_connection(cursor) with patch( "oracle_receipt_writer.oracledb.connect", return_value=mock_conn, ): write_receipt(SAMPLE_RECEIPT, SAMPLE_CONFIG, commit=True) # cursor.execute is called for: SELECT max(COD), SELECT ID_PART, then INSERTs execute_calls = cursor.execute.call_args_list insert_calls = [c for c in execute_calls if "INSERT INTO ACT_TEMP" in c.args[0]] assert len(insert_calls) == 3, "Expected 3 ACT_TEMP lines: expense, TVA, payment" # Validate accounting accounts are correct (MOL → 6022, with TVA → 4426) scds = [c.kwargs["scd"] for c in insert_calls] sccs = [c.kwargs["scc"] for c in insert_calls] assert scds == ["6022", "4426", "401"] assert sccs == ["401", "401", "5311"] def test_inserts_two_act_temp_lines_when_no_tva(self): receipt = {**SAMPLE_RECEIPT, "tva_total": 0} cursor = _build_mock_cursor() mock_conn = _build_mock_connection(cursor) with patch( "oracle_receipt_writer.oracledb.connect", return_value=mock_conn, ): write_receipt(receipt, SAMPLE_CONFIG, commit=True) insert_calls = [ c for c in cursor.execute.call_args_list if "INSERT INTO ACT_TEMP" in c.args[0] ] assert len(insert_calls) == 2 # No TVA line def test_invokes_pack_contafin_initializeaza_and_finalizeaza(self): cursor = _build_mock_cursor() mock_conn = _build_mock_connection(cursor) with patch( "oracle_receipt_writer.oracledb.connect", return_value=mock_conn, ): write_receipt(SAMPLE_RECEIPT, SAMPLE_CONFIG, commit=True) callproc_names = [c.args[0] for c in cursor.callproc.call_args_list] assert "PACK_CONTAFIN.INITIALIZEAZA_SCRIERE_ACT_RUL" in callproc_names assert "PACK_CONTAFIN.FINALIZEAZA_SCRIERE_ACT_RUL" in callproc_names def test_commit_true_calls_commit(self): cursor = _build_mock_cursor() mock_conn = _build_mock_connection(cursor) with patch( "oracle_receipt_writer.oracledb.connect", return_value=mock_conn, ): write_receipt(SAMPLE_RECEIPT, SAMPLE_CONFIG, commit=True) mock_conn.commit.assert_called_once() mock_conn.rollback.assert_not_called() def test_commit_false_triggers_rollback_for_cleanup(self): """Dry-run mode rolls back even on success — the cleanup path.""" cursor = _build_mock_cursor() mock_conn = _build_mock_connection(cursor) with patch( "oracle_receipt_writer.oracledb.connect", return_value=mock_conn, ): cod, mesaj = write_receipt(SAMPLE_RECEIPT, SAMPLE_CONFIG, commit=False) mock_conn.rollback.assert_called_once() mock_conn.commit.assert_not_called() # Cod and mesaj are still returned — dry-run gives caller visibility assert cod > 0 assert mesaj == "OK" def test_exception_during_insert_triggers_rollback(self): cursor = MagicMock() cursor.fetchone.side_effect = [(1,), (123,)] cursor.var.return_value = MagicMock(getvalue=lambda: "") # Cursor.execute raises on the INSERT (third execute call) execute_results = [None, None, RuntimeError("DB blew up")] cursor.execute.side_effect = execute_results mock_conn = _build_mock_connection(cursor) with patch( "oracle_receipt_writer.oracledb.connect", return_value=mock_conn, ): with pytest.raises(RuntimeError, match="DB blew up"): write_receipt(SAMPLE_RECEIPT, SAMPLE_CONFIG, commit=True) mock_conn.rollback.assert_called_once() mock_conn.commit.assert_not_called() def test_owns_connection_when_dict_config_passed(self): cursor = _build_mock_cursor() mock_conn = _build_mock_connection(cursor) with patch( "oracle_receipt_writer.oracledb.connect", return_value=mock_conn, ) as mock_connect: write_receipt(SAMPLE_RECEIPT, SAMPLE_CONFIG, commit=True) mock_connect.assert_called_once_with( user="TEST_USER", password="TEST_PASS", dsn="localhost:1521/TESTDB" ) mock_conn.close.assert_called_once() def test_does_not_close_pre_acquired_connection(self): """When caller passes oracledb.Connection directly, lifecycle is theirs.""" import oracledb cursor = _build_mock_cursor() # Spec as oracledb.Connection so isinstance(...) check succeeds mock_conn = MagicMock(spec=oracledb.Connection) mock_conn.cursor.return_value = cursor write_receipt(SAMPLE_RECEIPT, mock_conn, commit=True) mock_conn.commit.assert_called_once() mock_conn.close.assert_not_called() def test_partner_lookup_falls_back_to_zero_when_cui_unknown(self): cursor = MagicMock() # max_cod = 1, id_part lookup returns nothing cursor.fetchone.side_effect = [(1,), None] cursor.var.return_value = MagicMock(getvalue=lambda: "OK") mock_conn = _build_mock_connection(cursor) with patch( "oracle_receipt_writer.oracledb.connect", return_value=mock_conn, ): write_receipt(SAMPLE_RECEIPT, SAMPLE_CONFIG, commit=True) # All ACT_TEMP inserts must succeed even with id_part=0 insert_calls = [ c for c in cursor.execute.call_args_list if "INSERT INTO ACT_TEMP" in c.args[0] ] assert len(insert_calls) == 3 # First line (expense): id_partc=0 (partner unknown), id_partd=0 assert insert_calls[0].kwargs["id_partc"] == 0 def test_uses_receipt_date_for_year_and_month(self): cursor = _build_mock_cursor() mock_conn = _build_mock_connection(cursor) with patch( "oracle_receipt_writer.oracledb.connect", return_value=mock_conn, ): write_receipt(SAMPLE_RECEIPT, SAMPLE_CONFIG, commit=True) # INITIALIZEAZA call args: [id_util, datetime.now(), an, luna, ...] init_call = next( c for c in cursor.callproc.call_args_list if c.args[0] == "PACK_CONTAFIN.INITIALIZEAZA_SCRIERE_ACT_RUL" ) an, luna = init_call.args[1][2], init_call.args[1][3] assert an == 2026 assert luna == 5 def test_falls_back_to_now_when_receipt_date_missing(self): receipt = {**SAMPLE_RECEIPT, "receipt_date": None} cursor = _build_mock_cursor() mock_conn = _build_mock_connection(cursor) with patch( "oracle_receipt_writer.oracledb.connect", return_value=mock_conn, ): write_receipt(receipt, SAMPLE_CONFIG, commit=True) init_call = next( c for c in cursor.callproc.call_args_list if c.args[0] == "PACK_CONTAFIN.INITIALIZEAZA_SCRIERE_ACT_RUL" ) an = init_call.args[1][2] # Year should be current year (not 1970 or similar) assert an == datetime.now().year # ---------------------------------------------------------------------------- # Real Oracle dev DB integration tests (opt-in) # ---------------------------------------------------------------------------- _ORACLE_DSN = os.getenv("ORACLE_INTEGRATION_DSN") _ORACLE_USER = os.getenv("ORACLE_INTEGRATION_USER") _ORACLE_PASSWORD = os.getenv("ORACLE_INTEGRATION_PASSWORD") _real_db_unavailable = not all([_ORACLE_DSN, _ORACLE_USER, _ORACLE_PASSWORD]) @pytest.mark.skipif( _real_db_unavailable, reason="ORACLE_INTEGRATION_DSN/USER/PASSWORD not set — skipping real DB tests", ) class TestWriteReceiptRealDb: """Real Oracle dev DB tests. Use commit=False so dev DB stays untouched.""" @pytest.fixture def oracle_cfg(self): return { "user": _ORACLE_USER, "password": _ORACLE_PASSWORD, "dsn": _ORACLE_DSN, } def test_dry_run_returns_valid_cod_and_message(self, oracle_cfg): """Dry-run write should return (cod, message) and rollback the transaction.""" cod, mesaj = write_receipt(SAMPLE_RECEIPT, oracle_cfg, commit=False) assert isinstance(cod, int) and cod > 0, f"Expected positive int COD, got {cod!r}" assert isinstance(mesaj, str), f"Expected string message, got {type(mesaj)}" def test_dry_run_act_temp_row_rolled_back(self, oracle_cfg): """After commit=False, ACT_TEMP must NOT contain the row (rollback verified).""" import oracledb cod, _ = write_receipt(SAMPLE_RECEIPT, oracle_cfg, commit=False) # Verify the row was rolled back — query ACT_TEMP for the COD with oracledb.connect(**oracle_cfg) as verify_conn: with verify_conn.cursor() as verify_cursor: verify_cursor.execute( "SELECT COUNT(*) FROM ACT_TEMP WHERE COD = :cod AND AN = :an AND LUNA = :luna", cod=cod, an=SAMPLE_RECEIPT["receipt_date"].year, luna=SAMPLE_RECEIPT["receipt_date"].month, ) count = verify_cursor.fetchone()[0] assert count == 0, f"Rollback failed: {count} rows still in ACT_TEMP for COD={cod}" if __name__ == "__main__": pytest.main([__file__, "-v"])