fix: 配置 ROS_URL 后概览页 500（'Response' object has no attribute 'get'）（v0.1.3）
httpx 的 raise_for_status() 成功时返回响应对象本身，原写法把响应对象当成 JSON 存进了路由器状态。 同时给出可操作的 RouterOS 连接错误提示，并新增模拟 REST 接口的测试。 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Su YijunandClaude Opus 5.5 committed
1 parent
c8af9373ac
commit
ab5c9f78e5
7 files changed
+115
-18
No files matched your search
@@ -1,5 +1,11 @@
|
||||
# Changelog
|
||||
|
||||
## [0.1.3] - 2026-10-22
|
||||
|
||||
- 修复：配置 RouterOS API（`ROS_URL`）后，概览页报“服务器内部错误：AttributeError: 'Response' object has no attribute 'get'”，页面打不开。原因是读取 RouterOS 返回数据时把 HTTP 响应对象当成了 JSON。
|
||||
- RouterOS 连接失败时给出具体提示：401 账号或密码错误、403 权限不足、连不上、超时。轮询失败时保留上一次成功的路由器信息。
|
||||
- 新增模拟 RouterOS REST 接口的测试。
|
||||
|
||||
## [0.1.2] - 2026-10-22
|
||||
|
||||
debug2
|
||||
|
||||
@@ -8,6 +8,7 @@ Source & docs: https://github.com/suyijun8182/ros-dpi
|
||||
|
||||
## What's new
|
||||
|
||||
- **0.1.3**: fix the 500 error on the overview page when the RouterOS API (`ROS_URL`) is configured; clearer RouterOS connection errors.
|
||||
- **0.1.2**: first-run setup wizard: enter your MySQL connection in the browser, like EasySub. The compose file no longer needs database settings. A README guide covers creating the database.
|
||||
- **0.1.1**: full `docker-compose.yml` on this page (no `.env` needed), first-run notes, database auto-created on external MySQL.
|
||||
- **0.1.0**: first release. Overview, hosts, applications, flow Sankey, custom rules, setup guide. zh/en, light/dark.
|
||||
|
||||
@@ -1 +1 @@
|
||||
__version__ = "0.1.2"
|
||||
__version__ = "0.1.3"
|
||||
+38
-14
@@ -31,30 +31,38 @@ class RosPoller:
|
||||
return httpx.Client(base_url=s.ros_url.rstrip("/") + "/rest", auth=(s.ros_user, s.ros_password),
|
||||
verify=s.ros_verify_tls, timeout=10.0)
|
||||
|
||||
@staticmethod
|
||||
def _get(c: httpx.Client, path: str):
|
||||
# 注意：httpx 的 raise_for_status() 成功时返回响应对象本身，不能写成 `raise_for_status() or r.json()`
|
||||
r = c.get(path)
|
||||
r.raise_for_status()
|
||||
return r.json()
|
||||
|
||||
def poll(self):
|
||||
if not self.enabled:
|
||||
return
|
||||
try:
|
||||
with self._client() as c:
|
||||
get = lambda p: (r := c.get(p)).raise_for_status() or r.json() # noqa: E731
|
||||
ident = get("/system/identity")
|
||||
self.identity = ident.get("name", "") if isinstance(ident, dict) else ""
|
||||
self.resource = get("/system/resource")
|
||||
ifaces = get("/interface")
|
||||
arp = get("/ip/arp")
|
||||
leases = get("/ip/dhcp-server/lease")
|
||||
addrs = get("/ip/address")
|
||||
routes = get("/ip/route?dst-address=0.0.0.0/0")
|
||||
ident = self._get(c, "/system/identity")
|
||||
resource = self._get(c, "/system/resource")
|
||||
ifaces = self._get(c, "/interface")
|
||||
arp = self._get(c, "/ip/arp")
|
||||
leases = self._get(c, "/ip/dhcp-server/lease")
|
||||
addrs = self._get(c, "/ip/address")
|
||||
routes = self._get(c, "/ip/route?dst-address=0.0.0.0/0")
|
||||
try:
|
||||
nb6 = get("/ipv6/neighbor")
|
||||
addrs6 = get("/ipv6/address")
|
||||
nb6 = self._get(c, "/ipv6/neighbor")
|
||||
addrs6 = self._get(c, "/ipv6/address")
|
||||
except httpx.HTTPError:
|
||||
nb6, addrs6 = [], [] # IPv6 包未启用
|
||||
self._apply(ifaces, arp, leases, addrs, routes, nb6, addrs6)
|
||||
# 全部成功后再更新展示用的状态，失败时保留上一次的结果
|
||||
self.identity = ident.get("name", "") if isinstance(ident, dict) else ""
|
||||
self.resource = resource if isinstance(resource, dict) else {}
|
||||
self.ok, self.error = True, ""
|
||||
except Exception as e: # noqa: BLE001
|
||||
self.ok, self.error = False, f"{type(e).__name__}: {str(e)[:200]}"
|
||||
print(f"[ros] 轮询失败：{self.error}")
|
||||
self.ok, self.error = False, _explain(e)
|
||||
print(f"[ros] 轮询失败：{self.error}", flush=True)
|
||||
self.last_poll = int(time.time())
|
||||
|
||||
def _apply(self, ifaces, arp, leases, addrs, routes, nb6, addrs6):
|
||||
@@ -112,7 +120,7 @@ class RosPoller:
|
||||
self.engine.set_auto_nets(nets)
|
||||
|
||||
def status(self) -> dict:
|
||||
r = self.resource or {}
|
||||
r = self.resource if isinstance(self.resource, dict) else {}
|
||||
return {
|
||||
"enabled": self.enabled, "ok": self.ok, "error": self.error, "last_poll": self.last_poll,
|
||||
"identity": self.identity, "version": r.get("version"), "board": r.get("board-name"),
|
||||
@@ -122,6 +130,22 @@ class RosPoller:
|
||||
}
|
||||
|
||||
|
||||
def _explain(e: Exception) -> str:
|
||||
"""常见错误翻译成可操作的提示。"""
|
||||
if isinstance(e, httpx.HTTPStatusError):
|
||||
code = e.response.status_code
|
||||
if code == 401:
|
||||
return "RouterOS 拒绝登录（401）：检查 ROS_USER / ROS_PASSWORD，以及该用户的 address 是否允许本机 IP"
|
||||
if code == 403:
|
||||
return "RouterOS 权限不足（403）：用户组需要 read、rest-api 权限"
|
||||
return f"RouterOS 返回 HTTP {code}：{e.request.url.path}"
|
||||
if isinstance(e, httpx.ConnectError):
|
||||
return f"连不上 RouterOS：检查 ROS_URL，并确认 /ip service 中 www-ssl（或 www）已启用（{e}）"
|
||||
if isinstance(e, httpx.TimeoutException):
|
||||
return "连接 RouterOS 超时：检查 ROS_URL 与防火墙"
|
||||
return f"{type(e).__name__}: {str(e)[:200]}"
|
||||
|
||||
|
||||
def _int(v):
|
||||
try:
|
||||
return int(v)
|
||||
|
||||
@@ -0,0 +1,66 @@
|
||||
import httpx
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.config import get_settings
|
||||
from app.engine import engine
|
||||
from app.main import app
|
||||
from app.ros_poller import poller
|
||||
|
||||
# 模拟 RouterOS 7 REST 接口的返回（字段值都是字符串，与真实设备一致）
|
||||
ROS = {
|
||||
"/rest/system/identity": {"name": "MikroTik-Home"},
|
||||
"/rest/system/resource": {"version": "7.23beta", "board-name": "RB5009", "uptime": "3d2h", "cpu-load": "7",
|
||||
"total-memory": "1073741824", "free-memory": "805306368"},
|
||||
"/rest/interface": [{"name": "ether1", "type": "ether", "running": "true", "rx-byte": "1000", "tx-byte": "500"},
|
||||
{"name": "bridge", "type": "bridge", "running": "true", "rx-byte": "10", "tx-byte": "20"}],
|
||||
"/rest/ip/arp": [{"address": "192.168.88.30", "mac-address": "aa:bb:cc:dd:ee:30"}],
|
||||
"/rest/ip/dhcp-server/lease": [{"address": "192.168.88.31", "mac-address": "AA:BB:CC:DD:EE:31",
|
||||
"host-name": "iPhone", "comment": "小明的手机"}],
|
||||
"/rest/ip/address": [{"address": "192.168.88.1/24", "interface": "bridge"},
|
||||
{"address": "100.64.1.2/24", "interface": "ether1"}],
|
||||
"/rest/ip/route": [{"dst-address": "0.0.0.0/0", "gateway": "100.64.1.1", "immediate-gw": "100.64.1.1%ether1",
|
||||
"active": "true"}],
|
||||
"/rest/ipv6/neighbor": [{"address": "2408:8000:aa:bb::30", "mac-address": "AA:BB:CC:DD:EE:30"}],
|
||||
"/rest/ipv6/address": [{"address": "2408:8000:aa:bb::1/64", "interface": "bridge"},
|
||||
{"address": "fe80::1/64", "interface": "bridge"}],
|
||||
}
|
||||
|
||||
|
||||
def _use_mock(monkeypatch, status=200):
|
||||
def handler(req: httpx.Request):
|
||||
if status != 200:
|
||||
return httpx.Response(status, json={"error": status})
|
||||
body = ROS.get(req.url.path)
|
||||
return httpx.Response(200, json=body) if body is not None else httpx.Response(404, json={})
|
||||
|
||||
monkeypatch.setattr(get_settings(), "ros_url", "https://192.168.88.1")
|
||||
monkeypatch.setattr(poller, "_client",
|
||||
lambda: httpx.Client(base_url="https://192.168.88.1/rest", transport=httpx.MockTransport(handler)))
|
||||
|
||||
|
||||
def test_poll_success_populates_router_state(monkeypatch):
|
||||
_use_mock(monkeypatch)
|
||||
poller.poll()
|
||||
st = poller.status()
|
||||
assert st["ok"] and st["error"] == ""
|
||||
assert st["identity"] == "MikroTik-Home" and st["version"] == "7.23beta" and st["cpu_load"] == 7
|
||||
assert st["wan_interface"] == "ether1"
|
||||
assert engine.ip2mac["192.168.88.30"] == "AA:BB:CC:DD:EE:30"
|
||||
assert engine.mac2name["AA:BB:CC:DD:EE:31"] == "小明的手机"
|
||||
nets = {str(n) for n in engine.auto_nets}
|
||||
assert "2408:8000:aa:bb::/64" in nets and "100.64.1.0/24" not in nets # WAN 口的地址段不算 LAN
|
||||
assert engine.is_lan("2408:8000:aa:bb::30")
|
||||
|
||||
# 配置了 ROS 后，概览页能正常返回路由器信息（此前这里会 500）
|
||||
with TestClient(app) as c:
|
||||
if not c.get("/api/auth/status").json()["initialized"]:
|
||||
c.post("/api/auth/setup", json={"username": "admin", "password": "secret1"})
|
||||
tok = c.post("/api/auth/login", json={"username": "admin", "password": "secret1"}).json()["token"]
|
||||
r = c.get("/api/overview", headers={"Authorization": f"Bearer {tok}"})
|
||||
assert r.status_code == 200 and r.json()["router"]["identity"] == "MikroTik-Home"
|
||||
|
||||
def test_poll_401_gives_actionable_error_and_keeps_status_usable(monkeypatch):
|
||||
_use_mock(monkeypatch, status=401)
|
||||
poller.poll()
|
||||
st = poller.status()
|
||||
assert not st["ok"] and "401" in st["error"] and "ROS_USER" in st["error"]
|
||||
Generated
+2
-2
@@ -1,12 +1,12 @@
|
||||
{
|
||||
"name": "ros-dpi-frontend",
|
||||
"version": "0.1.2",
|
||||
"version": "0.1.3",
|
||||
"lockfileVersion": 3,
|
||||
"requires": true,
|
||||
"packages": {
|
||||
"": {
|
||||
"name": "ros-dpi-frontend",
|
||||
"version": "0.1.2",
|
||||
"version": "0.1.3",
|
||||
"dependencies": {
|
||||
"@fontsource/barlow": "^5.3.0",
|
||||
"@fontsource/barlow-semi-condensed": "^5.3.0",
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "ros-dpi-frontend",
|
||||
"version": "0.1.2",
|
||||
"version": "0.1.3",
|
||||
"private": true,
|
||||
"type": "module",
|
||||
"scripts": {
|
||||
|
||||
Reference in new issue
Block a user