import hashlib
import hmac
import json
import os
import sys
import tempfile
import threading
import unittest
import uuid
import urllib.error
import urllib.request
from datetime import datetime
from pathlib import Path
from unittest.mock import patch

sys.path.insert(0,str(Path(__file__).resolve().parents[1]))
import db
from engine import create_order, extract, process, receive
from scheduling import IST, next_delivery
from seed import seed
from server import Handler, ThreadingHTTPServer
from worker import run_once

class DatabaseCase(unittest.TestCase):
    def setUp(self):
        from types import SimpleNamespace
        folder=db.ROOT/'tests'/('run-'+uuid.uuid4().hex)
        folder.mkdir()
        self.tmp=SimpleNamespace(name=str(folder))
        self.previous=db.DB_PATH
        db.DB_PATH=Path(self.tmp.name)/'test.db'
        self.env=patch.dict(os.environ,{'ADMIN_PASSWORD':'test-password-12345','OLLAMA_MODEL':''})
        self.env.start()
        db.init()
        with db.connect() as c: seed(c)
    def tearDown(self):
        db.DB_PATH=self.previous
        self.env.stop()
        import shutil
        target=Path(self.tmp.name).resolve()
        self.assertEqual(target.parent,(db.ROOT/'tests').resolve())
        shutil.rmtree(target)
    def message(self,body,phone='919000000000',eid=None):
        with db.connect() as c: mid=receive(c,1,phone,body,eid)
        with db.connect() as c: process(c,mid)
        return mid

