feat(service-auto): săpt 3-phase2 — toate ipotezele confirmate + modul funcțional

Backend:
- service_auto module complet: router, service, schemas, 5 teste suites (22/22 passed)
- 5 endpoints: GET /ping, /firme, /tip-deviz, /masini, POST /comenzi
- SP_CREEAZA_COMANDA_PROTOTIP creat în MARIUSM_AUTO (VALID, 5.9ms)
- oracle_pool.py: session_callback backward-compat patch
- ROA_WEB user: grants SP-only confirmate (H3), mariusm_test pool switchat
- pyproject.toml: integration pytest marker înregistrat

Frontend:
- ComandaNoua.vue: date reale din Oracle (firme/tip-deviz/masini), nu hardcodate
- src/modules/service-auto/services/api.js: axios service cu Bearer token
- src/router/index.js: rută /service-auto/comanda-noua

Docs:
- decision-log.md: verdict MERGE, toate 6 ipoteze CONFIRMED
- learnings.md: 7 patterns reutilizabile
- grants-audit.md: arhitectura multi-tenant + proxy auth analysis + V_NOM_FIRME loop
- template-modul-oracle.md: rețetă completă pentru module Oracle noi
- TODO-phase2.md: 7 items concrete

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

backend/modules/service_auto/tests/test_comanda_persist.py

