import concurrent.futures, contextlib, http.client, json, sqlite3, tempfile, threading, time, unittest
from ledger import Ledger,Problem,canonical,quantity,formatted,timestamp
from app import make_server,DEMO_TOKENS


def event(event_id='e-1',quantity='12.500',occurred='2026-01-15T10:00:00Z',**extra):
    return {'source':'worker','event_id':event_id,'unit':'compute_minute','quantity':quantity,'occurred_at':occurred,**extra}

def correction(event_id='c-1',original='e-1',quantity='11.250',**extra):
    return {'source':'ops','event_id':event_id,'original_source':'worker','original_event_id':original,'replacement_quantity':quantity,'reason':'Meter reported the wrong duration',**extra}

class LedgerTests(unittest.TestCase):
    def setUp(self):self.temp=tempfile.TemporaryDirectory();self.path=self.temp.name+'/ledger.sqlite';self.l=Ledger(self.path,clock=lambda:'2026-02-10T12:00:00Z')
    def tearDown(self):self.temp.cleanup()
    def test_exact_precision_and_no_float(self):
        self.l.ingest('a',event(quantity='0.100'));self.l.ingest('a',event('e-2','0.200'))
        self.assertEqual(self.l.reconcile('a','2026-01')['rebuilt']['compute_minute']['quantity'],'0.300')
        for bad in [0.1,'-1','1e3','NaN','0.0001',' 1','1.']:
            with self.subTest(bad=bad),self.assertRaises(Problem):quantity(bad,'compute_minute')
        with self.assertRaises(Problem):quantity('1.0','api_call')
    def test_large_integer_totals_remain_exact(self):
        for n in range(10):self.l.ingest('a',event(str(n),'1000000000000.000'))
        self.assertEqual(self.l.reconcile('a','2026-01')['rebuilt']['compute_minute']['base_units'],'10000000000000000')
    def test_timestamp_timezone_sets_utc_period(self):
        self.l.ingest('a',event(occurred='2026-02-01T01:00:00+02:00'))
        self.assertEqual(self.l.state('a')['events'][0]['origin_period'],'2026-01')
        for bad in ['2026-01-01','2026-01-01T00:00:00','2026-01-01T00:00:00.1Z']:
            with self.assertRaises(Problem):timestamp(bad)
    def test_canonical_idempotency_and_conflicting_duplicate(self):
        first=self.l.ingest('a',event());second=self.l.ingest('a',event(quantity='12.5',occurred='2026-01-15T12:00:00+02:00'))
        self.assertEqual(first['seq'],second['seq']);self.assertTrue(second['duplicate'])
        with self.assertRaises(Problem) as cm:self.l.ingest('a',event(quantity='13'))
        self.assertEqual(cm.exception.status,409);self.assertEqual(len(self.l.state('a')['events']),1)
    def test_same_identity_separate_accounts(self):
        self.l.ingest('a',event());self.l.ingest('b',event(quantity='9'));self.assertEqual(self.l.state('b')['totals'][0]['quantity'],'9.000')
    def test_reordered_import_duplicates_reconcile(self):
        data=[event(str(n),str(n/10)) for n in range(20)]
        for x in data:self.l.ingest('a',x)
        for x in reversed(data):self.assertTrue(self.l.ingest('a',x)['duplicate'])
        self.assertEqual(len(self.l.state('a')['events']),20);self.assertTrue(self.l.reconcile('a','2026-01')['matches'])
    def test_open_correction_is_linked_reversal_and_replacement(self):
        original=self.l.ingest('a',event());result=self.l.correct('a',correction())
        self.assertEqual(result['original_seq'],original['seq']);self.assertFalse(result['pending'])
        self.assertEqual(self.l.reconcile('a','2026-01')['rebuilt']['compute_minute']['quantity'],'11.250')
        with self.l.connect() as db:self.assertEqual([r[0] for r in db.execute('SELECT amount FROM entries WHERE event_seq=? ORDER BY id',(result['seq'],))],['-12500','11250'])
    def test_correction_replay_and_conflict(self):
        self.l.ingest('a',event());self.l.correct('a',correction());self.assertTrue(self.l.correct('a',correction())['duplicate'])
        with self.assertRaises(Problem):self.l.correct('a',correction(quantity='2'))
    def test_correction_chain_requires_latest_revision(self):
        self.l.ingest('a',event());self.l.correct('a',correction())
        with self.assertRaises(Problem):self.l.correct('a',correction('c-2'))
        self.l.correct('a',correction('c-2','c-1','10.000',original_source='ops'))
        self.assertEqual(self.l.reconcile('a','2026-01')['rebuilt']['compute_minute']['quantity'],'10.000')
    def test_correction_cannot_link_other_account(self):
        self.l.ingest('a',event())
        with self.assertRaises(Problem) as cm:self.l.correct('b',correction())
        self.assertEqual(cm.exception.status,404)
    def test_closed_export_frozen_after_late_event_and_correction(self):
        self.l.ingest('a',event());self.l.close('a','2026-01');before=canonical(self.l.export('a','2026-01'))
        late=self.l.ingest('a',event('late','2'));change=self.l.correct('a',correction())
        self.assertTrue(late['pending']);self.assertTrue(change['pending'])
        for x in [late,change]:self.l.approve('a',x['seq'],'2026-02')
        self.l.close('a','2026-02');self.assertEqual(before,canonical(self.l.export('a','2026-01')))
        self.assertEqual(self.l.export('a','2026-02')['totals']['compute_minute']['quantity'],'0.750')
    def test_posting_cutoff_excludes_previously_ingested_unposted_events(self):
        self.l.close('a','2026-01');late=self.l.ingest('a',event());self.l.close('a','2026-02');frozen=self.l.export('a','2026-02')
        self.l.approve('a',late['seq'],'2026-03');self.assertEqual(frozen,self.l.export('a','2026-02'))
    def test_adjustment_target_and_duplicate_posting_rejected(self):
        self.l.close('a','2026-01');late=self.l.ingest('a',event())
        with self.assertRaises(Problem):self.l.approve('a',late['seq'],'2026-01')
        self.l.close('a','2026-02')
        with self.assertRaises(Problem):self.l.approve('a',late['seq'],'2026-02')
        self.l.approve('a',late['seq'],'2026-03')
        with self.assertRaises(Problem):self.l.approve('a',late['seq'],'2026-04')
    def test_close_idempotency_and_restart_export(self):
        self.l.ingest('a',event());self.l.close('a','2026-01');self.assertTrue(self.l.close('a','2026-01')['duplicate'])
        reopened=Ledger(self.path);self.assertEqual(self.l.export('a','2026-01'),reopened.export('a','2026-01'))
    def test_immutable_storage_guards(self):
        self.l.ingest('a',event());self.l.close('a','2026-01')
        with self.l.connect() as db:
            for table in ['events','entries','posting_batches','postings','closes','audit','periods']:
                with self.subTest(table=table),self.assertRaises(sqlite3.IntegrityError):db.execute(f'DELETE FROM {table}')
            with self.assertRaises(sqlite3.IntegrityError):db.execute("UPDATE periods SET status='open'")
    def test_corrupt_cache_detected_before_close(self):
        self.l.ingest('a',event())
        with self.l.connect() as db:db.execute("UPDATE totals SET amount='1'")
        self.assertFalse(self.l.reconcile('a','2026-01')['matches'])
        with self.assertRaises(Problem):self.l.close('a','2026-01')
    def test_concurrent_duplicate_ingest_once(self):
        with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:results=list(pool.map(lambda _:self.l.ingest('a',event()),range(16)))
        self.assertEqual(sum(not r['duplicate'] for r in results),1);self.assertEqual(len(self.l.state('a')['events']),1)
    def race(self,first):
        acquired=threading.Event();release=threading.Event();waiting=threading.Event()
        def observe(operation):
            if operation==first:acquired.set();self.assertTrue(release.wait(4))
        blocker=Ledger(self.path,transaction_observer=observe)
        def ingest():return (blocker if first=='ingest' else self.l).ingest('a',event())
        def close():return (blocker if first=='close' else self.l).close('a','2026-01')
        primary,secondary=(ingest,close) if first=='ingest' else (close,ingest)
        with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
            one=pool.submit(primary);self.assertTrue(acquired.wait(4))
            def second():waiting.set();return secondary()
            two=pool.submit(second);self.assertTrue(waiting.wait(4));time.sleep(.03);self.assertFalse(two.done());release.set();one.result();two.result()
        return self.l.export('a','2026-01')
    def test_ingest_wins_close_race_included(self):self.assertEqual(self.race('ingest')['totals']['compute_minute']['quantity'],'12.500')
    def test_close_wins_ingest_race_pending(self):self.assertEqual(self.race('close')['totals'],{});self.assertEqual(len(self.l.state('a')['pending']),1)
    def test_empty_period_close(self):self.l.close('a','2026-01');self.assertTrue(self.l.export('a','2026-01')['verified'])

