70 lines
2.8 KiB
Python
70 lines
2.8 KiB
Python
import json
|
|
import threading
|
|
import unittest
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
from mailbox_server import make_server
|
|
from mailbox_store import SqliteMailboxStore
|
|
|
|
|
|
def env(i, to="+821022223333"):
|
|
return {"id": i, "from": "+821011112222", "to": to, "kind": "chat.message",
|
|
"unit": "room1", "body": "SEALED", "stamp": 1}
|
|
|
|
|
|
class MailboxTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.store = SqliteMailboxStore(max_per_recipient=3)
|
|
self.srv = make_server(self.store)
|
|
self.base = f"http://127.0.0.1:{self.srv.server_address[1]}"
|
|
threading.Thread(target=self.srv.serve_forever, daemon=True).start()
|
|
|
|
def tearDown(self):
|
|
self.srv.shutdown()
|
|
|
|
def call(self, method, path, body=None, token=None):
|
|
req = urllib.request.Request(self.base + path, method=method,
|
|
data=json.dumps(body).encode() if body is not None else None)
|
|
req.add_header("Content-Type", "application/json")
|
|
if token:
|
|
req.add_header("X-Fm-Token", token)
|
|
try:
|
|
with urllib.request.urlopen(req) as r:
|
|
return r.status, json.loads(r.read())
|
|
except urllib.error.HTTPError as e:
|
|
return e.code, json.loads(e.read())
|
|
|
|
def test_send_fetch_ack(self):
|
|
self.assertEqual(self.call("POST", "/v1/send", env("a"))[0], 200)
|
|
self.assertEqual(self.call("POST", "/v1/send", env("a"))[0], 200) # 같은 id 는 한 번만
|
|
code, res = self.call("GET", "/v1/fetch?to=%2B821022223333", token="T1")
|
|
self.assertEqual(code, 200)
|
|
self.assertEqual([e["id"] for e in res["items"]], ["a"])
|
|
self.call("POST", "/v1/ack", {"to": "+821022223333", "ids": ["a"]}, token="T1")
|
|
self.assertEqual(self.call("GET", "/v1/fetch?to=%2B821022223333", token="T1")[1]["items"], [])
|
|
|
|
def test_other_token_is_refused(self):
|
|
self.call("GET", "/v1/fetch?to=%2B821022223333", token="T1")
|
|
self.assertEqual(self.call("GET", "/v1/fetch?to=%2B821022223333", token="T2")[0], 403)
|
|
self.assertEqual(self.call("GET", "/v1/fetch?to=%2B821022223333")[0], 403)
|
|
|
|
def test_bad_input(self):
|
|
self.assertEqual(self.call("POST", "/v1/send", {"id": "x"})[0], 400)
|
|
self.assertEqual(self.call("POST", "/v1/send", env("b", to="abc"))[0], 400)
|
|
self.assertEqual(self.call("GET", "/v1/fetch?to=abc", token="T")[0], 400)
|
|
|
|
def test_limit_per_recipient(self):
|
|
for i in "abc":
|
|
self.assertEqual(self.call("POST", "/v1/send", env(i))[0], 200)
|
|
self.assertEqual(self.call("POST", "/v1/send", env("d"))[0], 429)
|
|
|
|
def test_time(self):
|
|
code, res = self.call("GET", "/v1/time")
|
|
self.assertEqual(code, 200)
|
|
self.assertGreater(res["stamp"], 1_700_000_000 * 100)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|