CoinOX/tests/test_options_review_chart_async_lib.py

"""期权复盘自动 K 线:单元测试."""
from __future__ import annotations

import sqlite3
import unittest
from unittest.mock import MagicMock, patch

from lib.options.options_review_chart_async_lib import (
    OPTIONS_REVIEW_AUTO_CHART_TF,
    build_options_review_chart_filename,
    merge_auto_chart_into_images,
    schedule_options_review_exchange_chart,
    underlying_to_okx_perp_symbol,
)
from lib.options.options_review_db import init_options_review_tables
from lib.options.options_review_lib import save_review_entry, upsert_option_history_row


def _conn() -> sqlite3.Connection:
    c = sqlite3.connect(":memory:")
    c.row_factory = sqlite3.Row
    init_options_review_tables(c)
    return c


class OptionsReviewChartAsyncTests(unittest.TestCase):
    def test_underlying_to_okx_perp_symbol(self):
        self.assertEqual(
            underlying_to_okx_perp_symbol(
                "ETH-USD",
                normalize_symbol_input_fn=lambda s: f"{s}/USDT",
                normalize_okx_symbol_fn=lambda s: f"{s.split('/')[0]}/USDT:USDT"
                if "/" in s
                else s,
            ),
            "ETH/USDT:USDT",
        )

    def test_merge_auto_chart_keeps_manual_slots(self):
        merged = merge_auto_chart_into_images(
            [
                {"tf": "auto", "file": "old.png"},
                {"tf": "15m", "file": "m15.png"},
            ],
            "new_auto.png",
        )
        self.assertEqual(merged[0]["tf"], OPTIONS_REVIEW_AUTO_CHART_TF)
        self.assertEqual(merged[0]["file"], "new_auto.png")
        self.assertEqual(merged[1], {"tf": "15m", "file": "m15.png"})

    def test_chart_filename(self):
        self.assertEqual(
            build_options_review_chart_filename(12), "options_journal_chart_12.png"
        )

    @patch("lib.options.options_review_chart_async_lib.threading.Thread")
    def test_schedule_starts_daemon_thread(self, mock_thread_cls):
        thread = MagicMock()
        mock_thread_cls.return_value = thread
        schedule_options_review_exchange_chart(
            trade_id=7,
            exchange_symbol="ETH/USDT:USDT",
            title_prefix="t",
            journal_tfs=["15m", "1h"],
            journal_limit=100,
            marker_payload={},
            upload_folder="/tmp",
            generate_chart_fn=MagicMock(),
            get_db_fn=MagicMock(),
            now_str_fn=lambda: "2026-09-15 12:00:00",
        )
        mock_thread_cls.assert_called_once()
        self.assertTrue(mock_thread_cls.call_args.kwargs.get("daemon"))
        thread.start.assert_called_once()

    def test_writeback_merges_auto_image(self):
        import tempfile
        from pathlib import Path

        from lib.options.options_review_images_lib import parse_options_review_images_json

        with tempfile.TemporaryDirectory() as td:
            db_path = Path(td) / "t.db"

            def get_db():
                c = sqlite3.connect(str(db_path))
                c.row_factory = sqlite3.Row
                return c

            conn = get_db()
            init_options_review_tables(conn)
            upsert_option_history_row(
                conn,
                {
                    "history_key": "ex:chart1",
                    "inst_id": "ETH-USD-260328-2000-C",
                    "underlying": "ETH",
                    "opt_type": "C",
                    "realized_pnl": 1.0,
                    "created_at": "2026-09-01 10:00:00",
                    "closed_at": "2026-09-01 12:00:00",
                },
            )
            trade_id = conn.execute(
                "SELECT id FROM options_review_trades WHERE history_key='ex:chart1'"
            ).fetchone()["id"]
            save_review_entry(
                conn,
                trade_id,
                {
                    "strategy_tag": "测试",
                    "images": [{"tf": "15m", "file": "options_journal_x_15m.png"}],
                },
            )
            conn.commit()
            conn.close()

            chart_name = build_options_review_chart_filename(trade_id)
            generate = MagicMock(return_value=chart_name)

            with patch(
                "lib.options.options_review_chart_async_lib.threading.Thread"
            ) as mock_thread_cls:
                thread = MagicMock()
                mock_thread_cls.return_value = thread
                schedule_options_review_exchange_chart(
                    trade_id=trade_id,
                    exchange_symbol="ETH/USDT:USDT",
                    title_prefix="ETH review",
                    journal_tfs=["15m", "1h"],
                    journal_limit=200,
                    marker_payload={"entry_ts_ms": 1, "exit_ts_ms": 2},
                    upload_folder=td,
                    generate_chart_fn=generate,
                    get_db_fn=get_db,
                    now_str_fn=lambda: "2026-09-15 12:00:00",
                )
                target = mock_thread_cls.call_args.kwargs["target"]
                target()

            conn = get_db()
            try:
                row = conn.execute(
                    "SELECT images_json, image FROM options_review_entries WHERE trade_id=?",
                    (trade_id,),
                ).fetchone()
                self.assertEqual(row["image"], chart_name)
                images = parse_options_review_images_json(row["images_json"])
                self.assertEqual(images[0]["tf"], "auto")
                self.assertEqual(images[0]["file"], chart_name)
                self.assertTrue(any(x.get("tf") == "15m" for x in images))
            finally:
                conn.close()


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