| 1 | import csv |
| 2 | import io |
| 3 | import logging |
| 4 | from datetime import date |
| 5 | from decimal import Decimal |
| 6 | from pathlib import Path |
| 7 | from types import SimpleNamespace |
| 8 | from unittest.mock import AsyncMock |
| 9 | from uuid import UUID |
| 10 | |
| 11 | import httpx |
| 12 | import pytest |
| 13 | from app.nse_historical_daily import BOOTSTRAP, NseHistoricalDailyProvider, NseHistoricalResult, parse_csv, persist_daily_result, verified_identity |
| 14 | from app.persistence import SqliteResearchPersistence |
| 15 | from app.settings import Settings |
| 16 | |
| 17 | KEY=UUID(int=1) |
| 18 | START, END=date(2026,9,1),date(2026,9,4) |
| 19 | FIXTURE=Path(__file__).parent/'fixtures/nse_historical_daily.csv' |
| 20 | META=dict(globalInstrumentId=str(KEY),status='ACTIVE',assetType='EQUITY',country='IN',primaryExchange='NSE',currency='INR',providerMappings=[dict(provider='NSE',providerSymbol='POLYCAB',status='VERIFIED',resolutionSource='OFFICIAL_NSE_NIFTY500',currency='INR')]) |
| 21 | |
| 22 | def result(): return NseHistoricalResult(KEY,START,END,provider_symbol='POLYCAB') |
| 23 | def modified(**changes): |
| 24 | rows=list(csv.reader(io.StringIO(FIXTURE.read_text(encoding='utf-8-sig')))) |
| 25 | headers=[h.strip() for h in rows[0]] |
| 26 | for key,value in changes.items(): rows[1][headers.index(key)]=value |
| 27 | out=io.StringIO(newline=''); csv.writer(out).writerows(rows) |
| 28 | return out.getvalue().encode() |
| 29 | |
| 30 | def test_real_contract(): |
| 31 | r=result(); parse_csv(FIXTURE.read_bytes(),r,'INR') |
| 32 | assert r.rows_parsed==r.rows_accepted==4 and r.rows_rejected==0 |
| 33 | assert r.first_trading_date==START and r.last_trading_date==END |
| 34 | assert r.bars[-1].turnover==Decimal('13025481530.00') |
| 35 | assert r.bars[-1].volume==1559741 and type(r.bars[-1].volume) is int |
| 36 | assert r.bars[0].close==Decimal('8895.50') |
| 37 | assert r.bars[0].retrieved_at.utcoffset().total_seconds()==0 |
| 38 | |
| 39 | def test_optional_aliases_order_zero_precision(): |
| 40 | r=result(); parse_csv(b'Close,Date,Symbol,Series,Open,High,Low,Turnover (in Lacs)\n2,01-09-2026, polycab , eq ,1,2,1,100\n\n',r,'INR') |
| 41 | assert r.bars[0].turnover is r.bars[0].volume is r.bars[0].previous_close is None |
| 42 | r=result(); parse_csv(modified(**{'Total Traded Quantity':'0','Turnover ₹':'0','Prev Close':'-','Open Price':'8500.123456789012'}),r,'INR') |
| 43 | assert r.bars[-1].turnover==r.bars[-1].volume==0 and r.bars[-1].previous_close is None |
| 44 | assert r.bars[-1].open==Decimal('8500.123456789012') |
| 45 | |
| 46 | @pytest.mark.parametrize('field,value',[('Symbol','OTHER'),('Series','BE'),('Date','31-Sep-2026'),('Date','31-Aug-2026'),('Open Price','NaN'),('High Price','1'),('Low Price','0'),('Close Price','-1'),('Prev Close','0'),('Total Traded Quantity','9223372036854775808'),('Total Traded Quantity','1.5'),('Total Traded Quantity','-1'),('Turnover ₹','-1'),('Open Price','1,2'),('Close Price','inf')]) |
| 47 | def test_row_validation(field,value): |
| 48 | r=result(); parse_csv(modified(**{field:value}),r,'INR') |
| 49 | assert r.rows_parsed==4 and r.rows_accepted==3 and r.rows_rejected==1 |
| 50 | |
| 51 | def test_bigint_duplicates(): |
| 52 | r=result(); parse_csv(modified(**{'Total Traded Quantity':'9223372036854775807'}),r,'INR') |
| 53 | assert r.bars[-1].volume==9223372036854775807 |
| 54 | r=result(); parse_csv(modified(Date='03-Sep-2026'),r,'INR') |
| 55 | assert r.rows_accepted==2 and r.rejection_reasons=={'DUPLICATE_DATE':2} |
| 56 | |
| 57 | @pytest.mark.parametrize('body',[b'',b'<html>x</html>',b'{}',b'Date,Close\n1,2',b'Date,Symbol,Series,Open,High,Low,Close\n"unclosed']) |
| 58 | def test_bad_contract(body): |
| 59 | with pytest.raises((ValueError,csv.Error)): parse_csv(body,result(),'INR') |
| 60 | |
| 61 | @pytest.mark.parametrize('change',[dict(providerMappings=[]),dict(status='INACTIVE'),dict(assetType='ETF'),dict(globalInstrumentId=str(UUID(int=2))),dict(providerMappings=META['providerMappings']*2),dict(currency=None,providerMappings=[dict(provider='NSE',providerSymbol='X',status='VERIFIED')]),*[dict(providerMappings=[META['providerMappings'][0]|u]) for u in [dict(status='UNVERIFIED'),dict(active=False),dict(providerSymbol=' '),dict(resolutionSource='BROKER_IMPORT_IDENTITY'),dict(currency='USD')]]]) |
| 62 | def test_identity(change): |
| 63 | with pytest.raises(ValueError): verified_identity(META|change,KEY) |
| 64 | |
| 65 | async def fetch_with(handler,metadata=META,**settings): |
| 66 | async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: |
| 67 | sleep=AsyncMock() |
| 68 | provider=NseHistoricalDailyProvider(SimpleNamespace(global_instrument_metadata=AsyncMock(return_value=metadata)),Settings(**settings),client=client,sleep=sleep) |
| 69 | return await provider.fetch(KEY,start=START,end=END),sleep |
| 70 | |
| 71 | @pytest.mark.asyncio |
| 72 | async def test_session_query_no_cookie_logs(caplog): |
| 73 | calls=[] |
| 74 | def handler(req): |
| 75 | calls.append(req) |
| 76 | assert req.headers['user-agent'].startswith('Mozilla/') and req.headers['referer']==BOOTSTRAP |
| 77 | assert 'accept-language' in req.headers |
| 78 | if len(calls)==1: return httpx.Response(200,headers={'set-cookie':'session=secret-cookie; Path=/; Secure'}) |
| 79 | assert req.headers['cookie']=='session=secret-cookie' |
| 80 | assert dict(req.url.params)==dict(symbol='POLYCAB',series='EQ',csv='true',type='priceVolumeDeliverable',**{'from':'01-09-2026','to':'04-09-2026'}) |
| 81 | return httpx.Response(200,content=FIXTURE.read_bytes(),headers={'content-type':'text/csv'}) |
| 82 | with caplog.at_level(logging.INFO): r,sleep=await fetch_with(handler) |
| 83 | assert r.status=='SUCCESS' and len(calls)==2 and sleep.await_count>=1 |
| 84 | assert 'secret-cookie' not in caplog.text |
| 85 | |
| 86 | @pytest.mark.asyncio |
| 87 | @pytest.mark.parametrize('status,count',[(403,1),(404,1),(429,3),(500,3)]) |
| 88 | async def test_http_failures(status,count): |
| 89 | calls=[] |
| 90 | def handler(req): |
| 91 | if str(req.url)==BOOTSTRAP: return httpx.Response(200) |
| 92 | calls.append(req); return httpx.Response(status) |
| 93 | r,_=await fetch_with(handler) |
| 94 | assert r.failure_reason==f'HISTORICAL_HTTP_{status}' and len(calls)==count and not r.bars |
| 95 | |
| 96 | @pytest.mark.asyncio |
| 97 | @pytest.mark.parametrize('body,ctype',[(b'','text/csv'),(b'<html>x</html>','text/html'),(b'wrong,headers','text/csv'),(b'Date,Symbol,Series,Open,High,Low,Close\n"oops','text/csv')]) |
| 98 | async def test_provider_bad_response(body,ctype): |
| 99 | r,_=await fetch_with(lambda req: httpx.Response(200) if str(req.url)==BOOTSTRAP else httpx.Response(200,content=body,headers={'content-type':ctype})) |
| 100 | assert r.status=='UNAVAILABLE' and r.failure_reason and not r.bars |
| 101 | |
| 102 | @pytest.mark.asyncio |
| 103 | async def test_bootstrap_and_missing_identity(): |
| 104 | r,_=await fetch_with(lambda req:httpx.Response(403)) |
| 105 | assert r.failure_reason=='BOOTSTRAP_HTTP_403' |
| 106 | handler=AsyncMock() |
| 107 | r,_=await fetch_with(handler,metadata=META|dict(providerMappings=[])) |
| 108 | assert r.failure_reason=='NO_UNAMBIGUOUS_NSE_MAPPING'; handler.assert_not_called() |
| 109 | |
| 110 | @pytest.mark.asyncio |
| 111 | @pytest.mark.parametrize('error',[httpx.ReadTimeout,httpx.ConnectError]) |
| 112 | async def test_transport(error): |
| 113 | calls=[] |
| 114 | def handler(req): calls.append(req); raise error('unavailable') |
| 115 | r,_=await fetch_with(handler) |
| 116 | assert len(calls)==3 and r.failure_reason in {'BOOTSTRAP_TIMEOUT','BOOTSTRAP_CONNECTION_FAILURE'} |
| 117 | |
| 118 | @pytest.mark.asyncio |
| 119 | async def test_persistence(): |
| 120 | store=SqliteResearchPersistence() |
| 121 | repo=SimpleNamespace(upsert_daily_market_bars_async=AsyncMock(side_effect=store.upsert_daily_market_bars)) |
| 122 | r=result(); parse_csv(FIXTURE.read_bytes(),r,'INR'); r.status='SUCCESS' |
| 123 | other=r.bars[0].model_copy(update={'provider':'OTHER'}); store.upsert_daily_market_bar(other) |
| 124 | await persist_daily_result(repo,r); await persist_daily_result(repo,r) |
| 125 | r.bars[0]=r.bars[0].model_copy(update={'close':Decimal('8900')}); await persist_daily_result(repo,r) |
| 126 | loaded=store.load_daily_market_bars({KEY}) |
| 127 | assert len(loaded)==5 and other in loaded and r.persisted_rows==4 |
| 128 | assert store.load_daily_market_bars({KEY},provider='NSE')[0].close==Decimal('8900') |
| 129 | assert store.load_market_price_observations({KEY})==[] |
| 130 | await persist_daily_result(repo,result()) |
| 131 | assert store.load_daily_market_bars({KEY})==loaded |
| 132 | |
| 133 | @pytest.mark.asyncio |
| 134 | async def test_window_rejected_before_identity_or_network(): |
| 135 | resolver=SimpleNamespace(global_instrument_metadata=AsyncMock()) |
| 136 | async with httpx.AsyncClient(transport=httpx.MockTransport(lambda req: pytest.fail('network'))) as client: |
| 137 | p=NseHistoricalDailyProvider(resolver,Settings(nse_historical_request_window_days=2),client=client) |
| 138 | r=await p.fetch(KEY,start=START,end=END) |
| 139 | assert r.failure_reason=='REQUEST_WINDOW_EXCEEDED' |
| 140 | r=await p.fetch(KEY,start=END,end=START) |
| 141 | assert r.failure_reason=='INVALID_DATE_RANGE' |
| 142 | resolver.global_instrument_metadata.assert_not_called() |
| 143 | |
| 144 | @pytest.mark.asyncio |
| 145 | async def test_throttle_recovery_and_long_retry_after(): |
| 146 | calls=[] |
| 147 | def handler(req): |
| 148 | if str(req.url)==BOOTSTRAP: return httpx.Response(200) |
| 149 | calls.append(req) |
| 150 | return httpx.Response(429,headers={'retry-after':'5'}) if len(calls)==1 else httpx.Response(200,content=FIXTURE.read_bytes()) |
| 151 | r,sleep=await fetch_with(handler) |
| 152 | assert r.status=='SUCCESS' and len(calls)==2 |
| 153 | assert any(c.args==(5,) for c in sleep.await_args_list) |
| 154 | calls.clear() |
| 155 | def throttled(req): |
| 156 | if str(req.url)==BOOTSTRAP: return httpx.Response(200) |
| 157 | calls.append(req); return httpx.Response(429,headers={'retry-after':'120'}) |
| 158 | r,_=await fetch_with(throttled) |
| 159 | assert r.failure_reason=='HISTORICAL_HTTP_429' and len(calls)==1 |
| 160 | |
| 161 | @pytest.mark.asyncio |
| 162 | async def test_session_reused_serially(): |
| 163 | active=0 |
| 164 | calls=[] |
| 165 | async def handler(req): |
| 166 | nonlocal active |
| 167 | import asyncio |
| 168 | active+=1; assert active==1 |
| 169 | await asyncio.sleep(0) |
| 170 | active-=1; calls.append(req) |
| 171 | return httpx.Response(200,content=FIXTURE.read_bytes()) |
| 172 | import asyncio |
| 173 | async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: |
| 174 | p=NseHistoricalDailyProvider(SimpleNamespace(global_instrument_metadata=AsyncMock(return_value=META)),Settings(),client=client,sleep=AsyncMock()) |
| 175 | results=await asyncio.gather(p.fetch(KEY,start=START,end=END),p.fetch(KEY,start=START,end=END)) |
| 176 | assert all(r.status=='SUCCESS' for r in results) |
| 177 | assert sum(str(r.url)==BOOTSTRAP for r in calls)==1 |
| 178 | |
| 179 | @pytest.mark.asyncio |
| 180 | async def test_explicit_population_boundary(monkeypatch): |
| 181 | from app.market_data_population import IndiaMarketDataPopulationJobs |
| 182 | from app.repository import ResearchRepository |
| 183 | import app.nse_historical_daily as module |
| 184 | store=SqliteResearchPersistence() |
| 185 | repo=ResearchRepository(settings=Settings(),persistence=store) |
| 186 | resolver=SimpleNamespace(global_instrument_metadata=AsyncMock(return_value=META)) |
| 187 | async with httpx.AsyncClient(transport=httpx.MockTransport(lambda req:httpx.Response(200,content=FIXTURE.read_bytes()))) as client: |
| 188 | provider=NseHistoricalDailyProvider(resolver,Settings(),client=client,sleep=AsyncMock()) |
| 189 | monkeypatch.setattr(module,'NseHistoricalDailyProvider',lambda *args:provider) |
| 190 | jobs=IndiaMarketDataPopulationJobs(repo,None,resolver,Settings(),sleep=AsyncMock()) |
| 191 | r=await jobs.populate_daily_bars(KEY,start=START,end=END,identity_headers={}) |
| 192 | assert r.persisted_rows==4 |
| 193 | assert len((await repo.daily_market_bars_for_instruments({KEY},provider='NSE'))[KEY])==4 |
| 194 | assert store.load_market_price_observations({KEY})==[] |
| 195 | assert jobs.historical_provider.provider_name=='YAHOO_FINANCE' |
| 196 | |
| 197 | @pytest.mark.asyncio |
| 198 | async def test_persistence_failure_explicit(): |
| 199 | r=result(); parse_csv(FIXTURE.read_bytes(),r,'INR'); r.status='SUCCESS' |
| 200 | await persist_daily_result(SimpleNamespace(upsert_daily_market_bars_async=AsyncMock(side_effect=RuntimeError())),r) |
| 201 | assert r.failure_reason=='DAILY_BAR_PERSISTENCE_UNAVAILABLE' and r.persisted_rows==0 |
| 202 | |
| 203 | def test_malformed_row_shape_and_duplicate_header(): |
| 204 | r=result(); parse_csv(b'Date,Symbol,Series,Open,High,Low,Close\n01-09-2026,POLYCAB,EQ,1,2,1,2,extra\n',r,'INR') |
| 205 | assert r.rows_rejected==1 and r.rejection_reasons=={'MALFORMED_ROW':1} |
| 206 | with pytest.raises(ValueError): parse_csv(b'Date,Symbol,Series,Open,Open Price,High,Low,Close\n',result(),'INR') |
| 207 | |
| 208 | def test_observed_rupee_turnover_matches_quantity_times_average(): |
| 209 | # NSE average is rounded to two decimals; turnover is explicitly rupees. |
| 210 | for row in csv.DictReader(io.StringIO(FIXTURE.read_text(encoding='utf-8-sig'))): |
| 211 | row={k.strip():v for k,v in row.items()} |
| 212 | qty=Decimal(row['Total Traded Quantity'].replace(',','')) |
| 213 | average=Decimal(row['Average Price'].replace(',','')) |
| 214 | turnover=Decimal(row['Turnover ₹'].replace(',','')) |
| 215 | assert abs(turnover-qty*average)<=qty*Decimal('0.005') |
| 216 | |
| 217 | @pytest.mark.asyncio |
| 218 | async def test_generic_verified_symbol_not_primary_symbol(): |
| 219 | metadata=META|dict(primarySymbol='DO_NOT_USE',providerMappings=[META['providerMappings'][0]|dict(providerSymbol='ANOTHER')]) |
| 220 | def handler(req): |
| 221 | if str(req.url)==BOOTSTRAP: return httpx.Response(200) |
| 222 | assert req.url.params['symbol']=='ANOTHER' |
| 223 | return httpx.Response(200,content=FIXTURE.read_bytes().replace(b'POLYCAB',b'ANOTHER')) |
| 224 | r,_=await fetch_with(handler,metadata=metadata) |
| 225 | assert r.rows_accepted==4 and all(b.provider_symbol=='ANOTHER' and b.global_instrument_id==KEY for b in r.bars) |