DRRK/tests/smoke/test_ui_smoke.py

"""UI без браузера: streamlit.testing.v1.AppTest гоняет main.py и страницы
в этом же процессе. БД — временный SQLite (фикстура session_local),
MOEX/ЦБ — фейки на фикстурах; сеть не нужна."""

from __future__ import annotations

from datetime import date
from pathlib import Path

import pytest
from streamlit.testing.v1 import AppTest

from app.integrations.cbr import parse_dynamic
from app.services import fx_rates, market_data, securities
from tests.unit.fakes import FakeCbr

ROOT = Path(__file__).resolve().parents[2]
USER = "analyst@rshb.ru"
TIMEOUT = 60


def _logged_in(at: AppTest) -> AppTest:
    """Профиль для mock-провайдера (LOCATION=local), как если бы его ввели в форме."""
    at.session_state["local_user_email"] = USER
    at.session_state["local_username"] = "analyst"
    at.session_state["local_full_name"] = "Аналитик ДРРК"
    return at


def _securities_page() -> None:
    from app.ui.pages import securities as page

    page.render()


def _market_page() -> None:
    from app.ui.pages import market_data as page

    page.render()


@pytest.fixture(autouse=True)
def _local_mode(monkeypatch):
    monkeypatch.setenv("LOCATION", "local")
    monkeypatch.delenv("OPENBAO__SETTINGS__LOCATION", raising=False)


@pytest.fixture
def fake_moex_from_settings(monkeypatch, fixture_moex):
    """Сервисы создают клиента через MoexClient.from_settings() — подменяем."""

    class _Factory:
        @staticmethod
        def from_settings():
            return fixture_moex

    monkeypatch.setattr(securities, "MoexClient", _Factory)
    monkeypatch.setattr(market_data, "MoexClient", _Factory)
    return fixture_moex


def test_main_without_profile_shows_local_login_form(session_local) -> None:
    at = AppTest.from_file(str(ROOT / "main.py"), default_timeout=TIMEOUT).run()

    assert not at.exception
    assert [w.label for w in at.text_input] == ["Email", "Username", "Full name"]


def test_main_renders_default_page_for_logged_in_user(session_local) -> None:
    at = _logged_in(AppTest.from_file(str(ROOT / "main.py"), default_timeout=TIMEOUT))
    at.run()

    assert not at.exception
    assert [h.value for h in at.header] == ["Бумаги"]
    assert at.session_state["user_profile"]["email"] == USER
    assert any("Бумаг пока нет" in i.value for i in at.info)


def test_securities_page_lookup_preview_and_save(
    session_local, fake_moex_from_settings
) -> None:
    at = _logged_in(AppTest.from_function(_securities_page, default_timeout=TIMEOUT))
    at.session_state["user_profile"] = {"email": USER}
    at.run()

    at.text_input[0].input("ru000a106169")
    at.button[0].click().run()  # «Найти на Мосбирже» (submit формы)

    assert not at.exception
    assert not at.error, [e.value for e in at.error]
    assert any("РЭО-02" in m.value for m in at.markdown)

    save = next(b for b in at.button if b.label == "Сохранить")
    save.click().run()

    assert not at.exception
    assert not at.error, [e.value for e in at.error]
    table = at.dataframe[0].value
    assert list(table["ISIN"]) == ["RU000A106169"]
    assert list(table["Наименование"]) == ["РЭО-02"]
    assert list(table["Валюта"]) == ["RUB"]


def test_securities_page_unknown_isin_shows_error(
    session_local, fake_moex_from_settings
) -> None:
    at = AppTest.from_function(_securities_page, default_timeout=TIMEOUT)
    at.session_state["user_profile"] = {"email": USER}
    at.run()

    at.text_input[0].input("RU000A999999")
    at.button[0].click().run()

    assert not at.exception
    assert any("не найдена на Мосбирже" in e.value for e in at.error)


