Eric-ontology/tests/test_market_engine.py

import json
from pathlib import Path
import unittest

from app.market import MarketService
from app.market.models import InquiryRequest
from app.repository import RDFRepository


ROOT = Path(__file__).resolve().parents[1]
HAIER = "https://gesli.example/sample#Haier"
HISENSE = "https://gesli.example/sample#Hisense"


class MarketEngineTest(unittest.TestCase):
    def setUp(self):
        self.service = MarketService(ROOT, RDFRepository(ROOT))

    def test_default_inquiry_builds_three_ranked_solutions(self):
        result = self.service.recommend(InquiryRequest(customer=HAIER))

        self.assertEqual(
            result["pipeline"],
            {
                "discovered": 9,
                "rejected": 6,
                "feasible": 3,
                "generated_solutions": 3,
            },
        )
        self.assertEqual(result["recommendation"]["solution_type"], "均衡方案")
        self.assertEqual(result["recommendation"]["supplier_name"], "马士基青岛分支机构")
        self.assertEqual(
            {candidate["solution_type"] for candidate in result["candidates"]},
            {"经济方案", "均衡方案", "时效方案"},
        )

    def test_form_only_exposes_routes_with_market_context(self):
        options = self.service.form_options()

        self.assertEqual(options["origins"], ["青岛港", "青岛胶东机场"])
        self.assertEqual(options["destinations"], ["汉堡港", "法兰克福机场", "洛杉矶港"])
        self.assertEqual(options["route_types"], ["中转", "直航", "直达"])
        self.assertEqual(len(options["routes"]), 3)

    def test_customer_profile_changes_recommendation(self):
        haier = self.service.recommend(InquiryRequest(customer=HAIER))
        hisense = self.service.recommend(InquiryRequest(customer=HISENSE))

        self.assertEqual(haier["recommendation"]["solution_type"], "均衡方案")
        self.assertEqual(hisense["recommendation"]["solution_type"], "经济方案")
        self.assertNotEqual(
            haier["recommendation"]["final_quote_amount"],
            hisense["recommendation"]["final_quote_amount"],
        )

    def test_expired_quotes_are_rejected_with_reasons(self):
        result = self.service.recommend(InquiryRequest(customer=HAIER))
        rejected = {item["quote_no"]: item for item in result["rejected_candidates"]}

        self.assertIn("SQ-EGL-20260710-EXPIRED", rejected)
        self.assertTrue(
            any("报价已于2026-07-18失效" in reason for reason in rejected["SQ-EGL-20260710-EXPIRED"]["reasons"])
        )

    def test_dangerous_goods_uses_only_specialized_candidates(self):
        inquiry = InquiryRequest(
            customer="https://gesli.example/sample#Sailun",
            dangerous_goods=True,
        )
        result = self.service.recommend(inquiry)

        self.assertEqual(result["pipeline"]["feasible"], 3)
        self.assertTrue(
            all("危险品" in candidate["product_name"] for candidate in result["candidates"])
        )

    def test_hamburg_route_builds_three_solutions(self):
        result = self.service.recommend(InquiryRequest(
            customer="https://gesli.example/sample#Midea",
            destination="汉堡港",
            requested_departure_date="2026-07-28",
            max_transit_days=36,
            target_amount=42500,
        ))

        self.assertEqual(result["pipeline"]["feasible"], 3)
        self.assertEqual(result["pipeline"]["rejected"], 1)

    def test_time_sensitive_air_customer_gets_express_solution(self):
        result = self.service.recommend(InquiryRequest(
            customer="https://gesli.example/sample#QingdaoMedTech",
            origin="青岛胶东机场",
            destination="法兰克福机场",
            requested_departure_date="2026-07-27",
            max_transit_days=6,
            target_amount=68000,
            preferred_route_type="直达",
        ))

        self.assertEqual(result["recommendation"]["solution_type"], "时效方案")
        self.assertEqual(result["recommendation"]["supplier_name"], "汉莎货运青岛代理")

    def test_sales_response_does_not_expose_supplier_cost(self):
        result = self.service.recommend(InquiryRequest(customer=HAIER))
        serialized = json.dumps(result, ensure_ascii=False)

        self.assertNotIn("supplier_cost", serialized)
        self.assertNotIn("supplierCostAmount", serialized)
        self.assertIn("cost_risk_status", serialized)

    def test_decision_audit_is_recorded_as_rdf_assertions(self):
        result = self.service.recommend(InquiryRequest(customer=HAIER))
        audit = self.service.get_audit(result["run_id"])

        self.assertIsNotNone(audit)
        self.assertEqual(audit["candidate_count"], 3)
        self.assertGreaterEqual(audit["rdf_assertions"], 9)


if __name__ == "__main__":
    unittest.main()