@@ -0,0 +1,139 @@
+"""
+Integration test: SP_CREEAZA_COMANDA_PROTOTIP persist + reconnect verify — Săpt 9-10
+
+Verifies that:
+1. callproc SP_CREEAZA_COMANDA_PROTOTIP writes a row to DEV_ORDL
+2. After commit + disconnect, a NEW connection sees the row (true durability)
+3. Cleanup removes all prototype rows from DEV_ORDL and NOM_LUCRARI
+
+Run:
+    cd /workspace/roa2web
+    python -m pytest backend/modules/service_auto/tests/test_comanda_persist.py -v -m integration
+"""
+import os
+import sys
+import time
+
+import pytest
+import oracledb
+
+# Add backend to path so secrets/ is accessible
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', '..', '..'))
+
+HOST = "10.0.20.121"
+PORT = 1521
+SERVICE_NAME = "ROA"
+USER = "CONTAFIN_ORACLE"
+SECRETS_FILE = os.path.join(
+    os.path.dirname(__file__), '..', '..', '..', 'secrets', 'central.oracle_pass'
+)
+
+
+def _read_password() -> str:
+    with open(SECRETS_FILE) as f:
+        return f.read().strip()
+
+
+def _connect() -> oracledb.Connection:
+    return oracledb.connect(
+        user=USER,
+        password=_read_password(),
+        host=HOST,
+        port=PORT,
+        service_name=SERVICE_NAME,
+    )
+
+
+@pytest.mark.integration
+def test_comanda_persist_and_reconnect():
+    """
+    Full round-trip: callproc → commit → close → NEW connection → SELECT → assert exists.
+    Cleans up all inserted rows after verification.
+    """
+    # --- Phase 1: Call SP and commit ---
+    conn1 = _connect()
+    id_ordl = None
+    id_lucrare = None
+    try:
+        with conn1.cursor() as cursor:
+            out_id_ordl = cursor.var(oracledb.NUMBER)
+            out_nrord = cursor.var(oracledb.STRING)
+
+            t0 = time.perf_counter()
+            cursor.callproc(
+                "MARIUSM_AUTO.SP_CREEAZA_COMANDA_PROTOTIP",
+                [
+                    3,                      # p_tip IN NUMBER
+                    200,                    # p_id_masiniclient IN NUMBER
+                    'Test pytest persist',  # p_solicitari IN VARCHAR2
+                    110,                    # p_id_firma IN NUMBER
+                    out_id_ordl,            # p_id_ordl OUT NUMBER
+                    out_nrord,              # p_nrord OUT VARCHAR2
+                ],
+            )
+            t_callproc = time.perf_counter() - t0
+
+            id_ordl = int(out_id_ordl.getvalue())
+            nrord = out_nrord.getvalue() or ""
+            print(f"\n[PERSIST] callproc OK in {t_callproc*1000:.1f}ms → id_ordl={id_ordl} nrord={nrord}")
+
+            assert id_ordl > 0, f"Expected positive id_ordl, got {id_ordl}"
+            assert nrord.startswith("P"), f"Expected nrord starting with 'P', got {nrord!r}"
+
+        conn1.commit()
+        print(f"[PERSIST] commit OK on connection 1")
+    finally:
+        conn1.close()
+        print("[PERSIST] connection 1 closed")
+
+    # --- Phase 2: Open NEW connection and verify row exists ---
+    assert id_ordl is not None, "id_ordl must be set before reconnect phase"
+
+    conn2 = _connect()
+    try:
+        with conn2.cursor() as cursor:
+            cursor.execute(
+                "SELECT id_ordl, id_lucrare FROM MARIUSM_AUTO.DEV_ORDL WHERE id_ordl = :id",
+                {"id": id_ordl},
+            )
+            row = cursor.fetchone()
+
+        print(f"[PERSIST] NEW connection SELECT → row={row}")
+        assert row is not None, f"Row id_ordl={id_ordl} not found after reconnect — durability FAILED"
+        assert int(row[0]) == id_ordl, f"id_ordl mismatch: got {row[0]}, expected {id_ordl}"
+        id_lucrare = int(row[1])
+        print(f"[PERSIST] ASSERT PASSED: row exists in new connection ✅ (id_lucrare={id_lucrare})")
+    finally:
+        conn2.close()
+        print("[PERSIST] connection 2 closed")
+
+    # --- Phase 3: Cleanup — delete child then parent ---
+    assert id_lucrare is not None, "id_lucrare must be set for cleanup"
+
+    conn3 = _connect()
+    try:
+        with conn3.cursor() as cursor:
+            # Child first (FK constraint: DEV_ORDL → NOM_LUCRARI)
+            cursor.execute(
+                "DELETE FROM MARIUSM_AUTO.DEV_ORDL WHERE id_ordl = :id",
+                {"id": id_ordl},
+            )
+            deleted_ordl = cursor.rowcount
+            print(f"[PERSIST] DELETE DEV_ORDL id_ordl={id_ordl} → {deleted_ordl} row(s)")
+
+            # Parent second
+            cursor.execute(
+                "DELETE FROM MARIUSM_AUTO.NOM_LUCRARI WHERE id_lucrare = :id",
+                {"id": id_lucrare},
+            )
+            deleted_nom = cursor.rowcount
+            print(f"[PERSIST] DELETE NOM_LUCRARI id_lucrare={id_lucrare} → {deleted_nom} row(s)")
+
+        conn3.commit()
+        print("[PERSIST] cleanup commit OK ✅")
+
+        assert deleted_ordl == 1, f"Expected 1 DEV_ORDL row deleted, got {deleted_ordl}"
+        assert deleted_nom == 1, f"Expected 1 NOM_LUCRARI row deleted, got {deleted_nom}"
+    finally:
+        conn3.close()
+        print("[PERSIST] connection 3 (cleanup) closed")

backend/modules/service_auto/tests/test_diacritice_encoding.py