def test_market_page_update_buttons_and_tables(
    session_local, fake_moex_from_settings, add_trade, fixtures_dir: Path, monkeypatch
) -> None:
    from app.db.base import run_db

    run_db(securities.add_by_isin("RU000A106169", USER, moex=fake_moex_from_settings))
    cbr = FakeCbr(parse_dynamic((fixtures_dir / "cbr_dynamic_R01375.xml").read_bytes()))

    class _CbrFactory:
        @staticmethod
        def from_settings():
            return cbr

    monkeypatch.setattr(fx_rates, "CbrClient", _CbrFactory)

    at = AppTest.from_function(_market_page, default_timeout=TIMEOUT)
    at.session_state["user_profile"] = {"email": USER}
    at.run()
    at.date_input[0].set_value(date(2026, 10, 8)).run()

    assert not at.exception
    assert any("котировок в базе нет" in i.value for i in at.info)

    next(b for b in at.button if b.label == "Обновить котировки").click().run()
    assert not at.exception
    assert any("Нет бумаг с открытой позицией" in i.value for i in at.info)

    add_trade("RU000A106169", "BUY", date(2026, 10, 1), "100")
    next(b for b in at.button if b.label == "Обновить котировки").click().run()
    assert not at.exception
    assert any("Обновлено: 1" in s.value for s in at.success)
    quotes = at.dataframe[0].value
    assert list(quotes["Чистая цена, %"]) == ["101.42"]
    assert list(quotes["Источник цены"]) == ["WAPRICE"]

    next(b for b in at.button if b.label == "Обновить курс CNY").click().run()
    assert not at.exception
    assert any("Сохранено курсов CNY: 6" in s.value for s in at.success)
    rates = at.dataframe[1].value
    assert list(rates["За 1 CNY"])[0] == "12.7516"


def _with_reo_quote(moex, add_trade) -> None:  # noqa: ANN001
    from app.db.base import run_db

    run_db(securities.add_by_isin("RU000A106169", USER, moex=moex))
    add_trade("RU000A106169", "BUY", date(2026, 1, 1), "100")
    run_db(market_data.update(date(2026, 10, 8), "system", moex=moex))


def test_securities_page_nominal_history_and_manual_value(
    session_local, fake_moex_from_settings, add_trade
) -> None:
    _with_reo_quote(fake_moex_from_settings, add_trade)
    at = AppTest.from_function(_securities_page, default_timeout=TIMEOUT)
    at.session_state["user_profile"] = {"email": USER}
    at.run()

    history = at.dataframe[1].value
    assert list(history["Номинал"]) == ["645.19"]
    assert list(history["Источник"]) == ["MOEX"]

    save = next(b for b in at.button if b.label == "Сохранить номинал")
    at.text_input[1].input("640,5")  # [0] — ISIN в форме добавления
    save.click().run()
    assert [e.value for e in at.error] == ["Укажите причину правки."]

    at.date_input[0].set_value(date(2026, 10, 9))
    at.text_input[1].input("640,5")
    at.text_input[2].input("частичное погашение")
    next(b for b in at.button if b.label == "Сохранить номинал").click().run()

    assert not at.exception
    assert not at.error, [e.value for e in at.error]
    history = at.dataframe[1].value
    assert list(history["Номинал"]) == ["640.5", "645.19"]
    assert list(history["Источник"]) == ["ручной", "MOEX"]
    assert list(history["Причина"])[0] == "частичное погашение"


def test_market_page_manual_price(
    session_local, fake_moex_from_settings, add_trade
) -> None:
    _with_reo_quote(fake_moex_from_settings, add_trade)
    at = AppTest.from_function(_market_page, default_timeout=TIMEOUT)
    at.session_state["user_profile"] = {"email": USER}
    at.run()
    at.date_input[0].set_value(date(2026, 10, 8)).run()

    assert any("Ручных правок за эту дату нет" in c.value for c in at.caption)
    assert at.text_input[0].placeholder == "MOEX: 101.42"

    at.text_input[0].input("100,1")
    at.text_input[3].input("цена по RFQ")
    next(b for b in at.button if b.label == "Сохранить правку").click().run()

    assert not at.exception
    assert not at.error, [e.value for e in at.error]
    quotes = at.dataframe[0].value
    assert list(quotes["Чистая цена, %"]) == ["100.1"]
    assert list(quotes["Источник цены"]) == ["ручная"]
    assert list(quotes["Источник доходности"]) == ["YIELDATWAP"]
    manual = at.dataframe[1].value
    assert list(manual["Цена, MOEX"]) == ["101.42"]
    assert list(manual["Причина"]) == ["цена по RFQ"]
    assert list(manual["Кто"]) == [USER]
    assert at.text_input[0].value == "100.1"  # форма показывает действующую правку


# --- загрузка сделок QUIK ---------------------------------------------------
# AppTest не умеет загружать файлы в file_uploader, поэтому кладём результат
# разбора в session_state так же, как это делает on_change загрузчика.


def _quik_page() -> None:
    from app.ui.pages import quik_import as page

    page.render()


def _quik_app(upload) -> AppTest:  # noqa: ANN001
    at = AppTest.from_function(_quik_page, default_timeout=TIMEOUT)
    at.session_state["user_profile"] = {"email": USER}
    if upload is not None:
        at.session_state["quik_upload"] = upload
    return at.run()


