lou
96d44569fd
- proto.py: 帧协议 [type][req_id][len][payload], req_id 多路复用 - client.py: TLS 长连 + AUTH + 断线自动重连(全标准库, OpenWrt 可跑) - proxy_http.py: 本地 HTTP 透明代理(明文 80 走通道, 失败回 502) - proxy_dns.py: DNS 分流(内网域 -> 学校 DNS, 公网域 -> 云通道递归) - server.py: 云端 AUTH + HTTPS 套壳(统一 UA, 剔 hop-by-hop)+ DNS 递归 - deploy/hook.sh: REDIRECT 规则(内网段放行, 认证 eportal 绝不走通道) - 30 个新单测(40 总)+ smoke_channel 端到端冒烟 PASS
62 lines
1.8 KiB
Python
62 lines
1.8 KiB
Python
"""proto 帧协议单测: 打包/切帧/边界。"""
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from channel.proto import (
|
|
FrameReader,
|
|
HEADER_LEN,
|
|
MAX_PAYLOAD,
|
|
TYPE_DNS_REQ,
|
|
TYPE_HTTP_RESP,
|
|
pack_frame,
|
|
)
|
|
|
|
|
|
class ProtoTest(unittest.TestCase):
|
|
def test_pack_roundtrip(self) -> None:
|
|
payload = b"\x01\x02abc"
|
|
frame = pack_frame(TYPE_DNS_REQ, 7, payload)
|
|
self.assertEqual(len(frame), HEADER_LEN + len(payload))
|
|
reader = FrameReader()
|
|
reader.feed(frame)
|
|
got = reader.poll()
|
|
assert got is not None
|
|
self.assertEqual(got, (TYPE_DNS_REQ, 7, payload))
|
|
|
|
def test_reader_handles_partial_and_multiple(self) -> None:
|
|
f1 = pack_frame(TYPE_HTTP_RESP, 1, b"hello")
|
|
f2 = pack_frame(TYPE_DNS_REQ, 2, b"")
|
|
reader = FrameReader()
|
|
reader.feed(f1[:5])
|
|
self.assertIsNone(reader.poll())
|
|
reader.feed(f1[5:] + f2[:3])
|
|
got1 = reader.poll()
|
|
assert got1 is not None
|
|
self.assertEqual(got1, (TYPE_HTTP_RESP, 1, b"hello"))
|
|
self.assertIsNone(reader.poll()) # f2 还不完整
|
|
reader.feed(f2[3:])
|
|
got2 = reader.poll()
|
|
assert got2 is not None
|
|
self.assertEqual(got2, (TYPE_DNS_REQ, 2, b""))
|
|
|
|
def test_empty_payload_frame(self) -> None:
|
|
frame = pack_frame(0, 0)
|
|
reader = FrameReader()
|
|
reader.feed(frame)
|
|
got = reader.poll()
|
|
assert got is not None
|
|
self.assertEqual(got, (0, 0, b""))
|
|
|
|
def test_oversize_payload_rejected(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
pack_frame(0, 0, b"x" * (MAX_PAYLOAD + 1))
|
|
|
|
def test_bad_type_rejected(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
pack_frame(300, 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|