@@ -0,0 +1,109 @@
+"""
+Integration test — Hypothesis #4: diacritice encoding end-to-end.
+
+Verifies that RAISE_APPLICATION_ERROR(-20001, 'mesaj cu ă î ș ț â') flows correctly
+through the full chain: Oracle → oracledb → _handle_oracle_error → HTTPException.detail
+
+Two layers tested:
+  L1 (live Oracle)  — PL/SQL anonymous block raises ORA-20001 with diacritice;
+                      oracledb.DatabaseError.args[0].message must contain them intact.
+  L2 (_handle_oracle_error) — the HTTPException.detail must contain diacritice,
+                              ORA prefix stripped.
+
+No HTTP server needed.  Requires live Oracle connection (mariusm_test pool credentials).
+
+Run:
+    cd /workspace/roa2web
+    python -m pytest backend/modules/service_auto/tests/test_diacritice_encoding.py -v -m integration
+"""
+import os
+import sys
+import pytest
+import oracledb
+from fastapi import HTTPException
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', '..', '..'))
+
+from backend.modules.service_auto.services.comanda_service import _handle_oracle_error
+
+HOST = "10.0.20.121"
+PORT = 1521
+SERVICE_NAME = "ROA"
+SECRETS_FILE = os.path.join(
+    os.path.dirname(__file__), '..', '..', '..', 'secrets', 'mariusm_test.oracle_pass'
+)
+USER = "ROA_WEB"
+
+# Romanian diacritice in error message — full set
+DIACRITICE_MSG = "Client invalid: ă î ș ț â Ă Î Ș Ț Â"
+
+
+@pytest.fixture(scope="module")
+def oracle_conn():
+    """Live connection as ROA_WEB for the duration of this test module."""
+    with open(SECRETS_FILE) as f:
+        pwd = f.read().strip()
+    conn = oracledb.connect(user=USER, password=pwd, host=HOST, port=PORT, service_name=SERVICE_NAME)
+    yield conn
+    conn.close()
+
+
+@pytest.mark.integration
+def test_l1_oracle_encodes_diacritice_in_raise_application_error(oracle_conn):
+    """
+    L1: Oracle sends diacritice in RAISE_APPLICATION_ERROR message.
+    oracledb decodes them correctly through the NLS chain.
+    """
+    raised = None
+    with oracle_conn.cursor() as cur:
+        try:
+            cur.execute(
+                f"BEGIN RAISE_APPLICATION_ERROR(-20001, :msg); END;",
+                {"msg": DIACRITICE_MSG},
+            )
+        except oracledb.DatabaseError as e:
+            raised = e
+
+    assert raised is not None, "Expected ORA-20001 to be raised"
+    err = raised.args[0]
+    assert err.code == 20001, f"Expected code 20001, got {err.code}"
+
+    # Every Romanian diacritic character must survive the Oracle → Python round-trip
+    for char in "ăîșțâĂÎȘȚÂ":
+        assert char in err.message, (
+            f"Diacritic '{char}' lost in Oracle→oracledb encoding. "
+            f"Full message: {err.message!r}"
+        )
+
+
+@pytest.mark.integration
+def test_l2_handle_oracle_error_preserves_diacritice_in_http_detail(oracle_conn):
+    """
+    L2: _handle_oracle_error strips ORA prefix and passes diacritice into HTTPException.detail.
+    """
+    raised_db = None
+    with oracle_conn.cursor() as cur:
+        try:
+            cur.execute(
+                "BEGIN RAISE_APPLICATION_ERROR(-20001, :msg); END;",
+                {"msg": DIACRITICE_MSG},
+            )
+        except oracledb.DatabaseError as e:
+            raised_db = e
+
+    assert raised_db is not None
+
+    # Pass through the same handler used in production
+    with pytest.raises(HTTPException) as exc_info:
+        _handle_oracle_error(raised_db)
+
+    http_exc = exc_info.value
+    assert http_exc.status_code == 422
+    assert not http_exc.detail.startswith("ORA-"), (
+        f"ORA prefix not stripped from detail: {http_exc.detail!r}"
+    )
+    for char in "ăîșțâĂÎȘȚÂ":
+        assert char in http_exc.detail, (
+            f"Diacritic '{char}' lost in _handle_oracle_error. "
+            f"detail: {http_exc.detail!r}"
+        )

backend/modules/service_auto/tests/test_error_mapping.py