def _fixture_upload(fixtures_dir: Path, comments: list[str] | None = None):
    from dataclasses import replace

    from app.imports.quik import parse_quik_xlsx
    from app.ui.pages.quik_import import QuikUpload

    result = parse_quik_xlsx(
        (fixtures_dir / "quik_sample_300326_anon.xlsx").read_bytes()
    )
    if comments is not None:
        result.trades = [
            replace(t, comment=c) for t, c in zip(result.trades, comments, strict=True)
        ]
    return QuikUpload("сделки_300326.xlsx", result)


def _add_fixture_security(session_local) -> None:
    from app.db.base import run_db
    from app.db.models import Security

    async def _add() -> None:
        async with session_local() as session:
            session.add(
                Security(
                    isin="RU000A10EVH2",
                    secid="RU000A10EVH2",
                    name="ЕврХол3P06",
                    currency_id=1,
                    created_by=USER,
                    updated_by=USER,
                )
            )
            await session.commit()

    run_db(_add())


def test_quik_page_without_file(session_local) -> None:
    at = _quik_app(None)

    assert not at.exception
    assert [h.value for h in at.header] == ["Загрузка сделок QUIK"]
    assert not at.info and not at.button


def test_quik_page_format_error_is_shown(session_local) -> None:
    from app.ui.pages.quik_import import QuikUpload

    at = _quik_app(QuikUpload("old.xls", None, "Не удалось открыть файл как .xlsx."))

    assert not at.exception
    assert [e.value for e in at.error] == ["Не удалось открыть файл как .xlsx."]


def test_quik_page_preview_of_real_fixture(session_local, fixtures_dir: Path) -> None:
    _add_fixture_security(session_local)

    at = _quik_app(_fixture_upload(fixtures_dir))

    assert not at.exception
    assert not at.error, [e.value for e in at.error]
    assert at.selectbox[0].value == "SALES"  # по комментарию /pp
    assert [m.value for m in at.metric] == ["6", "0", "0", "6"]
    rows = at.dataframe[0].value
    assert list(rows["Строка"]) == [2, 3, 4, 5, 6, 7]
    assert set(rows["ISIN"]) == {"RU000A10EVH2"}
    assert list(rows["Комиссия ТС"])[0] == "8217.6"
    totals = at.dataframe[1].value
    assert list(totals["Операция"]) == ["BUY", "SELL"]
    assert list(totals["Кол-во"]) == ["1098000", "1080000"]
    assert any("Файл проверен: 6 строк в SALES" in s.value for s in at.success)
    assert at.button[0].label == "Загрузить в SALES: 6 стр."
    assert not at.button[0].disabled


def test_quik_page_saves_operations(session_local, fixtures_dir: Path) -> None:
    from sqlalchemy import func, select

    from app.db.base import run_db
    from app.db.models import ImportBatch, Operation

    _add_fixture_security(session_local)
    at = _quik_app(_fixture_upload(fixtures_dir))

    at.button[0].click().run()

    assert not at.exception
    assert not at.error, [e.value for e in at.error]
    # После st.rerun() AppTest оставляет в дереве элементы прерванного прогона,
    # поэтому проверяем первый элемент страницы, а не весь список.
    assert at.success[0].value == (
        "«сделки_300326.xlsx»: загружено в SALES 6 стр., исключено 0."
    )
    assert "quik_upload" not in at.session_state  # файл сброшен
    assert at.session_state["quik_file_nonce"] == 1  # поле загрузки — новое

    async def _counts() -> tuple[int, int]:
        async with session_local() as session:
            return (
                await session.scalar(select(func.count(ImportBatch.id))),
                await session.scalar(select(func.count(Operation.id))),
            )

    assert run_db(_counts()) == (1, 6)


def test_quik_page_explains_why_save_is_disabled(
    session_local, fixtures_dir: Path
) -> None:
    at = _quik_app(_fixture_upload(fixtures_dir))  # бумаги нет в справочнике

    assert not at.exception
    assert at.button[0].disabled
    assert any(
        "Загрузить нельзя: нет в справочнике бумаг: ЕврХол3P06" in c.value
        for c in at.caption
    )


