import datetime as dt import sqlite3 import pandas as pd import pytest from app.screener import snapshot as snap from app.screener.schema import ensure_screener_schema @pytest.fixture def conn(tmp_path): db_path = tmp_path / "snap.db" c = sqlite3.connect(db_path) ensure_screener_schema(c) yield c c.close() def _stub_listing(monkeypatch): df = pd.DataFrame([ {"Code": "005930", "Name": "삼성전자", "Market": "KOSPI", "Marcap": 420_000_000_000_000, "Open": 70000, "High": 72000, "Low": 69500, "Close": 71000, "Volume": 12_000_000, "Amount": 840_000_000_000}, {"Code": "035420", "Name": "NAVER", "Market": "KOSPI", "Marcap": 30_000_000_000_000, "Open": 215000, "High": 220000, "Low": 213000, "Close": 218000, "Volume": 1_000_000, "Amount": 218_000_000_000}, {"Code": "091990", "Name": "셀트리온헬스케어우", "Market": "KOSDAQ", "Marcap": 10_000_000_000_000, "Open": 60000, "High": 61000, "Low": 59500, "Close": 60500, "Volume": 500_000, "Amount": 30_250_000_000}, ]) monkeypatch.setattr(snap, "fetch_master_listing", lambda: df) def _stub_flow(monkeypatch, mapping): def fake_flow(ticker, *, client): if mapping is None: return None v = mapping.get(ticker) if v is None: return None return { "date": dt.date(2026, 5, 12).isoformat(), "foreign_net": v["foreign_net"], "institution_net": v["institution_net"], } monkeypatch.setattr(snap, "fetch_flow_naver", fake_flow) def test_refresh_daily_writes_master_and_prices(conn, monkeypatch): _stub_listing(monkeypatch) _stub_flow(monkeypatch, None) summary = snap.refresh_daily(conn, dt.date(2026, 5, 12), flow_top_n=10, rate_limit_sec=0) assert summary["master_count"] == 3 assert summary["prices_count"] == 3 assert summary["flow_count"] == 0 row = conn.execute( "SELECT close FROM krx_daily_prices WHERE ticker='005930' AND date='2026-05-12'" ).fetchone() assert row[0] == 71000 def test_refresh_daily_writes_flow_for_top_n(conn, monkeypatch): _stub_listing(monkeypatch) _stub_flow(monkeypatch, { "005930": {"foreign_net": 12_000_000_000, "institution_net": 4_000_000_000}, "035420": {"foreign_net": -3_000_000_000, "institution_net": 8_000_000_000}, }) summary = snap.refresh_daily(conn, dt.date(2026, 5, 12), flow_top_n=2, rate_limit_sec=0) assert summary["flow_count"] == 2 row = conn.execute( "SELECT foreign_net FROM krx_flow WHERE ticker='005930'" ).fetchone() assert row[0] == 12_000_000_000 def test_master_flags_preferred(conn, monkeypatch): _stub_listing(monkeypatch) _stub_flow(monkeypatch, None) snap.refresh_daily(conn, dt.date(2026, 5, 12), flow_top_n=0, rate_limit_sec=0) pref = conn.execute( "SELECT is_preferred FROM krx_master WHERE ticker='091990'" ).fetchone() assert pref[0] == 1 def test_refresh_daily_is_idempotent(conn, monkeypatch): _stub_listing(monkeypatch) _stub_flow(monkeypatch, None) snap.refresh_daily(conn, dt.date(2026, 5, 12), flow_top_n=0, rate_limit_sec=0) snap.refresh_daily(conn, dt.date(2026, 5, 12), flow_top_n=0, rate_limit_sec=0) cnt = conn.execute( "SELECT count(*) FROM krx_daily_prices WHERE date='2026-05-12'" ).fetchone()[0] assert cnt == 3 def test_fetch_flow_naver_parses_html(): """Real HTML structure parse with synthetic naver-like markup.""" html = """

날짜
2026.05.12	71,000	500	0.71%	12,000,000	4,000,000,000	12,000,000,000	1	53.0
2026.05.09	70,500	-200	-0.28%	10,000,000	2,000,000,000	5,000,000,000	1	52.8

""" class FakeResp: status_code = 200 text = html class FakeClient: def get(self, url, params): return FakeResp() out = snap.fetch_flow_naver("005930", client=FakeClient()) assert out == { "date": "2026-05-12", "foreign_net": 12_000_000_000, "institution_net": 4_000_000_000, }