import ast,json,threading,time,unittest,urllib.request
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer
from urllib.parse import urlparse,parse_qs
from datetime import datetime,timezone
source=Path(__file__).with_name('v2_stats_dashboard.py').read_text(encoding='utf-8')
tree=ast.parse(source)
node=next(n for n in tree.body if isinstance(n,ast.ClassDef) and n.name=='StatsHandler')
class Tests(unittest.TestCase):
 def test_health_during_blocked_report(self):
  started=threading.Event();release=threading.Event();calls=[]
  def assemble():started.set();release.wait(5);calls.append(1);return {'fixture':True}
  ns=dict(globals(),DATA_REQUEST_LOCK=threading.RLock(),assemble_data=assemble,log=lambda *a:None)
  exec(compile(ast.Module(body=[node],type_ignores=[]),'actual_handler','exec'),ns)
  handler=ns['StatsHandler'];handler._check_auth=lambda self:True
  server=ThreadingHTTPServer(('127.0.0.1',0),handler)
  t=threading.Thread(target=server.serve_forever,daemon=True);t.start()
  base='http://127.0.0.1:'+str(server.server_port)
  try:
   with ThreadPoolExecutor() as pool:
    future=pool.submit(lambda:json.load(urllib.request.urlopen(base+'/json',timeout=8)))
    self.assertTrue(started.wait(2))
    try:
     for _ in range(5):
      begin=time.monotonic();r=json.load(urllib.request.urlopen(base+'/healthz',timeout=1))
      self.assertEqual(r['status'],'ok');self.assertLess(time.monotonic()-begin,1)
     self.assertEqual(calls,[])
    finally:release.set()
    self.assertEqual(future.result(),{'fixture':True})
   handler._check_auth=lambda self:False
   handler._send_unauthorized=lambda self:(self.send_response(401),self.end_headers())
   with self.assertRaises(urllib.error.HTTPError) as err:urllib.request.urlopen(base+'/healthz')
   self.assertEqual(err.exception.code,401)
  finally:release.set();server.shutdown();server.server_close()
 def test_read_write_serialization(self):
  ns=dict(globals(),DATA_REQUEST_LOCK=threading.RLock())
  exec(compile(ast.Module(body=[node],type_ignores=[]),'actual_handler','exec'),ns)
  cls=ns['StatsHandler'];active=[];overlap=[]
  def data(self):
   active.append(1);overlap.append(len(active));time.sleep(.04);active.pop()
  cls._do_GET_data=data;cls._do_POST_data=data;cls._check_auth=lambda self:True
  h=object.__new__(cls);h.path='/json'
  with ThreadPoolExecutor() as pool:list(pool.map(lambda i:h.do_GET() if i%2 else h.do_POST(),range(6)))
  self.assertEqual(max(overlap),1)
if __name__=='__main__':unittest.main()