def test_quik_page_other_portfolio_and_unknown_rows(
    session_local, fixtures_dir: Path
) -> None:
    _add_fixture_security(session_local)
    upload = _fixture_upload(fixtures_dir, ["/pp", "/pp", "/uw", "/trde", "", "/pp"])

    at = _quik_app(upload)

    assert not at.exception
    assert any("Строки без портфеля: 2" in s.value for s in at.subheader)
    assert any("Ждут решения строк без портфеля: 2" in i.value for i in at.info)
    assert not at.success

    at.radio[0].set_value("Исключить все").run()

    assert not at.exception
    assert any("Пойдёт в SALES: 3" in s.value for s in at.subheader)
    assert any("исключено 2" in s.value for s in at.success)

    at.radio[0].set_value("Отнести все к SALES").run()
    assert any("Пойдёт в SALES: 5" in s.value for s in at.subheader)

    at.selectbox[0].set_value("UW").run()
    assert not at.exception
    assert any("Пойдёт в UW: 1" in s.value for s in at.subheader)


def test_quik_page_rejects_same_file_same_portfolio(
    session_local, fixtures_dir: Path
) -> None:
    from datetime import date

    from app.db.base import run_db
    from app.repositories.import_batches import ImportBatchRepository

    _add_fixture_security(session_local)
    upload = _fixture_upload(fixtures_dir)

    async def _loaded() -> None:
        async with session_local() as session:
            await ImportBatchRepository(session).add(
                source="QUIK",
                portfolio_id=2,  # SALES
                file_name="сделки_300326.xlsx",
                content_sha256=upload.result.content_hash,
                date_from=date(2026, 3, 30),
                date_to=date(2026, 3, 30),
                rows_total=6,
                loaded_rows=[1, 2, 3, 4, 5, 6],
                rows_excluded=0,
                user=USER,
            )
            await session.commit()

    run_db(_loaded())

    at = _quik_app(upload)

    assert not at.exception
    assert any("Этот файл уже загружен в SALES" in e.value for e in at.error)
    assert [m.value for m in at.metric] == ["6", "0", "6", "0"]
    assert not at.success


# --- ручные операции -------------------------------------------------------


def _manual_page() -> None:
    from app.ui.pages import manual_operations as page

    page.render()


def _manual_app() -> AppTest:
    at = AppTest.from_function(_manual_page, default_timeout=TIMEOUT)
    at.session_state["user_profile"] = {"email": USER}
    return at.run()


def test_manual_page_without_securities(session_local) -> None:
    at = _manual_app()

    assert not at.exception
    assert any("Бумаг пока нет" in i.value for i in at.info)
    assert any("Ручных операций пока нет" in c.value for c in at.caption)


def test_manual_page_adds_coupon(session_local) -> None:
    _add_fixture_security(session_local)
    at = _manual_app()
    assert at.radio[0].value == "CPN"
    assert at.button[0].label == "Сохранить: купон"

    at.text_input[0].input("125 000,50")
    at.text_area[0].input("купон №3")
    at.button[0].click().run()

    assert not at.exception
    assert not at.error, [e.value for e in at.error]
    assert any("Сохранено: купон, RU000A10EVH2" in s.value for s in at.success)
    rows = at.dataframe[0].value
    assert list(rows["Операция"]) == ["Купон"]
    assert list(rows["Сумма"]) == ["125000.5"]
    assert list(rows["Валюта"]) == ["RUB"]
    assert list(rows["Кто ввёл"]) == [USER]


def test_manual_page_validation_and_uwfee_portfolio(session_local) -> None:
    _add_fixture_security(session_local)
    at = _manual_app()

    at.button[0].click().run()  # купон без суммы
    assert [e.value for e in at.error] == ["Укажите сумму."]

    at.radio[0].set_value("UWFEE").run()
    assert not at.exception
    assert at.selectbox[0].options == ["UW — Андеррайтинг"]
    assert at.button[0].label == "Сохранить: комиссия за андеррайтинг"
    assert len(at.text_input) == 1  # без количества


# --- фондирование ----------------------------------------------------------


def _funding_page() -> None:
    from app.ui.pages import funding as page

    page.render()


