first commit

test/test_controller.py

@@ -0,0 +1,75 @@
+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)
+
+
+if __name__ == '__main__':
+    unittest.main()