frindlymap/server/test_verify.py

import json
import re
import threading
import unittest
from urllib.parse import quote
import urllib.error
import urllib.request

from mailbox_server import make_server
from mailbox_store import SqliteMailboxStore
from phone_verify import SmsSender, Verifier

PHONE = "+821022223333"
TOKEN_A = "a" * 32
TOKEN_B = "b" * 32


class Capture(SmsSender):
    def __init__(self):
        self.sent, self.ok = [], True

    def send(self, phone, text):
        if self.ok:
            self.sent.append((phone, text))
        return self.ok

    def last_code(self):
        return re.search(r"\b(\d{6})\b", self.sent[-1][1]).group(1)


class Clock:
    def __init__(self):
        self.t = 1000.0

    def __call__(self):
        return self.t


class VerifierTest(unittest.TestCase):
    def setUp(self):
        self.sms, self.clock = Capture(), Clock()
        self.v = Verifier(self.sms, self.clock)

    def test_correct_code_confirms_once(self):
        self.assertEqual(self.v.start(PHONE), "ok")
        code = self.sms.last_code()
        self.assertTrue(self.v.confirm(PHONE, code))
        self.assertFalse(self.v.confirm(PHONE, code))  # 한 번 쓴 번호는 못 쓴다

    def test_wrong_code_and_tries_limit(self):
        self.v.start(PHONE)
        code = self.sms.last_code()
        for _ in range(5):
            self.assertFalse(self.v.confirm(PHONE, "000000" if code != "000000" else "111111"))
        self.assertFalse(self.v.confirm(PHONE, code))  # 5번 틀리면 맞는 번호도 거절

    def test_expires_after_five_minutes(self):
        self.v.start(PHONE)
        code = self.sms.last_code()
        self.clock.t += 301
        self.assertFalse(self.v.confirm(PHONE, code))

    def test_send_rate_limit_per_hour(self):
        for _ in range(3):
            self.assertEqual(self.v.start(PHONE), "ok")
        self.assertEqual(self.v.start(PHONE), "rate")
        self.clock.t += 3601
        self.assertEqual(self.v.start(PHONE), "ok")

    def test_sender_failure_does_not_count(self):
        self.sms.ok = False
        self.assertEqual(self.v.start(PHONE), "fail")
        self.sms.ok = True
        for _ in range(3):
            self.assertEqual(self.v.start(PHONE), "ok")

    def test_new_request_replaces_old_code(self):
        self.v.start(PHONE)
        old = self.sms.last_code()
        self.v.start(PHONE)
        new = self.sms.last_code()
        if old != new:
            self.assertFalse(self.v.confirm(PHONE, old))
        self.assertTrue(self.v.confirm(PHONE, new) or old == new)


class ServerVerifyTest(unittest.TestCase):
    def setUp(self):
        self.sms = Capture()
        self.store = SqliteMailboxStore()
        self.srv = make_server(self.store, verifier=Verifier(self.sms), strict=True)
        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 verify(self, token):
        self.assertEqual(self.call("POST", "/v1/verify/start", {"phone": PHONE})[0], 200)
        return self.call("POST", "/v1/verify/confirm", {"phone": PHONE, "code": self.sms.last_code(), "token": token})

    def test_strict_mode_blocks_unverified_squatter(self):
        self.assertEqual(self.call("GET", f"/v1/fetch?to={quote(PHONE)}", token=TOKEN_A)[0], 403)  # 선점 시도 막힘

    def test_verified_device_opens_mailbox(self):
        self.assertEqual(self.verify(TOKEN_A)[0], 200)
        self.assertEqual(self.call("GET", f"/v1/fetch?to={quote(PHONE)}", token=TOKEN_A)[0], 200)
        self.assertEqual(self.call("GET", f"/v1/fetch?to={quote(PHONE)}", token=TOKEN_B)[0], 403)

    def test_real_owner_takes_over_after_new_verification(self):
        self.verify(TOKEN_A)
        self.assertEqual(self.verify(TOKEN_B)[0], 200)  # 폰을 바꾼 진짜 주인
        self.assertEqual(self.call("GET", f"/v1/fetch?to={quote(PHONE)}", token=TOKEN_B)[0], 200)
        self.assertEqual(self.call("GET", f"/v1/fetch?to={quote(PHONE)}", token=TOKEN_A)[0], 403)

    def test_wrong_code_rejected_and_short_token_rejected(self):
        self.call("POST", "/v1/verify/start", {"phone": PHONE})
        self.assertEqual(self.call("POST", "/v1/verify/confirm", {"phone": PHONE, "code": "123", "token": TOKEN_A})[0], 403)
        self.assertEqual(self.call("POST", "/v1/verify/confirm", {"phone": PHONE, "code": self.sms.last_code(), "token": "짧음"})[0], 400)

    def test_rate_limit_returns_429(self):
        for _ in range(3):
            self.assertEqual(self.call("POST", "/v1/verify/start", {"phone": PHONE})[0], 200)
        self.assertEqual(self.call("POST", "/v1/verify/start", {"phone": PHONE})[0], 429)

    def test_bad_phone_rejected(self):
        self.assertEqual(self.call("POST", "/v1/verify/start", {"phone": "010-1234"})[0], 400)


class NoVerifierTest(unittest.TestCase):
    def test_endpoints_unavailable_and_tofu_still_works(self):
        store = SqliteMailboxStore()
        srv = make_server(store)
        threading.Thread(target=srv.serve_forever, daemon=True).start()
        base = f"http://127.0.0.1:{srv.server_address[1]}"
        req = urllib.request.Request(base + "/v1/verify/start", method="POST", data=json.dumps({"phone": PHONE}).encode())
        with self.assertRaises(urllib.error.HTTPError) as cm:
            urllib.request.urlopen(req)
        self.assertEqual(cm.exception.code, 400)
        self.assertTrue(store.check_token(PHONE, TOKEN_A))  # 문자 확인을 안 쓰면 예전처럼 처음 쓴 기기에 묶인다
        srv.shutdown()


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