def test_funding_page_saves_rate_and_shows_it_on_date(session_local) -> None:
    at = AppTest.from_function(_funding_page, default_timeout=TIMEOUT)
    at.session_state["user_profile"] = {"email": USER}
    at.run()

    assert not at.exception
    assert at.selectbox[0].options == ["UW — Андеррайтинг", "SALES — Продажи"]
    assert any("Ставок пока нет" in c.value for c in at.caption)
    on_date = at.dataframe[0].value
    assert list(on_date["Валюта"]) == ["CNY", "RUB", "USD"]
    assert on_date["Ставка, % год."].isna().all()

    save = next(b for b in at.button if b.label == "Сохранить ставку")
    save.click().run()
    assert [e.value for e in at.error] == ["Укажите ставку."]

    at.selectbox[1].set_value("RUB")
    at.date_input[1].set_value(date(2026, 7, 28))
    at.text_input[0].input("18,5")
    next(b for b in at.button if b.label == "Сохранить ставку").click().run()

    assert not at.exception
    assert not at.error, [e.value for e in at.error]
    history = at.dataframe[1].value
    assert list(history["Ставка, % год."]) == ["18.5"]
    assert list(history["Валюта"]) == ["RUB"]
    assert list(history["Кто"]) == [USER]

    at.date_input[0].set_value(date(2026, 7, 27)).run()
    assert at.dataframe[0].value["Ставка, % год."].isna().all()
    at.date_input[0].set_value(date(2026, 10, 8)).run()
    assert list(at.dataframe[0].value["Ставка, % год."]) == [None, "18.5", None]

    at.selectbox[0].set_value("SALES").run()
    assert any("Ставок пока нет" in c.value for c in at.caption)


# --- остатки ---------------------------------------------------------------


def _positions_page() -> None:
    from app.ui.pages import positions as page

    page.render()


def test_positions_page(session_local, fixtures_dir: Path) -> None:
    from app.db.base import run_db
    from app.imports.quik import parse_quik_xlsx
    from app.services import quik_import

    at = AppTest.from_function(_positions_page, default_timeout=TIMEOUT)
    at.session_state["user_profile"] = {"email": USER}
    at.run()
    assert not at.exception
    assert any("Открытых позиций на эту дату нет" in c.value for c in at.caption)

    _add_fixture_security(session_local)
    parsed = parse_quik_xlsx(
        (fixtures_dir / "quik_sample_300326_anon.xlsx").read_bytes()
    )
    plan = quik_import.build_plan(
        parsed.trades, run_db(quik_import.load_context(parsed)), "SALES", {}
    )
    run_db(
        quik_import.save_import(
            parsed,
            file_name="сделки_300326.xlsx",
            portfolio_code="SALES",
            decisions={},
            expected_rows=[row.trade.ordinal for row in plan.to_load],
            user=USER,
        )
    )
    bought = sum(t.quantity for t in parsed.trades if t.operation == "BUY")
    sold = sum(t.quantity for t in parsed.trades if t.operation == "SELL")

    at.date_input[0].set_value(date(2026, 3, 30)).run()
    assert not at.exception
    rows = at.dataframe[0].value
    assert list(rows["Портфель"]) == ["SALES"]
    assert list(rows["ISIN"]) == ["RU000A10EVH2"]
    assert list(rows["Остаток, шт"]) == [format((bought - sold).normalize(), "f")]
    assert not at.warning
    assert len(at.get("download_button")) == 1

    at.selectbox[0].set_value("UW").run()
    assert any("Открытых позиций на эту дату нет" in c.value for c in at.caption)


# --- производственный календарь ---------------------------------------------


def _calendar_page() -> None:
    from app.ui.pages import business_calendar as page

    page.render()


def test_calendar_page_saves_year(session_local, fixtures_dir: Path) -> None:
    from app.imports.xmlcalendar import parse_calendar_json

    at = AppTest.from_function(_calendar_page, default_timeout=TIMEOUT)
    at.session_state["user_profile"] = {"email": USER}
    at.run()
    assert not at.exception
    assert any("Календарь пока не загружен" in c.value for c in at.caption)
    assert any("не загружен" in w.value for w in at.warning)

    at.session_state["calendar_upload"] = parse_calendar_json(
        (fixtures_dir / "xmlcalendar_ru_2025.json").read_bytes()
    )
    at.run()
    assert not at.exception
    text = " ".join(m.value for m in at.markdown)
    assert "Год 2025: нерабочих дней 118, сокращённых 4." in text
    assert "Рабочие выходные: 01.11." in text

    next(b for b in at.button if b.label == "Сохранить календарь на 2025").click()
    at.run()
    assert not at.exception
    assert not at.error, [e.value for e in at.error]
    years = at.dataframe[0].value
    assert list(years["Год"]) == [2025]
    assert list(years["Рабочие выходные"]) == [1]
    assert list(years["Кто загрузил"]) == [USER]


def test_calendar_page_shows_format_error(session_local) -> None:
    at = AppTest.from_function(_calendar_page, default_timeout=TIMEOUT)
    at.session_state["user_profile"] = {"email": USER}
    at.session_state["calendar_upload"] = "Файл не читается как JSON."
    at.run()

    assert [e.value for e in at.error] == ["Файл не читается как JSON."]