106 lines
3.2 KiB
Python
106 lines
3.2 KiB
Python
from __future__ import annotations
|
||
|
||
import os
|
||
|
||
import pytest
|
||
|
||
from db import RepoUnavailable, load_runtime_env
|
||
from repo import OracleSelectRepo
|
||
|
||
|
||
def test_load_runtime_env_fills_blank_from_oms_env(tmp_path, monkeypatch):
|
||
monkeypatch.delenv("RPA_ORACLE_HOST", raising=False)
|
||
monkeypatch.delenv("RPA_ORACLE_PASSWORD", raising=False)
|
||
monkeypatch.setenv("RPA_ORACLE_USER", "keep-me")
|
||
select_env = tmp_path / "select.env"
|
||
oms_env = tmp_path / "oms.env"
|
||
select_env.write_text(
|
||
"RPA_ORACLE_HOST=192.168.12.244\nRPA_ORACLE_PASSWORD=\nRPA_ORACLE_USER=from-select\n",
|
||
encoding="utf-8",
|
||
)
|
||
oms_env.write_text(
|
||
"RPA_ORACLE_PASSWORD=not-for-git\nRPA_ORACLE_USER=from-oms\n",
|
||
encoding="utf-8",
|
||
)
|
||
load_runtime_env([select_env, oms_env])
|
||
assert os.environ["RPA_ORACLE_HOST"] == "192.168.12.244"
|
||
assert os.environ["RPA_ORACLE_PASSWORD"] == "not-for-git"
|
||
assert os.environ["RPA_ORACLE_USER"] == "keep-me"
|
||
|
||
|
||
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": "907",
|
||
"multi_platform_flag": 0,
|
||
}
|
||
],
|
||
)
|
||
assert conn.committed is True
|
||
sql = " ".join(executed[0][0].split())
|
||
assert "rpa_cart_settlement_goods_data" in sql.lower()
|
||
assert "create_user_id" in sql
|
||
assert "create_user_name" in sql
|
||
assert "update_time" in sql
|
||
assert "update_user_id" in sql
|
||
assert "update_user_name" in sql
|
||
assert "SYSDATE, 0, '系统管理员'" in sql
|
||
assert sql.count("SYSDATE") == 2
|
||
assert executed[0][1]["oms_id"] == 2026100201
|
||
assert "create_user_name" not in executed[0][1]
|
||
assert "rpa_sales_goods_mapping_data" in executed[1][0].lower()
|
||
assert executed[1][1]["deal_flag"] == 1
|