"""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."]