@@ -0,0 +1,97 @@
+"""
+Unit tests — Oracle error code → HTTP status mapping.
+
+Tests _handle_oracle_error() directly.  No live Oracle connection required.
+
+Covered mappings:
+  ORA-20001 … ORA-20999  → HTTP 422  (business errors from RAISE_APPLICATION_ERROR)
+  ORA-12541/12170/12154/12560 → HTTP 503  (Oracle network / TNS unreachable)
+  ORA-01017              → HTTP 500  (bad credentials)
+  ORA-00942              → HTTP 500  (missing object / grant)
+  ORA-<other>            → HTTP 500  (unexpected error)
+"""
+import pytest
+from unittest.mock import MagicMock
+from fastapi import HTTPException
+
+from backend.modules.service_auto.services.comanda_service import _handle_oracle_error
+
+
+def _make_oracle_error(code: int, msg: str = "test error"):
+    """
+    Build a minimal mock that mimics oracledb.DatabaseError structure.
+    _handle_oracle_error only accesses e.args[0].code and e.args[0].message,
+    so a MagicMock is sufficient — no live oracledb import needed.
+    """
+    err_obj = MagicMock()
+    err_obj.code = code
+    err_obj.message = f"ORA-{code:05d}: {msg}"
+    exc = MagicMock()
+    exc.args = (err_obj,)
+    return exc
+
+
+# ---------------------------------------------------------------------------
+# Parametrised: code → expected HTTP status + fragment in detail
+# ---------------------------------------------------------------------------
+@pytest.mark.parametrize("code, status, detail_fragment", [
+    # Business errors — RAISE_APPLICATION_ERROR range
+    (20001, 422, "mesaj business 20001"),
+    (20500, 422, "mesaj business 20500"),
+    (20999, 422, "mesaj business 20999"),
+    # Oracle network / TNS unreachable
+    (12541, 503, "indisponibil"),
+    (12170, 503, "indisponibil"),
+    (12154, 503, "indisponibil"),
+    (12560, 503, "indisponibil"),
+    # Credential / config errors
+    (1017, 500, "configurare"),
+    # Missing object / grant
+    (942,  500, "internă"),
+    # Catch-all unexpected errors
+    (4031, 500, "neașteptată"),   # ORA-04031: unable to allocate memory
+    (600,  500, "neașteptată"),   # ORA-00600: internal error
+])
+def test_error_mapping(code, status, detail_fragment):
+    msg = f"mesaj business {code}" if 20001 <= code <= 20999 else "test error"
+    exc = _make_oracle_error(code, msg)
+
+    with pytest.raises(HTTPException) as exc_info:
+        _handle_oracle_error(exc)
+
+    http_exc = exc_info.value
+    assert http_exc.status_code == status, (
+        f"ORA-{code:05d} → expected HTTP {status}, got {http_exc.status_code}"
+    )
+    assert detail_fragment in http_exc.detail, (
+        f"ORA-{code:05d} detail: expected '{detail_fragment}' in '{http_exc.detail}'"
+    )
+
+
+# ---------------------------------------------------------------------------
+# ORA-20xxx: verify ORA prefix is stripped from business message
+# ---------------------------------------------------------------------------
+def test_20001_strips_ora_prefix():
+    """The 422 detail must not contain the 'ORA-20001:' prefix."""
+    exc = _make_oracle_error(20001, "Mai exista o comanda cu numarul P01-42")
+    with pytest.raises(HTTPException) as exc_info:
+        _handle_oracle_error(exc)
+    detail = exc_info.value.detail
+    assert not detail.startswith("ORA-"), (
+        f"ORA prefix not stripped: '{detail}'"
+    )
+    assert "Mai exista" in detail
+
+
+# ---------------------------------------------------------------------------
+# Edge case: missing code attribute (e.g. thin-mode edge case)
+# ---------------------------------------------------------------------------
+def test_no_code_attribute_maps_to_500():
+    """If the error object has no .code, fall through to generic 500."""
+    err_obj = MagicMock(spec=[])  # spec=[] means NO attributes
+    exc = MagicMock()
+    exc.args = (err_obj,)
+
+    with pytest.raises(HTTPException) as exc_info:
+        _handle_oracle_error(exc)
+    assert exc_info.value.status_code == 500

