import contextlib
from datetime import datetime,timezone
import http.client
import io
import json
from pathlib import Path
import tempfile
import threading
from types import SimpleNamespace
import unittest
from gauntlet import Store,make_server,Blocked
from activity_privacy import clean
import host_replay_adapter as adapter

class ActivityTests(unittest.TestCase):
 def setUp(self):
  self.tmp=tempfile.TemporaryDirectory();self.p=Path(self.tmp.name);self.store=Store(self.p/'events.db')
  self.silence=contextlib.redirect_stdout(io.StringIO());self.silence.__enter__()
 def tearDown(self):self.silence.__exit__(None,None,None);self.tmp.cleanup()
 def test_cursor_preserves_all_history(self):
  for i in range(620):self.store.event('MARK',index=i)
  cursor=0;ids=[]
  while True:
   page=self.store.activity(after=cursor,limit=200);ids += [e['id'] for e in page['events']]
   if not page['has_more']:break
   cursor=page['next_after']
  self.assertEqual(ids,list(range(1,621)))
  self.assertEqual(len(self.store.snapshot()['events']),200)
 def test_search_and_type(self):
  self.store.event('RUN_BLOCKED',code='MISSING_BARS');self.store.event('OTHER',code='MISSING_BARS')
  p=self.store.activity(q='missing_bars',kind='RUN_BLOCKED')
  self.assertEqual(len(p['events']),1);self.assertEqual(p['events'][0]['code'],'MISSING_BARS')
 def test_redaction(self):
  self.store.event('TEST',password='private',nested={'token':'private'},source='https://user:pw@example.test/path?api_key=private')
  text=json.dumps(self.store.activity());self.assertNotIn('private',text);self.assertNotIn('user:pw',text)
 def test_archived_import_once(self):
  from import_history import import_file
  file=self.p/'attempt_1_year_1950.json';file.write_text(json.dumps({'events':[{'type':'MARK','at':'1950-01-01','price':'1'}]}))
  self.assertEqual(import_file(self.store,file),1);self.assertEqual(import_file(self.store,file),0)
  self.assertEqual(self.store.activity()['events'][0]['origin'],'IMPORTED_ARCHIVED_LEDGER')
 def test_legacy_credentials_redacted_on_read(self):
  with self.store.connect() as db:db.execute('INSERT INTO events(created_at,kind,payload) VALUES(?,?,?)',('fixture','OLD',json.dumps({'password':'NEVER_SHOW'})))
  self.assertNotIn('NEVER_SHOW',json.dumps(self.store.activity()))
 def test_invalid_cursor(self):
  with self.assertRaises(ValueError):self.store.activity(after=-1)
 def test_http_activity_and_export(self):
  self.store.event('RUN_BLOCKED',code='fixture')
  server=make_server(self.store,0);thread=threading.Thread(target=server.serve_forever);thread.start()
  try:
   for path in ['/api/activity?q=fixture','/api/export']:
    conn=http.client.HTTPConnection(*server.server_address);conn.request('GET',path);r=conn.getresponse();self.assertEqual(r.status,200);self.assertIn('fixture',r.read().decode());conn.close()
  finally:server.shutdown();server.server_close();thread.join()
 def test_activity_visible_before_year_finishes(self):
  bars=self.p/'bars.jsonl';macro=self.p/'macro.jsonl';macro.write_text('')
  bars.write_text(json.dumps({'at':'2015-01-01T00:00:00Z','available_at':'2015-01-01T00:00:00Z','data_origin':'HISTORICAL_INTRADAY_QUOTES','bar_interval_seconds':60,'quotes':{'BTC/USD':'300'},'source_url':'https://example.test/FIXTURE_ONLY'})+'\n')
  def decide(ctx):
   self.assertEqual(self.store.activity(kind='REPLAY_EVALUATION_STARTED')['events'][0]['activity']['at'],'2015-01-01T00:00:00+00:00')
   return [{'action':'HOLD','reason':'Fixture only'}]
  strategy=SimpleNamespace(__file__=__file__,decide=decide,capital_floor=lambda _: '0')
  with self.assertRaisesRegex(Blocked,'YEAR_END_COVERAGE_MISSING'):
   adapter.replay({'calendar_year':2015,'attempt':1,'_activity_db':str(self.p/'events.db')},strategy,bars,macro,self.p/'out')
  self.assertTrue(self.store.activity(kind='REPLAY_DECISION')['events'])

if __name__=='__main__':unittest.main()