import sys
import json
import uuid
import shutil
import unittest
from pathlib import Path
from datetime import datetime,timezone
sys.path.insert(0,str(Path(__file__).resolve().parents[1]))
import db
from linked import ingest,matched_flow,ensure_number
from engine import process

class LinkedTests(unittest.TestCase):
    def setUp(self):
        self.previous=db.DB_PATH
        self.folder=db.ROOT/'tests'/('linked-'+uuid.uuid4().hex);self.folder.mkdir()
        db.DB_PATH=self.folder/'test.db'
        import os
        from unittest.mock import patch
        self.env=patch.dict(os.environ,{'ADMIN_PASSWORD':'test-password-12345','OLLAMA_MODEL':''});self.env.start()
        db.init()
    def tearDown(self):
        db.DB_PATH=self.previous;self.env.stop()
        if self.folder.resolve().parent!=(db.ROOT/'tests').resolve():raise AssertionError('Invalid cleanup target')
        shutil.rmtree(self.folder)
    def event(self,history=False,kind='text',jid='919111222333@s.whatsapp.net',mid='one',body='order 5 cleaner',direction='in',phone='919111222333'):
        return {'type':'messages','messages':[{'jid':jid,'id':mid,'phone':phone,'name':'Retailer','isGroup':jid.endswith('@g.us'),'direction':direction,'kind':kind,'body':body,'history':history,'timestamp':datetime.now(timezone.utc).timestamp(),'raw':{}}]}
    def flow(self,db,action='reply',keywords='order',priority=1,active=1):
        return db.execute('INSERT INTO keyword_flows(business_id,name,keywords,action,response,priority,active) VALUES(1,?,?,?,?,?,?)',('My flow',keywords,action,'Thanks, team will contact you.',priority,active)).lastrowid
    def test_history_never_replies_or_books(self):
        with db.connect() as c:
            self.flow(c);ingest(c,self.event(history=True));self.assertEqual(db.one(c,'SELECT state FROM messages')['state'],'history')
            self.assertEqual(c.execute('SELECT COUNT(*) FROM outbox').fetchone()[0],0)
            self.assertIsNone(db.one(c,'SELECT last_inbound_at FROM conversations')['last_inbound_at'])
    def test_no_flow_no_reply(self):
        with db.connect() as c:
            ingest(c,self.event());self.assertEqual(db.one(c,'SELECT state FROM messages')['state'],'linked_waiting')
            self.assertEqual(c.execute('SELECT COUNT(*) FROM outbox').fetchone()[0],0)
    def test_reply_flow_and_event_deduplication(self):
        with db.connect() as c:
            self.flow(c);ingest(c,self.event());ingest(c,self.event())
            self.assertEqual(c.execute('SELECT COUNT(*) FROM linked_messages').fetchone()[0],1)
            self.assertEqual(c.execute('SELECT COUNT(*) FROM outbox').fetchone()[0],1)
            self.assertEqual(db.one(c,'SELECT name FROM customers')['name'],'Retailer')
    def test_groups_and_unresolved_lids_are_view_only(self):
        with db.connect() as c:
            self.flow(c)
            ingest(c,self.event(jid='123456@g.us',phone=None));ingest(c,self.event(jid='123456789012@lid',phone=None,mid='two'))
            self.assertEqual(c.execute('SELECT COUNT(*) FROM linked_chats').fetchone()[0],2)
            self.assertEqual(c.execute('SELECT COUNT(*) FROM customers').fetchone()[0],0)
            self.assertEqual(c.execute('SELECT COUNT(*) FROM outbox').fetchone()[0],0)
    def test_takeover_and_pause_block_flows(self):
        with db.connect() as c:
            ingest(c,self.event());self.flow(c);c.execute('UPDATE conversations SET takeover=1')
            ingest(c,self.event(mid='two'));self.assertEqual(c.execute('SELECT COUNT(*) FROM outbox').fetchone()[0],0)
            c.execute('UPDATE conversations SET takeover=0');c.execute("UPDATE settings SET value='false' WHERE key='automation'")
            ingest(c,self.event(mid='three'));self.assertEqual(c.execute('SELECT COUNT(*) FROM outbox').fetchone()[0],0)
    def test_priority_disabled_and_modes(self):
        with db.connect() as c:
            low=self.flow(c,priority=1);high=self.flow(c,priority=5)
            self.assertEqual(matched_flow(c,'ORDER please')['id'],high)
            c.execute('UPDATE keyword_flows SET active=0 WHERE id=?',(high,));self.assertEqual(matched_flow(c,'order')['id'],low)
            c.execute("UPDATE keyword_flows SET match_mode='exact' WHERE id=?",(low,));self.assertIsNone(matched_flow(c,'order please'))
    def test_order_flow_queues_confirmation_not_old_history(self):
        from seed import seed
        with db.connect() as c:
            seed(c);self.flow(c,action='order',keywords='bhej')
            event=self.event(phone='919000000000',jid='919000000000@s.whatsapp.net',body='5 box cleaner bhej do')
            ingest(c,event);mid=db.one(c,'SELECT id FROM messages')['id'];process(c,mid)
            self.assertEqual(c.execute('SELECT COUNT(*) FROM orders').fetchone()[0],0)
            self.assertEqual(json.loads(db.one(c,'SELECT pending_json FROM conversations')['pending_json'])['stage'],'confirm')
            ingest(c,self.event(phone='919000000000',jid='919000000000@s.whatsapp.net',body='haan',mid='two'))
            mid=db.one(c,'SELECT id FROM messages ORDER BY id DESC LIMIT 1')['id'];process(c,mid)
            self.assertEqual(c.execute('SELECT COUNT(*) FROM orders').fetchone()[0],1)

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