class ApiTests(unittest.TestCase):
    def setUp(self):
        self.temp=tempfile.TemporaryDirectory();self.l=Ledger(self.temp.name+'/ledger.sqlite');self.server=make_server(self.l,DEMO_TOKENS,0);self.thread=threading.Thread(target=self.server.serve_forever,daemon=True);self.thread.start()
    def tearDown(self):self.server.shutdown();self.server.server_close();self.thread.join();self.temp.cleanup()
    def request(self,method,path,body=None,token='demo-operator-token'):
        conn=http.client.HTTPConnection('127.0.0.1',self.server.server_port,timeout=3);headers={'Authorization':'Bearer '+token}
        raw=json.dumps(body).encode() if body is not None else None
        if raw:headers['Content-Type']='application/json'
        conn.request(method,path,raw,headers);r=conn.getresponse();out=(r.status,r.read(),r.getheaders());conn.close();return out
    def test_auth_and_cross_account(self):
        self.assertEqual(self.request('GET','/api/accounts/demo/state',token='bad')[0],401)
        self.assertEqual(self.request('GET','/api/accounts/other/state')[0],403)
        self.assertEqual(self.request('POST','/api/accounts/other/events',event())[0],403)
    def test_roles(self):
        self.assertEqual(self.request('POST','/api/accounts/demo/events',event(),token='demo-viewer-token')[0],403)
        self.assertEqual(self.request('POST','/api/accounts/demo/events',event(),token='demo-ingest-token')[0],201)
        self.assertEqual(self.request('POST','/api/accounts/demo/close',{'period':'2026-01'},token='demo-ingest-token')[0],403)
        self.assertEqual(self.request('GET','/api/accounts/demo/state',token='demo-viewer-token')[0],200)
    def test_export_and_csv_formula_safety(self):
        self.l.ingest('demo',event());self.l.correct('demo',correction(reason='=HYPERLINK("https://invalid.example")'));self.l.close('demo','2026-01')
        status,raw,_=self.request('GET','/api/accounts/demo/export?period=2026-01');self.assertEqual(status,200);self.assertTrue(json.loads(raw)['verified'])
        status,raw,_=self.request('GET','/api/accounts/demo/csv?period=2026-01');self.assertEqual(status,200);self.assertIn(b"'=HYPERLINK",raw)
    def test_invalid_and_static_paths(self):
        self.assertEqual(self.request('POST','/api/accounts/demo/events',[])[0],400)
        self.assertEqual(self.request('GET','/../ledger.py')[0],404)
        status,raw,headers=self.request('GET','/');self.assertEqual(status,200);self.assertIn(b'Usage reconciliation',raw);self.assertTrue(any(k.lower()=='content-security-policy' for k,v in headers))

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