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"
INDIA_RETAIL = "https://gesli.example/sample#IndiaMockRetail"


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.assertIn("方案优选", {candidate["solution_type"] for candidate in result["candidates"]})
        self.assertIn("relationship", result["candidates"][0])
        self.assertEqual(result["weights"]["price"], 0.03)
        self.assertEqual(result["weights"]["spot_space"], 0.15)

    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"]), 5)

    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"]["supplier_name"], "汉莎货运青岛代理")

    def test_india_solution_and_relationship_outweigh_lowest_purchase_price(self):
        result = self.service.recommend(InquiryRequest(
            customer=INDIA_RETAIL,
            destination="印度那瓦舍瓦港",
            requested_departure_date="2026-07-28",
            max_transit_days=30,
            target_amount=46000,
        ))

        top = result["candidates"][0]
        lowest_price = max(result["candidates"], key=lambda item: item["price_score"])

        self.assertEqual(top["solution_type"], "方案优选")
        self.assertEqual(top["supplier_name"], "印度航线Mock船公司直营")
        self.assertLess(top["price_score"], lowest_price["price_score"])
        self.assertGreater(
            top["relationship_advantage_score"],
            lowest_price["relationship_advantage_score"],
        )
        self.assertEqual(top["relationship"]["tier"], "战略合作")
        self.assertEqual(top["relationship"]["supplier_name"], "印度航线Mock船公司直营")
        self.assertEqual(top["relationship"]["supplier_channel"], "船公司直营")
        self.assertEqual(top["spot_space"]["status"], "可立即订舱")
        self.assertEqual(top["spot_space"]["teu"], 18)
        self.assertGreater(top["spot_space_score"], lowest_price["spot_space_score"])

    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"], 20)
        self.assertIn("business_explanation", audit)
        self.assertIn("现舱", audit["business_explanation"]["selection_basis"])


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