backend/modules/service_auto/tests/test_grants_integration.py

@@ -0,0 +1,161 @@
+"""
+Integration tests — Hypothesis #3: ROA_WEB grant-scoped access.
+
+Proves:
+  - ROA_WEB cannot INSERT directly into MARIUSM_AUTO.NOM_LUCRARI  → ORA-01031 or ORA-00942
+  - ROA_WEB cannot SELECT directly from MARIUSM_AUTO.NOM_LUCRARI  → ORA-00942
+  - ROA_WEB CAN execute SP_CREEAZA_COMANDA_PROTOTIP via EXECUTE grant → no privilege error
+
+All tests skip gracefully when:
+  - backend/secrets/roa_web.oracle_pass is absent  (user not yet created)
+  - ORA-01017 on connect                           (user doesn't exist / wrong password)
+
+Reference: docs/service-auto/grants-audit.md §3.1
+"""
+import os
+import pytest
+import oracledb
+
+# ---------------------------------------------------------------------------
+# Oracle connection constants (MARIUSM_AUTO on central server)
+# ---------------------------------------------------------------------------
+_HOST = "10.0.20.121"
+_PORT = 1521
+_SERVICE_NAME = "ROA"
+_ROA_WEB_USER = "ROA_WEB"
+_SECRETS_DIR = os.path.normpath(
+    os.path.join(os.path.dirname(__file__), "..", "..", "..", "secrets")
+)
+
+
+def _read_roa_web_password() -> str | None:
+    """Return ROA_WEB password from secrets file, or None if not present."""
+    path = os.path.join(_SECRETS_DIR, "roa_web.oracle_pass")
+    if not os.path.exists(path):
+        return None
+    with open(path) as f:
+        return f.read().strip() or None
+
+
+# ---------------------------------------------------------------------------
+# Shared fixture — one connection per test module
+# ---------------------------------------------------------------------------
+@pytest.fixture(scope="module")
+def roa_web_connection():
+    """
+    Yields a sync oracledb.Connection as ROA_WEB.
+    Skips the whole module if the user does not yet exist on the server.
+    """
+    password = _read_roa_web_password()
+    if not password:
+        pytest.skip(
+            "ROA_WEB password file not found in backend/secrets/ — "
+            "user not yet created (Phase B deferred, see grants-audit.md §3)"
+        )
+
+    try:
+        conn = oracledb.connect(
+            user=_ROA_WEB_USER,
+            password=password,
+            host=_HOST,
+            port=_PORT,
+            service_name=_SERVICE_NAME,
+        )
+    except oracledb.DatabaseError as e:
+        err = e.args[0]
+        code = getattr(err, "code", 0)
+        if code == 1017:
+            pytest.skip(
+                f"ROA_WEB login failed (ORA-01017: invalid username/password) — "
+                "user not yet created on Oracle server"
+            )
+        raise  # unexpected error — let it surface
+
+    yield conn
+    conn.close()
+
+
+# ---------------------------------------------------------------------------
+# Negative tests — direct DML/SELECT must be rejected
+# ---------------------------------------------------------------------------
+@pytest.mark.integration
+def test_insert_direct_fails(roa_web_connection):
+    """
+    ROA_WEB has no INSERT privilege on NOM_LUCRARI.
+    Expected: ORA-01031 (insufficient privileges) or ORA-00942 (no table grant).
+    """
+    conn = roa_web_connection
+    with conn.cursor() as cur:
+        with pytest.raises(oracledb.DatabaseError) as exc_info:
+            cur.execute(
+                "INSERT INTO MARIUSM_AUTO.NOM_LUCRARI (nrord, id_mod) "
+                "VALUES ('TEST_GRANT_PROBE', 1200)"
+            )
+        err = exc_info.value.args[0]
+        assert err.code in (1031, 942), (
+            f"Expected ORA-01031 or ORA-00942, got ORA-{err.code}: {err.message}"
+        )
+
+
+@pytest.mark.integration
+def test_select_direct_fails(roa_web_connection):
+    """
+    ROA_WEB has no SELECT privilege on NOM_LUCRARI.
+    Expected: ORA-00942 (table or view does not exist).
+    """
+    conn = roa_web_connection
+    with conn.cursor() as cur:
+        with pytest.raises(oracledb.DatabaseError) as exc_info:
+            cur.execute(
+                "SELECT COUNT(*) FROM MARIUSM_AUTO.NOM_LUCRARI WHERE ROWNUM < 2"
+            )
+            cur.fetchone()
+        err = exc_info.value.args[0]
+        assert err.code == 942, (
+            f"Expected ORA-00942, got ORA-{err.code}: {err.message}"
+        )
+
+
+# ---------------------------------------------------------------------------
+# Positive test — SP execution must succeed (EXECUTE grant)
+# ---------------------------------------------------------------------------
+@pytest.mark.integration
+def test_exec_sp_succeeds(roa_web_connection):
+    """
+    ROA_WEB has EXECUTE on SP_CREEAZA_COMANDA_PROTOTIP.
+    The SP call must not raise ORA-01031 (insufficient privileges).
+    A FK violation (ORA-02291) is acceptable — it proves the SP was reached.
+    Transaction is always rolled back — no test data left in prod.
+    """
+    conn = roa_web_connection
+    with conn.cursor() as cur:
+        out_id_ordl = cur.var(oracledb.NUMBER)
+        out_nrord = cur.var(oracledb.STRING)
+
+        try:
+            cur.callproc(
+                "MARIUSM_AUTO.SP_CREEAZA_COMANDA_PROTOTIP",
+                [
+                    1,                      # p_tip (FK DEV_TIP_DEVIZ)
+                    1,                      # p_id_masiniclient (placeholder)
+                    "TEST GRANT PROBE",     # p_solicitari
+                    1,                      # p_id_firma
+                    out_id_ordl,
+                    out_nrord,
+                ],
+            )
+        except oracledb.DatabaseError as e:
+            conn.rollback()
+            err = e.args[0]
+            # ORA-02291: FK violation — SP ran, data was bad → EXECUTE grant works
+            if err.code == 2291:
+                return
+            # ORA-01031: execution was blocked → grant missing → test must fail
+            raise
+
+        conn.rollback()  # ALWAYS rollback — never persist test data
+
+        id_ordl = out_id_ordl.getvalue()
+        nrord = out_nrord.getvalue()
+        assert id_ordl is not None, "SP must return a non-null p_id_ordl"
+        assert nrord, "SP must return a non-empty p_nrord"