class WorkflowTests(DatabaseCase):
    def test_order_requires_confirmation_and_saved_area(self):
        self.message('Sir 5 box floor cleaner aur 10 handwash bhej dena.')
        with db.connect() as c:
            self.assertEqual(c.execute('SELECT COUNT(*) FROM orders').fetchone()[0],0)
            conv=db.one(c,'SELECT * FROM conversations')
            self.assertEqual(json.loads(conv['pending_json'])['stage'],'confirm')
        self.message('haan')
        with db.connect() as c:
            o=db.one(c,'SELECT * FROM orders')
            self.assertEqual(o['customer_id'],1)
            self.assertEqual(c.execute('SELECT COUNT(*) FROM order_items').fetchone()[0],2)
            self.assertEqual(o['route_id'],1)
            self.assertEqual(c.execute('SELECT COUNT(*) FROM outbox').fetchone()[0],2)
        run_once()
        with db.connect() as c: self.assertEqual(db.one(c,'SELECT state FROM outbox')['state'],'simulated')
    def test_new_customer_missing_area_followup(self):
        self.message('2 carton tissue chahiye','919111111111')
        with db.connect() as c:
            self.assertEqual(json.loads(db.one(c,'SELECT pending_json FROM conversations')['pending_json'])['stage'],'area')
        self.message('Demo Market B','919111111111')
        self.message('Shop 12, main market near bus stand','919111111111')
        self.message('YES','919111111111')
        with db.connect() as c:
            self.assertEqual(db.one(c,'SELECT * FROM orders')['route_id'],2)
    def test_duplicate_webhook_and_order(self):
        mid=self.message('5 box cleaner bhej do',eid='wamid.unique')
        with db.connect() as c: self.assertEqual(receive(c,1,'919000000000','different text','wamid.unique'),mid)
        self.message('yes')
        self.message('5 box cleaner bhej do')
        confirm=self.message('yes')
        with db.connect() as c:
            self.assertEqual(c.execute('SELECT COUNT(*) FROM orders').fetchone()[0],1)
            self.assertEqual(db.one(c,'SELECT state FROM messages WHERE id=?',(confirm,))['state'],'review')
            self.assertIn('Duplicate',db.one(c,'SELECT body FROM notifications')['body'])
    def test_repeat_confirmation(self):
        self.message('5 box cleaner bhej do');self.message('yes')
        self.message('last wala repeat kar do')
        with db.connect() as c:
            self.assertEqual(c.execute('SELECT COUNT(*) FROM orders').fetchone()[0],1)
            draft=json.loads(db.one(c,'SELECT pending_json FROM conversations')['pending_json'])
            self.assertEqual(draft['items'][0]['quantity'],5)
        self.message('no')
        with db.connect() as c: self.assertIsNone(db.one(c,'SELECT pending_json FROM conversations')['pending_json'])
    def test_ambiguous_products_unknown_text_and_units(self):
        with db.connect() as c:
            for text in ('5 box cleaner aur 2 mystery chahiye','5 box cleaner and 2 carton tissue plus unknown','5 pcs cleaner bhej do','cleaner bhej do'):
                _,issues=extract(c,text);self.assertTrue(issues,text)
            c.execute("INSERT INTO products(business_id,name,sku,unit,aliases) VALUES(1,'Other cleaner','OTHER','box','cleaner')")
            _,issues=extract(c,'5 box cleaner bhej do');self.assertTrue(issues)
    def test_takeover_stops_processing_and_queued_replies(self):
        self.message('5 box cleaner bhej do')
        with db.connect() as c: c.execute('UPDATE conversations SET takeover=1')
        mid=self.message('yes')
        run_once()
        with db.connect() as c:
            self.assertEqual(db.one(c,'SELECT state FROM messages WHERE id=?',(mid,))['state'],'human')
            self.assertEqual(db.one(c,'SELECT state FROM outbox')['state'],'suppressed')
            self.assertEqual(c.execute('SELECT COUNT(*) FROM orders').fetchone()[0],0)
    def test_media_is_saved_and_reviewed(self):
        with db.connect() as c: mid=receive(c,1,'919000000000','voice note',kind='audio',raw={'audio':{'id':'media123'}})
        with db.connect() as c: process(c,mid)
        with db.connect() as c:
            m=db.one(c,'SELECT * FROM messages WHERE id=?',(mid,));self.assertEqual(m['state'],'review');self.assertIn('media123',m['raw_json'])
    def test_ai_failure_does_not_lose_message(self):
        with patch.dict(os.environ,{'OLLAMA_MODEL':'test'}),patch('llm.understand',side_effect=TimeoutError):
            mid=self.message('5 box cleaner')
        with db.connect() as c: self.assertEqual(db.one(c,'SELECT state FROM messages WHERE id=?',(mid,))['state'],'review')
    def test_ai_low_confidence_cannot_create_draft(self):
        result={'intent':'NEW_ORDER','confidence':0.3,'repeat':False,'area_id':1,'items':[{'product_id':1,'quantity':5,'unit':'box'}],'issues':[]}
        with patch.dict(os.environ,{'OLLAMA_MODEL':'test'}),patch('llm.understand',return_value=result):mid=self.message('5 box cleaner')
        with db.connect() as c:
            self.assertEqual(db.one(c,'SELECT state FROM messages WHERE id=?',(mid,))['state'],'review')
            self.assertIsNone(db.one(c,'SELECT pending_json FROM conversations')['pending_json'])
    def test_calendar_cutoff_holiday_departure(self):
        with db.connect() as c:
            self.assertEqual(next_delivery(c,1,datetime(2026,10,8,12,tzinfo=IST)),'2026-10-12')
            self.assertEqual(next_delivery(c,1,datetime(2026,10,11,19,59,tzinfo=IST)),'2026-10-12')
            self.assertEqual(next_delivery(c,1,datetime(2026,10,11,20,0,tzinfo=IST)),'2026-10-19')
            c.execute("INSERT INTO holidays(business_id,date) VALUES(1,'2026-10-12')")
            self.assertEqual(next_delivery(c,1,datetime(2026,10,8,12,tzinfo=IST)),'2026-10-19')
            c.execute('DELETE FROM holidays')
            c.execute("INSERT INTO delivery_schedules(route_id,date,departed) VALUES(1,'2026-10-12',1)")
            self.assertEqual(next_delivery(c,1,datetime(2026,10,8,12,tzinfo=IST)),'2026-10-19')
            c.execute('UPDATE routes SET active=0 WHERE id=1')
            with self.assertRaises(ValueError):next_delivery(c,1)
    def test_same_day_cutoff_and_timezone(self):
        with db.connect() as c:
            c.execute("UPDATE routes SET cutoff_days=0,cutoff_time='07:00' WHERE id=1")
            self.assertEqual(next_delivery(c,1,datetime(2026,10,12,6,59,tzinfo=IST)),'2026-10-12')
            self.assertEqual(next_delivery(c,1,datetime(2026,10,12,7,0,tzinfo=IST)),'2026-10-19')
            with self.assertRaises(ValueError):next_delivery(c,1,datetime(2026,10,12,6,59))
    def test_backup_restores_orders(self):
        self.message('5 box cleaner bhej do');self.message('yes')
        target=Path(self.tmp.name)/'backup.db';db.backup(str(target))
        import sqlite3
        from contextlib import closing
        with closing(sqlite3.connect(target)) as c:self.assertEqual(c.execute('SELECT COUNT(*) FROM orders').fetchone()[0],1)
    def test_csv_import_is_transactional(self):
        from import_customers import import_csv
        path=Path(self.tmp.name)/'customers.csv'
        path.write_text('name,phone,area,address\nRetailer One,919111222333,Demo Market A,Shop 12\nBad Area,919111222334,Unknown,Shop 13\n',encoding='utf-8')
        with self.assertRaises(ValueError):
            with db.connect() as c:import_csv(c,path)
        with db.connect() as c:self.assertIsNone(db.one(c,"SELECT id FROM customers WHERE phone='919111222333'"))
        path.write_text('name,phone,area,address\nRetailer One,919111222333,Demo Market A,Shop 12\n',encoding='utf-8')
        with db.connect() as c:self.assertEqual(import_csv(c,path),1)
        with db.connect() as c:self.assertEqual(db.one(c,"SELECT area_id FROM customers WHERE phone='919111222333'")['area_id'],1)
    def test_number_isolation_and_customer_reuse(self):
        with db.connect() as c:
            c.execute("INSERT INTO whatsapp_numbers(business_id,label,phone_number_id) VALUES(1,'Branch 2','123456')")
            mid=receive(c,2,'919000000000','2 carton tissue chahiye','wamid.branch2')
        with db.connect() as c:process(c,mid)
        with db.connect() as c:
            self.assertEqual(c.execute('SELECT COUNT(*) FROM customers').fetchone()[0],6)
            conv=db.one(c,'SELECT * FROM conversations');self.assertEqual(conv['number_id'],2)
    def test_adapter_source_events(self):
        from adapters import wa_akg,wechaty
        self.assertEqual(wa_akg({'data':{'key':{'id':'a'},'from':'919000000000@s.whatsapp.net','type':'TEXT','content':'test'}})['phone'],'919000000000')
        self.assertEqual(wechaty({'id':'b','phone':'919000000000','text':'test'})['external_id'],'wechaty-b')
        with self.assertRaises(ValueError):wa_akg({'data':{'isGroup':True}})
        with self.assertRaises(ValueError):wechaty({'phone':'919000000000','text':'test'})

