89 lines
4.3 KiB
Python
89 lines
4.3 KiB
Python
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'])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|