124 lines
4.9 KiB
Python
124 lines
4.9 KiB
Python
import json
|
|
import sys
|
|
import types
|
|
from pathlib import Path
|
|
|
|
# Stub external dependencies if not present in host python
|
|
for mod in ["sqlalchemy", "sqlalchemy.ext", "sqlalchemy.ext.asyncio", "sqlalchemy.orm", "asyncpg", "pydantic", "pydantic_settings", "httpx"]:
|
|
if mod not in sys.modules:
|
|
m = types.ModuleType(mod)
|
|
sys.modules[mod] = m
|
|
|
|
sys.modules["pydantic_settings"].BaseSettings = type("BaseSettings", (), {})
|
|
sys.modules["pydantic_settings"].SettingsConfigDict = lambda **k: None
|
|
sys.modules["pydantic"].BaseModel = type("BaseModel", (), {})
|
|
|
|
# Stub specific symbols needed by app.db and app.models
|
|
sys.modules["sqlalchemy"].select = lambda *a, **k: None
|
|
sys.modules["sqlalchemy"].update = lambda *a, **k: None
|
|
sys.modules["sqlalchemy"].func = types.SimpleNamespace(gen_random_uuid=lambda: None, now=lambda: None)
|
|
sys.modules["sqlalchemy"].BigInteger = type("BigInteger", (), {})
|
|
sys.modules["sqlalchemy"].DateTime = lambda *a, **k: None
|
|
sys.modules["sqlalchemy"].Index = lambda *a, **k: None
|
|
sys.modules["sqlalchemy"].Integer = type("Integer", (), {})
|
|
sys.modules["sqlalchemy"].String = lambda *a, **k: None
|
|
sys.modules["sqlalchemy"].Text = type("Text", (), {})
|
|
sys.modules["sqlalchemy.dialects.postgresql"] = types.ModuleType("sqlalchemy.dialects.postgresql")
|
|
sys.modules["sqlalchemy.dialects.postgresql"].JSONB = lambda *a, **k: None
|
|
sys.modules["sqlalchemy.dialects.postgresql"].UUID = lambda *a, **k: None
|
|
sys.modules["sqlalchemy.orm"].DeclarativeBase = type("DeclarativeBase", (), {})
|
|
sys.modules["sqlalchemy.orm"].Mapped = type("Mapped", (), {})
|
|
sys.modules["sqlalchemy.orm"].mapped_column = lambda *a, **k: None
|
|
sys.modules["sqlalchemy.ext.asyncio"].create_async_engine = lambda *a, **k: None
|
|
sys.modules["sqlalchemy.ext.asyncio"].async_sessionmaker = lambda *a, **k: None
|
|
|
|
# Add 'ai procurement' to sys.path
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "ai procurement"))
|
|
|
|
from app.drafts import (
|
|
new_quote_payload,
|
|
merge_quote_patch,
|
|
build_quote_request,
|
|
compute_quote_missing,
|
|
)
|
|
|
|
def test_wizard_conversion():
|
|
mock_path = Path(__file__).resolve().parents[1] / "tests" / "wizard_mock_data.json"
|
|
with open(mock_path, encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
|
|
# 1. Start with clean payload
|
|
payload = new_quote_payload()
|
|
|
|
# 2. Extract patch from wizard mock data (simulating NLU patch extraction)
|
|
s1 = data["step1_topic_and_supplier"]
|
|
s2 = data["step2_line_items"]
|
|
s3 = data["step3_termsheets"]
|
|
s4 = data["step4_projects"]
|
|
s5 = data["step5_review_and_metadata"]
|
|
|
|
patch = {
|
|
"supplier_name": s1["supplier_name"],
|
|
"is_official": s1["is_official"],
|
|
"project_names": s4["projects"],
|
|
"expiry_date": s5["expiry_date"],
|
|
"notes": s5["notes"],
|
|
"items": [
|
|
{
|
|
"material_name": it["material_name"],
|
|
"brand_name": it["brand"],
|
|
"specs": it["specs"],
|
|
"quantity": it["quantity"],
|
|
"unit_name": it["unit"],
|
|
"unit_price": it["unit_price"],
|
|
}
|
|
for it in s2
|
|
],
|
|
"payment_terms": [
|
|
{
|
|
"label": ts["label"],
|
|
"milestones": ts["milestones"],
|
|
}
|
|
for ts in s3
|
|
],
|
|
}
|
|
|
|
# 3. Merge patch into payload
|
|
changed = merge_quote_patch(payload, patch)
|
|
assert changed, "Patch should have changed payload"
|
|
|
|
# 4. Simulate resolution
|
|
payload["supplier_id"] = "11111111-1111-1111-1111-111111111111"
|
|
payload["project_ids"] = ["22222222-2222-2222-2222-222222222222"]
|
|
for i, it in enumerate(payload["items"], start=1):
|
|
it["material_id"] = f"33333333-3333-3333-3333-33333333333{i}"
|
|
it["unit_id"] = 3
|
|
|
|
# 5. Build QuoteCreate request
|
|
body, problems = build_quote_request(payload)
|
|
assert problems == [], f"Expected no problems, got {problems}"
|
|
assert body is not None, "Body should not be None"
|
|
|
|
assert body["supplier_id"] == "11111111-1111-1111-1111-111111111111"
|
|
assert body["is_official"] is True
|
|
assert len(body["line_items"]) == 2
|
|
assert body["line_items"][0]["temp_id"] == "t1"
|
|
assert "A3، سایز ۱۴" in body["line_items"][0]["notes"]
|
|
assert "فولاد مبارکه" in body["line_items"][0]["notes"]
|
|
|
|
assert len(body["termsheets"]) == 1
|
|
assert body["termsheets"][0]["temp_id"] == "p1"
|
|
assert len(body["termsheets"][0]["schedule_steps"]) == 3
|
|
assert body["termsheets"][0]["schedule_steps"][0]["step_no"] == 1
|
|
assert body["termsheets"][0]["schedule_steps"][0]["amount"] == 300000000.0
|
|
|
|
assert len(body["item_termsheets"]) == 2
|
|
assert body["item_termsheets"][0]["line_item_temp_id"] == "t1"
|
|
assert body["item_termsheets"][0]["termsheet_temp_id"] == "p1"
|
|
assert body["item_termsheets"][0]["unit_price"] == 48500.0
|
|
|
|
print("ALL WIZARD CONVERSION ASSERTIONS PASSED SUCCESSFULLY! [OK]")
|
|
|
|
if __name__ == "__main__":
|
|
test_wizard_conversion()
|