Dmitry
4552efbf5d
Отменённые и незавершённые операции удаляются из event по raw_tinvest_operation, а не только из текущего окна синка. Бумага, которую MOEX не котирует (структурные ноты, внебиржевые облигации), получает цену из портфеля брокера. Логотип и цвет бренда копируются в instrument (миграция e8b21f6a90c3).
89 lines
3.2 KiB
Python
89 lines
3.2 KiB
Python
"""Issuer logos: named by the broker, filled onto instruments, exposed as a public URL."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from sqlalchemy import select
|
|
|
|
from factories import make_instrument
|
|
from fintracker.branding import logo_url
|
|
from fintracker.db import get_sessionmaker
|
|
from fintracker.models import Instrument, RawTinvestInstrument
|
|
from fintracker.sources.tinvest.sync import _fill_logos
|
|
|
|
|
|
def test_the_url_points_at_the_160_px_picture_without_the_extension_twice():
|
|
assert logo_url("sber.png") == "https://invest-brands.cdn-tinkoff.ru/sberx160.png"
|
|
assert logo_url("pik1.png") == "https://invest-brands.cdn-tinkoff.ru/pik1x160.png"
|
|
assert logo_url("noext") == "https://invest-brands.cdn-tinkoff.ru/noextx160.png"
|
|
|
|
|
|
def test_no_logo_is_no_url():
|
|
assert logo_url(None) is None
|
|
assert logo_url("") is None
|
|
|
|
|
|
async def _raw(uid: str, *, isin: str | None, brand: object) -> None:
|
|
async with get_sessionmaker()() as session:
|
|
session.add(
|
|
RawTinvestInstrument(
|
|
uid=uid, kind="share", isin=isin, figi=None, ticker=None, payload={"brand": brand}
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
|
|
async def _logo(instrument_id: int) -> tuple[str | None, str | None]:
|
|
async with get_sessionmaker()() as session:
|
|
row = (
|
|
await session.execute(select(Instrument).where(Instrument.id == instrument_id))
|
|
).scalar_one()
|
|
return row.logo_name, row.logo_color
|
|
|
|
|
|
async def _set(instrument_id: int, **fields: object) -> None:
|
|
async with get_sessionmaker()() as session:
|
|
row = await session.get(Instrument, instrument_id)
|
|
assert row is not None
|
|
for key, value in fields.items():
|
|
setattr(row, key, value)
|
|
await session.commit()
|
|
|
|
|
|
async def fill() -> int:
|
|
async with get_sessionmaker()() as session:
|
|
filled = await _fill_logos(session)
|
|
await session.commit()
|
|
return filled
|
|
|
|
|
|
async def test_a_logo_is_matched_by_uid(app):
|
|
sber = await make_instrument(ticker="SBER")
|
|
await _set(sber, tinvest_uid="uid-sber")
|
|
await _raw("uid-sber", isin=None, brand={"logo_name": "sber.png", "logo_base_color": "#21A038"})
|
|
|
|
assert await fill() == 1
|
|
assert await _logo(sber) == ("sber.png", "#21A038")
|
|
|
|
|
|
async def test_a_paper_without_a_uid_is_matched_by_isin(app):
|
|
"""Imported from a Sber report: no T-Invest uid, but the same ISIN as one T-Invest knows."""
|
|
lkoh = await make_instrument(ticker="LKOH")
|
|
await _set(lkoh, isin="RU0009024277")
|
|
await _raw("uid-lkoh", isin="RU0009024277", brand={"logo_name": "lukoil.png"})
|
|
|
|
assert await fill() == 1
|
|
assert await _logo(lkoh) == ("lukoil.png", None)
|
|
|
|
|
|
async def test_a_logo_already_set_is_kept_and_an_undescribed_paper_stays_bare(app):
|
|
kept = await make_instrument(ticker="KEPT")
|
|
await _set(kept, tinvest_uid="uid-kept", logo_name="mine.png", logo_color="#000000")
|
|
await _raw("uid-kept", isin=None, brand={"logo_name": "other.png"})
|
|
bare = await make_instrument(ticker="BARE")
|
|
await _set(bare, tinvest_uid="uid-bare")
|
|
await _raw("uid-bare", isin=None, brand={}) # the broker names no logo for it
|
|
|
|
assert await fill() == 0
|
|
assert await _logo(kept) == ("mine.png", "#000000")
|
|
assert await _logo(bare) == (None, None)
|