import json
from pathlib import Path
import sys
import tempfile
import threading
import unittest
from http.server import ThreadingHTTPServer
from urllib.request import Request, urlopen
from urllib.error import HTTPError

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from app import Controller, make_handler


class ControllerTests(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.root = Path(self.tmp.name)
        self.controller = Controller(self.root / 'data', self.root / 'downloads', ledger=None)
        self.server = ThreadingHTTPServer(('127.0.0.1', 0), make_handler(self.controller))
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()
        self.url = 'http://127.0.0.1:' + str(self.server.server_port)

    def tearDown(self):
        self.server.shutdown();self.server.server_close();self.thread.join();self.tmp.cleanup()

    def request(self, path, body=None, token=None, origin=None, host=None):
        headers = {}
        if token: headers['X-Session-Token'] = token
        if origin: headers['Origin'] = origin
        if host: headers['Host'] = host
        if body is not None: headers['Content-Type'] = 'application/json'
        request = Request(self.url + path, data=None if body is None else json.dumps(body).encode(), headers=headers)
        with urlopen(request, timeout=5) as response:
            return response.status, response.read()

    def test_auth_and_origin_boundaries(self):
        for path, kwargs in [('/api/state', {}), ('/api/state', {'token': self.controller.bridge_token}),
                            ('/api/state', {'token': self.controller.ui_token, 'origin': 'https://example.com'}),
                            ('/bridge/heartbeat', {'token': self.controller.ui_token})]:
            with self.assertRaises(HTTPError) as error:
                self.request(path, {}, **kwargs)
            self.assertEqual(error.exception.code, 403)
        with self.assertRaises(HTTPError):
            self.request('/api/session', host='attacker.example')
        code, body = self.request('/api/state', {}, token=self.controller.ui_token)
        self.assertEqual(code, 200)
        self.assertNotIn(self.controller.bridge_token, body.decode())

    def test_static_and_unknown_routes(self):
        self.assertIn('新闻工作台', self.request('/')[1].decode())
        self.assertEqual(self.request('/browser/auto-news.user.js')[0], 200)
        with self.assertRaises(HTTPError):
            self.request('/../store.py')
        with self.assertRaises(HTTPError):
            self.request('/bridge/scan', {}, token=self.controller.bridge_token)

    def test_download_to_queue_without_network(self):
        self.controller.manager.requester = lambda *_: {'code': 1, 'data': {'list': [dict(title='模拟新闻', src='模拟来源', content='<p>模拟正文</p>', url='https://example.com/news')]}}
        self.controller.ui('download', dict(key='fake-key', channel='模拟频道', target=1))
        self.controller.manager.thread.join(5)
        state = self.controller.ui('state', {})
        self.assertEqual(state['download']['saved'], 1)
        self.assertEqual(len(state['jobs']), 1)
        self.assertEqual(state['jobs'][0]['state'], 'queued')
        self.assertTrue(state['jobs'][0]['source_keys'])
        self.assertNotIn('fake-key', json.dumps(state))
        self.controller.scan(self.controller.manager.root)
        self.assertEqual(len(self.controller.store.state()['jobs']), 1)

    def test_bridge_skip_continues_queue(self):
        store = self.controller.store
        store.enqueue(ROOT / 'test/fixtures/article')
        store.configure(dict(fid=125, minimum=1, maximum=1, interval=0, history_checked=True))
        store.control('start')
        job = store.next(dict(owner='tab-a', account=dict(uid='123', name='测试账号')))['job']
        code, body = self.request('/bridge/skip', dict(owner='tab-a', id=job['id'], epoch=store.run['epoch']),
                                  token=self.controller.bridge_token)
        self.assertEqual(code, 200)
        self.assertTrue(json.loads(body)['ok'])
        self.assertEqual(store.state()['jobs'][0]['state'], 'skipped')
        self.assertIsNone(store.run['active'])


    def test_ui_remove_selected_articles(self):
        store = self.controller.store
        store.enqueue(ROOT / 'test/fixtures/article')
        ids = [job['id'] for job in store.state()['jobs']]
        with self.assertRaises(HTTPError):
            self.request('/api/remove', {'ids': ids}, token=self.controller.bridge_token)
        with self.assertRaises(HTTPError):
            self.request('/bridge/remove', {'ids': ids}, token=self.controller.bridge_token)
        code, body = self.request('/api/remove', {'ids': ids}, token=self.controller.ui_token)
        self.assertEqual(code, 200)
        self.assertEqual(json.loads(body)['removed'], 1)
        self.assertEqual(store.state()['jobs'], [])

    def test_ui_historical_result_reconciliation(self):
        store = self.controller.store
        store.enqueue(ROOT / 'test/fixtures/article')
        store.configure(dict(fid=125, minimum=1, maximum=1, interval=0, history_checked=True))
        store.control('start')
        job = store.next(dict(owner='tab-a', account=dict(uid='123', name='测试账号')))['job']
        for state in ('importing','formatting','awaiting_user_review','user_submitting'):
            store.event(dict(owner='tab-a',id=job['id'],state=state,epoch=store.run['epoch'],snapshot={'title':job['title']}))
        store.control('stop')
        data=dict(id=job['id'],article_key=job['article_key'],confirmed=True,thread_id='236999',first_post_pid='238791',
            post_url='https://www.dww11.com/thread/23699911.html',title=job['title'],author_uid='123',fid=125,
            state='accepted_pending_moderation',evidence={'thread_id':'236999'})
        with self.assertRaises(HTTPError):
            self.request('/api/reconcile',data,token=self.controller.bridge_token)
        self.assertEqual(self.request('/api/reconcile',data,token=self.controller.ui_token)[0],200)
        self.assertEqual(store.run['batch']['count'],0)
        self.assertFalse(store.run['running'])


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