class HTTPTests(DatabaseCase):
    def setUp(self):
        super().setUp()
        self.server=ThreadingHTTPServer(('127.0.0.1',0),Handler)
        self.thread=threading.Thread(target=self.server.serve_forever,daemon=True);self.thread.start()
        self.base='http://127.0.0.1:'+str(self.server.server_port)
        self.headers={}
    def tearDown(self):
        self.server.shutdown();self.server.server_close();self.thread.join();super().tearDown()
    def req(self,path,method='GET',data=None,headers=None):
        req=urllib.request.Request(self.base+path,data=json.dumps(data).encode() if data is not None else None,method=method,headers={**self.headers,**(headers or {})})
        try:
            with urllib.request.urlopen(req) as res:return res.status,res.read(),dict(res.headers)
        except urllib.error.HTTPError as res:
            result=res.code,res.read(),dict(res.headers)
            res.close()
            return result
    def login(self,user='admin',password='test-password-12345'):
        status,body,headers=self.req('/api/login','POST',{'username':user,'password':password})
        self.assertEqual(status,200)
        self.headers={'Cookie':headers['Set-Cookie'].split(';')[0],'X-CSRF-Token':json.loads(body)['csrf']}
    def test_auth_csrf_and_static_secret_protection(self):
        self.assertEqual(self.req('/api/state')[0],401)
        self.assertEqual(self.req('/.env')[0],401)
        self.assertEqual(self.req('/')[0],200)
        self.login()
        self.assertEqual(self.req('/api/customers','POST',{'name':'New','phone':'919111222333'},headers={'X-CSRF-Token':''})[0],403)
        self.assertEqual(self.req('/api/customers','POST',{'name':'New','phone':'919111222333'})[0],200)
    def test_signed_webhook_durable_and_idempotent(self):
        payload={'entry':[{'changes':[{'value':{'metadata':{'phone_number_id':'local'},'messages':[{'from':'919000000000','id':'wamid-test','type':'text','text':{'body':'5 box cleaner bhej do'}}]}}]}]}
        raw=json.dumps(payload).encode();sig='sha256='+hmac.new(b'app-secret',raw,hashlib.sha256).hexdigest()
        with patch.dict(os.environ,{'WHATSAPP_APP_SECRET':'app-secret','WHATSAPP_VERIFY_TOKEN':'verify-me'}):
            self.assertEqual(self.req('/webhook/whatsapp','POST',payload)[0],403)
            self.assertEqual(self.req('/webhook/whatsapp','POST',payload,{'X-Hub-Signature-256':sig})[0],200)
            self.assertEqual(self.req('/webhook/whatsapp','POST',payload,{'X-Hub-Signature-256':sig})[0],200)
            status,body,_=self.req('/webhook/whatsapp?hub.mode=subscribe&hub.verify_token=verify-me&hub.challenge=hello')
            self.assertEqual((status,body),(200,b'hello'))
        with db.connect() as c:self.assertEqual(c.execute('SELECT COUNT(*) FROM webhook_events').fetchone()[0],1)
        run_once()
        with db.connect() as c:self.assertEqual(c.execute("SELECT COUNT(*) FROM messages WHERE direction='in'").fetchone()[0],1)
    def test_driver_cannot_access_admin_or_other_orders(self):
        with db.connect() as c:
            c.execute('INSERT INTO users(business_id,username,password_hash,role,driver_id) VALUES(1,?,?,\'driver\',1)',('driver',db.password_hash('driver-password-123')))
            oid=create_order(c,1,[{'product_id':1,'quantity':5}])
        self.login('driver','driver-password-123')
        self.assertEqual(self.req('/api/state')[0],403)
        self.assertEqual(self.req('/api/driver')[0],200)
        self.assertEqual(self.req('/api/delivery','POST',{'order_id':oid,'status':'Delivered'})[0],400)
    def test_staff_cannot_change_configuration(self):
        with db.connect() as c:c.execute('INSERT INTO users(business_id,username,password_hash,role) VALUES(1,?,?,\'staff\')',('staff',db.password_hash('staff-password-123')))
        self.login('staff','staff-password-123')
        self.assertEqual(self.req('/api/settings','POST',{'automation':'false'})[0],403)
        self.assertEqual(self.req('/api/state')[0],200)
    def test_staff_can_book_review_and_queue_confirmation(self):
        mid=self.message('cleaner bhej do')
        self.login()
        status,body,_=self.req('/api/orders','POST',{'customer_id':1,'items':[{'product_id':1,'quantity':5}],'review_message_id':mid})
        self.assertEqual(status,200,body)
        with db.connect() as c:
            self.assertEqual(db.one(c,'SELECT state FROM messages WHERE id=?',(mid,))['state'],'review_resolved')
            self.assertTrue(json.loads(db.one(c,'SELECT payload_json FROM outbox ORDER BY id DESC LIMIT 1')['payload_json'])['manual'])
    def test_unconfigured_number_event_is_retained(self):
        with db.connect() as c:c.execute('INSERT INTO webhook_events(body_hash,raw_body,created_at) VALUES(?,?,?)',('bad',json.dumps({'entry':[{'changes':[{'value':{'metadata':{'phone_number_id':'unknown'},'messages':[{'id':'test','from':'919000000000'}]}}]}]}),db.now()))
        run_once()
        with db.connect() as c:
            event=db.one(c,'SELECT * FROM webhook_events');self.assertEqual(event['state'],'failed');self.assertIn('unknown',event['raw_body'])

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