backend/modules/service_auto/tests/test_pool_concurrency.py

@@ -0,0 +1,154 @@
+"""
+Integration test — Hypothesis #2: oracle pool concurrency — no state leak between connections.
+
+Verifies that:
+  1. Two concurrent asyncio tasks can acquire connections from the same pool simultaneously
+  2. Each connection operates independently (no shared cursor state, no cross-connection leakage)
+  3. session_callback (if set) runs per-connection, not per-pool
+
+Uses the 'mariusm_test' pool (ROA_WEB user).  Requires live Oracle connection.
+
+Run:
+    cd /workspace/roa2web
+    python -m pytest backend/modules/service_auto/tests/test_pool_concurrency.py -v -m integration
+"""
+import asyncio
+import os
+import sys
+import time
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', '..', '..'))
+
+from shared.database.oracle_pool import OracleMultiPool
+
+SECRETS_FILE = os.path.join(
+    os.path.dirname(__file__), '..', '..', '..', 'secrets', 'mariusm_test.oracle_pass'
+)
+HOST = "10.0.20.121"
+PORT = 1521
+SERVICE_NAME = "ROA"
+USER = "ROA_WEB"
+
+
+@pytest.fixture(scope="module")
+def pool():
+    """Dedicated OracleMultiPool instance (not the global one) for isolation."""
+    with open(SECRETS_FILE) as f:
+        pwd = f.read().strip()
+    p = OracleMultiPool.__new__(OracleMultiPool)
+    p._pools = {}
+    p._pool_configs = {}
+    p._initialized = False
+    import asyncio as _asyncio
+    p._pool_lock = _asyncio.Lock()
+    p.register_server(
+        server_id="concurrency_test",
+        host=HOST, port=PORT,
+        user=USER, password=pwd,
+        service_name=SERVICE_NAME,
+        min_connections=2,
+        max_connections=5,
+    )
+    yield p
+    # Cleanup
+    import asyncio as _asyncio
+    _asyncio.get_event_loop().run_until_complete(p.close_pool())
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_two_concurrent_connections_return_correct_results(pool):
+    """
+    Two asyncio tasks run simultaneously on the same pool.
+    Each uses a different bind value — results must not cross.
+    """
+    results = {}
+
+    async def query_task(task_id: int, expected_val: int):
+        async with pool.get_connection("concurrency_test") as conn:
+            with conn.cursor() as cur:
+                # Short sleep to maximise overlap window
+                await asyncio.sleep(0.01)
+                cur.execute("SELECT :v FROM DUAL", {"v": expected_val})
+                row = cur.fetchone()
+                results[task_id] = row[0]
+
+    await asyncio.gather(
+        query_task(1, 111),
+        query_task(2, 222),
+    )
+
+    assert results[1] == 111, f"Task 1 expected 111, got {results[1]}"
+    assert results[2] == 222, f"Task 2 expected 222, got {results[2]}"
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_session_callback_runs_per_connection(pool):
+    """
+    Register a session_callback that writes to a list.
+    Verify it fires each time a new connection is acquired.
+    session_callback must NOT bleed state across connections.
+    """
+    callback_log = []
+
+    def schema_callback(connection, requested_tag):
+        """Simulates ALTER SESSION SET CURRENT_SCHEMA; logs invocation."""
+        callback_log.append(id(connection))
+
+    # Register a second server config with session_callback
+    with open(SECRETS_FILE) as f:
+        pwd = f.read().strip()
+
+    pool.register_server(
+        server_id="cb_test",
+        host=HOST, port=PORT,
+        user=USER, password=pwd,
+        service_name=SERVICE_NAME,
+        min_connections=1,
+        max_connections=3,
+        session_callback=schema_callback,
+    )
+
+    # Acquire two connections sequentially (pool min=1 so first reuses, second may create)
+    conn_ids = []
+    async with pool.get_connection("cb_test") as conn:
+        conn_ids.append(id(conn))
+    async with pool.get_connection("cb_test") as conn:
+        conn_ids.append(id(conn))
+
+    # Callback must have fired at least once (at pool creation / first acquire)
+    assert len(callback_log) >= 1, (
+        "session_callback never called — pool did not invoke it on connection creation"
+    )
+
+    await pool.close_pool("cb_test")
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_ten_concurrent_queries_no_errors(pool):
+    """
+    Stress: 10 concurrent queries on pool with max_connections=5.
+    All must complete without errors (pool queues excess requests via POOL_GETMODE_WAIT).
+    """
+    errors = []
+
+    async def single_query(i: int):
+        try:
+            async with pool.get_connection("concurrency_test") as conn:
+                with conn.cursor() as cur:
+                    cur.execute("SELECT :i FROM DUAL", {"i": i})
+                    val = cur.fetchone()[0]
+                    assert val == i, f"Expected {i}, got {val}"
+        except Exception as e:
+            errors.append(f"task {i}: {e}")
+
+    t0 = time.perf_counter()
+    await asyncio.gather(*[single_query(i) for i in range(10)])
+    elapsed = time.perf_counter() - t0
+
+    assert not errors, f"Errors in concurrent queries: {errors}"
+    print(f"\n[CONCURRENCY] 10 queries completed in {elapsed*1000:.0f}ms, no errors ✅")