75 lines
2.0 KiB
Python
75 lines
2.0 KiB
Python
from __future__ import annotations
|
||
|
||
import pytest
|
||
|
||
from db import RepoUnavailable
|
||
from repo import OracleSelectRepo
|
||
|
||
|
||
def test_from_env_without_user_raises(monkeypatch):
|
||
monkeypatch.delenv("RPA_ORACLE_USER", raising=False)
|
||
monkeypatch.delenv("RPA_ORACLE_PASSWORD", raising=False)
|
||
monkeypatch.delenv("RPA_ORACLE_HOST", raising=False)
|
||
with pytest.raises(RepoUnavailable):
|
||
OracleSelectRepo.from_env()
|
||
|
||
|
||
def test_save_result_inserts_cart_then_updates_mapping():
|
||
executed: list[tuple[str, dict]] = []
|
||
|
||
class FakeCursor:
|
||
def __enter__(self):
|
||
return self
|
||
|
||
def __exit__(self, *args):
|
||
return False
|
||
|
||
def execute(self, sql, params=None):
|
||
executed.append((sql, params or {}))
|
||
|
||
class FakeConn:
|
||
def __init__(self):
|
||
self.committed = False
|
||
|
||
def cursor(self):
|
||
return FakeCursor()
|
||
|
||
def commit(self):
|
||
self.committed = True
|
||
|
||
def rollback(self):
|
||
raise AssertionError("不应 rollback")
|
||
|
||
def close(self):
|
||
pass
|
||
|
||
conn = FakeConn()
|
||
repo = OracleSelectRepo(connect=lambda: conn)
|
||
repo.save_result(
|
||
2026100201,
|
||
1,
|
||
"选品成功：1号药城",
|
||
[
|
||
{
|
||
"operate_type": 1,
|
||
"oms_id": 2026100201,
|
||
"platform_type": 66,
|
||
"supplier_name": "广东壹号药业有限公司",
|
||
"bar_code": "06901234567890",
|
||
"purchase_price": 90,
|
||
"purchase_qty": 10,
|
||
"purchase_total": 900,
|
||
"status": 1,
|
||
"fail_reason": None,
|
||
"push_status": 0,
|
||
"org_code": "901",
|
||
"multi_platform_flag": 0,
|
||
}
|
||
],
|
||
)
|
||
assert conn.committed is True
|
||
assert "rpa_cart_settlement_goods_data" in executed[0][0].lower()
|
||
assert executed[0][1]["oms_id"] == 2026100201
|
||
assert "rpa_sales_goods_mapping_data" in executed[1][0].lower()
|
||
assert executed[1][1]["deal_flag"] == 1
|