obsidian-wiki/tests/test_business_facts.py

import json
import tempfile
import unittest
from contextlib import redirect_stdout
from io import StringIO
from pathlib import Path

from obsidian_wiki.business_facts import (
    business_db_path_for_vault,
    query_business_question,
    sync_kingdee_orders,
)


class BusinessFactsTests(unittest.TestCase):
    def setUp(self):
        self.tempdir = tempfile.TemporaryDirectory()
        self.vault = Path(self.tempdir.name)
        (self.vault / "_meta").mkdir()
        self.db_path = business_db_path_for_vault(self.vault)

    def tearDown(self):
        self.tempdir.cleanup()

    def test_monthly_order_count_deduplicates_order_lines(self):
        sync_kingdee_orders(
            self.vault,
            [
                {
                    "order_id": "1001",
                    "order_no": "SO-001",
                    "order_date": "2026-06-03",
                    "customer_id": "C01",
                    "customer_name": "Customer A",
                    "total_amount": 1200,
                    "updated_at": "2026-06-03T08:00:00Z",
                    "items": [
                        {"line_id": "1", "product_id": "P01", "product_name": "Part 1", "quantity": 10, "amount": 500},
                        {"line_id": "2", "product_id": "P02", "product_name": "Part 2", "quantity": 20, "amount": 700},
                    ],
                },
                {
                    "order_id": "1002",
                    "order_no": "SO-002",
                    "order_date": "2026-06-15",
                    "customer_id": "C02",
                    "customer_name": "Customer B",
                    "total_amount": 900,
                    "updated_at": "2026-06-15T08:00:00Z",
                    "items": [
                        {"line_id": "1", "product_id": "P03", "product_name": "Part 3", "quantity": 5, "amount": 900},
                    ],
                },
                {
                    "order_id": "1003",
                    "order_no": "SO-003",
                    "order_date": "2026-07-01",
                    "customer_id": "C01",
                    "customer_name": "Customer A",
                    "total_amount": 300,
                    "updated_at": "2026-07-01T08:00:00Z",
                },
            ],
            source_payload={"orders": 3},
            synced_at="2026-07-06T09:00:00Z",
        )

        result = query_business_question(self.vault, "2026-06 order count")

        self.assertEqual(result["template"], "monthly_order_count")
        self.assertEqual(result["data"]["order_count"], 2)
        self.assertIn("2", result["answer"])
        self.assertIn("sales order number", result["answer"])
        self.assertIn("2026-07-06T09:00:00Z", result["answer"])

    def test_repeated_sync_updates_existing_orders_without_duplicates(self):
        first = [
            {
                "order_id": "1001",
                "order_no": "SO-001",
                "order_date": "2026-06-03",
                "customer_id": "C01",
                "customer_name": "Customer A",
                "total_amount": 1200,
                "updated_at": "2026-06-03T08:00:00Z",
            }
        ]
        second = [{**first[0], "total_amount": 1500, "updated_at": "2026-06-04T08:00:00Z"}]

        sync_kingdee_orders(self.vault, first, source_payload={"orders": first}, synced_at="2026-07-06T09:00:00Z")
        sync_kingdee_orders(self.vault, second, source_payload={"orders": second}, synced_at="2026-07-06T10:00:00Z")

        count = query_business_question(self.vault, "2026-06 order count")
        amount = query_business_question(self.vault, "2026-06 sales amount")

        self.assertEqual(count["data"]["order_count"], 1)
        self.assertEqual(amount["data"]["sales_amount"], 1500.0)
        self.assertIn("2026-07-06T10:00:00Z", amount["answer"])

    def test_query_warns_when_last_sync_is_before_today(self):
        sync_kingdee_orders(
            self.vault,
            [
                {
                    "order_id": "1001",
                    "order_no": "SO-001",
                    "order_date": "2026-06-03",
                    "customer_name": "Customer A",
                    "total_amount": 1200,
                }
            ],
            source_payload={"orders": 1},
            synced_at="2026-07-03T09:00:00Z",
        )

        result = query_business_question(self.vault, "2026-06 order count", now="2026-07-06T00:00:00Z")

        self.assertTrue(result["is_stale"])
        self.assertIn("may not include data after 2026-07-03T09:00:00Z", result["answer"])

    def test_cli_sync_payload_can_be_loaded_from_json_file(self):
        payload_path = self.vault / "kingdee.json"
        payload_path.write_text(
            json.dumps(
                {
                    "orders": [
                        {
                            "order_id": "1001",
                            "order_no": "SO-001",
                            "order_date": "2026-06-03",
                            "customer_name": "Customer A",
                            "total_amount": 1200,
                        }
                    ]
                }
            ),
            encoding="utf-8",
        )

        from obsidian_wiki.cli import main

        with redirect_stdout(StringIO()):
            sync_code = main(["business", "sync", "kingdee", "--vault", str(self.vault), "--input", str(payload_path)])
        query_output = StringIO()
        with redirect_stdout(query_output):
            query_code = main(["business", "query", "2026-06 order count", "--vault", str(self.vault), "--json"])

        self.assertEqual(sync_code, 0)
        self.assertEqual(query_code, 0)
        self.assertEqual(json.loads(query_output.getvalue())["data"]["order_count"], 1)


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