zhixing-system/zhixing-server/tests/unit/sector_radar/test_cli.py

import json
from datetime import date
from decimal import Decimal

import pytest

from zhixing_server.modules.sector_radar.application.build import (
    BuildDateOutcome,
    BuildOutcomeStatus,
    BuildSummary,
)
from zhixing_server.modules.sector_radar.presentation import cli
from zhixing_server.modules.sector_radar.presentation.cli import build_parser


def test_sector_radar_cli_parses_single_range_and_retry_modes() -> None:
    parser = build_parser()

    single = parser.parse_args(["--trade-date", "2026-08-28"])
    date_range = parser.parse_args(["--start-date", "2026-08-18", "--end-date", "2026-08-28"])
    retry = parser.parse_args(["--retry-publication-id", "publication-a"])

    assert single.trade_date == date(2026, 8, 28)
    assert date_range.start_date == date(2026, 8, 18)
    assert date_range.end_date == date(2026, 8, 28)
    assert retry.retry_publication_id == "publication-a"


class FakeSettings:
    log_level = "INFO"
    tushare_token = "secret-token"
    database_url = "postgresql://unused"
    sector_radar_max_retries = 3
    sector_radar_retry_backoff_seconds = 1.0
    sector_radar_request_interval_seconds = 0.2
    sector_radar_advisory_lock_key = 7_380_522
    sector_radar_coverage_threshold = Decimal("0.99")


class FakeRepository:
    closed = False

    def __init__(self, database_url: str, *, advisory_lock_key: int) -> None:
        assert database_url == "postgresql://unused"
        assert advisory_lock_key == 7_380_522

    def close(self) -> None:
        self.closed = True


@pytest.mark.parametrize(
    ("outcome_status", "coverage", "expected_code"),
    (("success", Decimal(1), 0), ("partial", Decimal("0.8"), 2), ("failed", Decimal(0), 1)),
)
def test_cli_main_returns_summary_exit_code_and_json(
    monkeypatch: pytest.MonkeyPatch,
    capsys: pytest.CaptureFixture[str],
    outcome_status: BuildOutcomeStatus,
    coverage: Decimal,
    expected_code: int,
) -> None:
    summary = BuildSummary(
        (
            BuildDateOutcome(
                date(2026, 8, 28),
                outcome_status,
                "publication-a",
                coverage,
                2,
                6,
            ),
        )
    )

    class FakeSourceFactory:
        @staticmethod
        def from_token(token: str, **kwargs: object) -> object:
            assert token == "secret-token"
            assert kwargs
            return object()

    class FakeBuild:
        def __init__(self, source: object, repository: object, **kwargs: object) -> None:
            assert source is not None
            assert repository is not None
            assert kwargs

        def execute(self, command: object) -> BuildSummary:
            assert command is not None
            return summary

    monkeypatch.setattr(cli, "get_settings", FakeSettings)
    monkeypatch.setattr(cli, "TushareSectorRadarAdapter", FakeSourceFactory)
    monkeypatch.setattr(cli, "PostgresSectorRadarRepository", FakeRepository)
    monkeypatch.setattr(cli, "BuildSectorRadar", FakeBuild)

    exit_code = cli.main(["--trade-date", "2026-08-28"])
    output = json.loads(capsys.readouterr().out)

    assert exit_code == expected_code
    assert output["status"] == summary.status
    assert output["exit_code"] == expected_code
    assert "secret-token" not in str(output)


def test_cli_initialization_failure_is_redacted(
    monkeypatch: pytest.MonkeyPatch,
    capsys: pytest.CaptureFixture[str],
) -> None:
    class FailingSourceFactory:
        @staticmethod
        def from_token(token: str, **kwargs: object) -> object:
            del token, kwargs
            raise RuntimeError("private provider detail secret-token")

    monkeypatch.setattr(cli, "get_settings", FakeSettings)
    monkeypatch.setattr(cli, "TushareSectorRadarAdapter", FailingSourceFactory)

    exit_code = cli.main(["--trade-date", "2026-08-28"])
    captured = capsys.readouterr()
    output = json.loads(captured.out)

    assert exit_code == 1
    assert output["status"] == "failed"
    assert output["error_type"] == "RuntimeError"
    assert "private provider detail" not in captured.out
    assert "secret-token" not in captured.out