348 lines
18 KiB
Python
348 lines
18 KiB
Python
"""Local SQL/state-machine checks; SQLite does not validate MySQL row locks."""
|
|
import asyncio
|
|
import sqlite3
|
|
from uuid import uuid4
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from banban.dao.binding import BindingDAO
|
|
from banban.service.joint_binding import JointBindingService
|
|
from banban.service.traffic_handoff import TrafficError, normalize_iccid
|
|
from banban.service import traffic_handoff
|
|
|
|
DEVICE = "TalkingQ_test_joint"
|
|
ICCID = "898604E91123D1334901"
|
|
|
|
|
|
class Result:
|
|
def __init__(self, cursor):
|
|
self.rows = [dict(row) for row in cursor.fetchall()] if cursor.description else []
|
|
def mappings(self): return self
|
|
def first(self): return self.rows[0] if self.rows else None
|
|
def one(self):
|
|
assert len(self.rows) == 1
|
|
return self.rows[0]
|
|
def all(self): return self.rows
|
|
|
|
|
|
class Session:
|
|
def __init__(self, connection): self.connection = connection
|
|
async def execute(self, statement, params=None):
|
|
return Result(self.connection.execute(str(statement).replace(" FOR UPDATE", ""), params or {}))
|
|
async def commit(self): self.connection.commit()
|
|
async def rollback(self): self.connection.rollback()
|
|
async def close(self): self.connection.rollback()
|
|
|
|
|
|
@pytest.fixture
|
|
def state(monkeypatch):
|
|
conn = sqlite3.connect(":memory:")
|
|
conn.row_factory = sqlite3.Row
|
|
conn.executescript("""
|
|
CREATE TABLE device_bindings (id INTEGER PRIMARY KEY, device_id TEXT UNIQUE, owner_user_id INTEGER,
|
|
child_id INTEGER, status INTEGER, binding_instance_id TEXT, unbound_at TEXT, updated_at TEXT);
|
|
CREATE TABLE device_settings (device_id TEXT PRIMARY KEY, iccid TEXT UNIQUE);
|
|
CREATE TABLE device_binding_operations (operation_id TEXT PRIMARY KEY, binding_instance_id TEXT,
|
|
device_id TEXT, owner_user_id INTEGER, action TEXT, iccid TEXT,
|
|
traffic_status TEXT DEFAULT 'pending', status TEXT DEFAULT 'pending',
|
|
handoff_attempt INTEGER DEFAULT 0, handoff_started INTEGER DEFAULT 0, last_error TEXT,
|
|
UNIQUE(binding_instance_id, action));
|
|
CREATE TABLE device_family_members (device_id TEXT, status INTEGER, removed_at TEXT, updated_at TEXT);
|
|
CREATE TABLE device_bind_history (device_id TEXT, child_id INTEGER, bound_by_user_id INTEGER,
|
|
unbound_by_user_id INTEGER, bind_source INTEGER, bound_at TEXT, unbound_at TEXT, unbind_reason TEXT);
|
|
""")
|
|
conn.execute("INSERT INTO device_bindings VALUES (1, ?, 7, NULL, 1, ?, NULL, NULL)", (DEVICE, str(uuid4())))
|
|
conn.execute("INSERT INTO device_settings VALUES (?, ?)", (DEVICE, ICCID))
|
|
conn.execute("INSERT INTO device_family_members VALUES (?, 1, NULL, NULL)", (DEVICE,))
|
|
conn.commit()
|
|
service = JointBindingService()
|
|
service.test_relation = {"status": "bound", "relationState": "open"}
|
|
service.test_relation_calls = []
|
|
async def relation(source, request_id, *, close=False):
|
|
service.test_relation_calls.append((source, request_id, close))
|
|
if close: service.test_relation["relationState"] = "closing"
|
|
return {**service.test_relation, "sourceBindingId": source, "pendingHandoff": False,
|
|
"acceptingHandoff": service.test_relation["relationState"] == "open",
|
|
"canUnbind": service.test_relation["relationState"] == "closing" and service.test_relation["status"] in {"unbound", "never_bound"}}
|
|
monkeypatch.setattr("banban.service.joint_binding.binding_relation", relation)
|
|
async def session(): return Session(conn)
|
|
monkeypatch.setattr(service, "get_session", session)
|
|
yield service, conn
|
|
conn.close()
|
|
|
|
|
|
async def confirm(service, op, iccid=ICCID, success=True, code="OK"):
|
|
return await service.report(op["operation_id"], 7, op["action"], f"joint-{op['operation_id']}", iccid, code, success)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bind_failure_preserves_banban_and_request_identity(state, monkeypatch):
|
|
service, conn = state
|
|
calls = []
|
|
async def unavailable(*args):
|
|
calls.append(args)
|
|
raise TrafficError("TRAFFIC_UNAVAILABLE", "timeout", 503)
|
|
monkeypatch.setattr("banban.service.joint_binding.create_handoff", unavailable)
|
|
op = await service.start(DEVICE, 7, "bind")
|
|
for _ in range(2):
|
|
assert (await service.start(DEVICE, 7, "bind"))["operation_id"] == op["operation_id"]
|
|
with pytest.raises(TrafficError): await service.prepare(op["operation_id"], 7)
|
|
assert calls[0] == calls[1]
|
|
assert conn.execute("SELECT status FROM device_bindings").fetchone()[0] == 1
|
|
assert (await service.overview(DEVICE, 7))["bind"]["status"] == "pending"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_iccid_never_removes_binding_or_completes_unbind(state):
|
|
service, conn = state
|
|
conn.execute("UPDATE device_settings SET iccid=NULL")
|
|
conn.commit()
|
|
op = await service.start(DEVICE, 7, "unbind")
|
|
with pytest.raises(TrafficError, match="流量卡信息"): await service.prepare(op["operation_id"], 7)
|
|
with pytest.raises(TrafficError): await service.finish_unbind(op["operation_id"], 7)
|
|
assert conn.execute("SELECT status FROM device_bindings").fetchone()[0] == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_successful_unbind_is_atomic_and_idempotent(state):
|
|
service, conn = state
|
|
op = await service.start(DEVICE, 7, "unbind")
|
|
await service.prepare(op["operation_id"], 7)
|
|
with pytest.raises(TrafficError): await service.finish_unbind(op["operation_id"], 7)
|
|
result = await confirm(service, op, success=False, code="NETWORK_ERROR")
|
|
assert result["status"] == "partial_failed"
|
|
with pytest.raises(TrafficError): await service.finish_unbind(op["operation_id"], 7)
|
|
service.test_relation["status"] = "unbound"
|
|
await confirm(service, op)
|
|
for _ in range(2):
|
|
result = await service.finish_unbind(op["operation_id"], 7)
|
|
assert result["status"] == "completed"
|
|
assert conn.execute("SELECT status FROM device_bindings").fetchone()[0] == 0
|
|
assert conn.execute("SELECT status FROM device_family_members").fetchone()[0] == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_local_unbind_failure_rolls_back_and_keeps_4g_receipt(state, monkeypatch):
|
|
service, conn = state
|
|
op = await service.start(DEVICE, 7, "unbind")
|
|
await service.prepare(op["operation_id"], 7)
|
|
service.test_relation["status"] = "unbound"
|
|
await confirm(service, op)
|
|
original = BindingDAO.unbind
|
|
async def fail_after_update(self, *args, **kwargs):
|
|
await original(self, *args, **kwargs)
|
|
raise RuntimeError("local transaction failure")
|
|
monkeypatch.setattr(BindingDAO, "unbind", fail_after_update)
|
|
with pytest.raises(RuntimeError): await service.finish_unbind(op["operation_id"], 7)
|
|
assert conn.execute("SELECT status FROM device_bindings").fetchone()[0] == 1
|
|
assert (await service.get(op["operation_id"], 7))["traffic_status"] == "server_confirmed"
|
|
monkeypatch.setattr(BindingDAO, "unbind", original)
|
|
assert (await service.finish_unbind(op["operation_id"], 7))["status"] == "completed"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_owner_device_action_and_lifecycle_guards(state):
|
|
service, conn = state
|
|
with pytest.raises(TrafficError): await service.start(DEVICE, 8, "bind")
|
|
op = await service.start(DEVICE, 7, "unbind")
|
|
await service.prepare(op["operation_id"], 7)
|
|
with pytest.raises(TrafficError): await service.get(op["operation_id"], 8)
|
|
with pytest.raises(TrafficError): await confirm(service, op, iccid="898604E91123D1334999")
|
|
with pytest.raises(TrafficError): await service.report(op["operation_id"], 7, "bind", f"joint-{op['operation_id']}", ICCID, "OK", True)
|
|
with pytest.raises(TrafficError): await service.report(op["operation_id"], 7, "unbind", "other-request", ICCID, "OK", True)
|
|
conn.execute("UPDATE device_bindings SET binding_instance_id=?", (str(uuid4()),))
|
|
conn.commit()
|
|
with pytest.raises(TrafficError): await confirm(service, op)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_consumed_token_recovers_only_the_original_server_relation(state, monkeypatch):
|
|
service, _ = state
|
|
async def consumed(*args): raise TrafficError("HANDOFF_TOKEN_USED", "used")
|
|
monkeypatch.setattr("banban.service.joint_binding.create_handoff", consumed)
|
|
op = await service.start(DEVICE, 7, "bind")
|
|
result = await service.prepare(op["operation_id"], 7)
|
|
assert result["operation"]["status"] == "completed"
|
|
assert result["operation"]["traffic_status"] == "server_confirmed"
|
|
assert "recovery_only" not in result
|
|
assert (await service.portal(DEVICE, 7))["iccid"] == ICCID
|
|
await service.start(DEVICE, 7, "unbind")
|
|
with pytest.raises(TrafficError): await service.portal(DEVICE, 7)
|
|
with pytest.raises(TrafficError): await service.prepare(op["operation_id"], 7)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("status", ["unbound", "never_bound"])
|
|
async def test_already_released_relation_finishes_without_client_receipt_or_iccid(state, status):
|
|
service, conn = state
|
|
conn.execute("UPDATE device_settings SET iccid=NULL")
|
|
conn.commit()
|
|
service.test_relation["status"] = status
|
|
op = await service.start(DEVICE, 7, "unbind")
|
|
checked = await service.check(op["operation_id"], 7)
|
|
assert checked["traffic_status"] == "pending" # A plain GET on an open relation cannot authorize unbinding.
|
|
prepared = await service.prepare(op["operation_id"], 7)
|
|
assert prepared["operation"]["traffic_status"] == "server_confirmed"
|
|
assert "target_app_id" not in prepared # No redundant jump to the 4G mini-program.
|
|
assert (await service.finish_unbind(op["operation_id"], 7))["status"] == "completed"
|
|
assert conn.execute("SELECT status FROM device_bindings").fetchone()[0] == 0
|
|
close_calls = [call for call in service.test_relation_calls if call[2]]
|
|
assert len(set(close_calls)) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unknown_relation_and_legacy_success_receipt_never_unbind(state):
|
|
service, conn = state
|
|
service.test_relation["status"] = "unknown"
|
|
op = await service.start(DEVICE, 7, "unbind")
|
|
conn.execute("UPDATE device_binding_operations SET traffic_status='client_confirmed'")
|
|
conn.commit()
|
|
with pytest.raises(TrafficError) as failure: await service.finish_unbind(op["operation_id"], 7)
|
|
assert failure.value.code == "HANDOFF_RELATION_UNKNOWN"
|
|
assert conn.execute("SELECT status FROM device_bindings").fetchone()[0] == 1
|
|
with pytest.raises(TrafficError) as failure: await service.start(DEVICE, 7, "bind")
|
|
assert failure.value.code == "UNBIND_PENDING"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_close_network_failure_is_bounded_and_preserves_local_unbinding_marker(state, monkeypatch):
|
|
service, conn = state
|
|
calls = []
|
|
async def unavailable(source, request_id, *, close=False):
|
|
calls.append((source, request_id, close))
|
|
raise TrafficError("TRAFFIC_UNAVAILABLE", "timeout", 503)
|
|
monkeypatch.setattr("banban.service.joint_binding.binding_relation", unavailable)
|
|
op = await service.start(DEVICE, 7, "unbind")
|
|
for _ in range(2):
|
|
with pytest.raises(TrafficError): await service.prepare(op["operation_id"], 7)
|
|
assert calls[0] == calls[1]
|
|
assert conn.execute("SELECT status FROM device_bindings").fetchone()[0] == 1
|
|
assert (await service.overview(DEVICE, 7))["unbind"]["operation_id"] == op["operation_id"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_late_close_response_cannot_unbind_a_new_binding_instance(state, monkeypatch):
|
|
service, conn = state
|
|
async def changed(source, request_id, *, close=False):
|
|
conn.execute("UPDATE device_bindings SET binding_instance_id=?", (str(uuid4()),))
|
|
conn.commit()
|
|
return {"status": "unbound", "canUnbind": True, "relationState": "closing"}
|
|
monkeypatch.setattr("banban.service.joint_binding.binding_relation", changed)
|
|
op = await service.start(DEVICE, 7, "unbind")
|
|
with pytest.raises(TrafficError) as failure: await service.finish_unbind(op["operation_id"], 7)
|
|
assert failure.value.code == "STALE_OPERATION"
|
|
assert conn.execute("SELECT status FROM device_bindings").fetchone()[0] == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_consumed_token_unknown_or_released_relation_is_not_success(state, monkeypatch):
|
|
service, _ = state
|
|
async def consumed(*args): raise TrafficError("HANDOFF_TOKEN_USED", "used")
|
|
monkeypatch.setattr("banban.service.joint_binding.create_handoff", consumed)
|
|
op = await service.start(DEVICE, 7, "bind")
|
|
for status in ["unknown", "unbound", "never_bound"]:
|
|
service.test_relation["status"] = status
|
|
with pytest.raises(TrafficError): await service.prepare(op["operation_id"], 7)
|
|
assert (await service.get(op["operation_id"], 7))["status"] != "completed"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_successful_client_receipt_cannot_override_active_original_binding(state):
|
|
service, conn = state
|
|
op = await service.start(DEVICE, 7, "unbind")
|
|
await service.prepare(op["operation_id"], 7)
|
|
result = await confirm(service, op)
|
|
assert result["traffic_status"] == "pending"
|
|
with pytest.raises(TrafficError): await service.finish_unbind(op["operation_id"], 7)
|
|
assert conn.execute("SELECT status FROM device_bindings").fetchone()[0] == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_expired_token_rotates_only_application_key(state, monkeypatch):
|
|
service, _ = state
|
|
calls = []
|
|
async def expired(*args):
|
|
calls.append(args)
|
|
raise TrafficError("HANDOFF_TOKEN_EXPIRED", "expired")
|
|
monkeypatch.setattr("banban.service.joint_binding.create_handoff", expired)
|
|
op = await service.start(DEVICE, 7, "bind")
|
|
for _ in range(2):
|
|
with pytest.raises(TrafficError): await service.prepare(op["operation_id"], 7)
|
|
assert calls[0][:2] == calls[1][:2]
|
|
assert calls[0][2] != calls[1][2]
|
|
assert (await service.start(DEVICE, 7, "bind"))["operation_id"] == op["operation_id"]
|
|
|
|
|
|
def test_binding_generation_changes_only_for_new_ownership():
|
|
row = {"status": 1, "owner_user_id": 7, "binding_instance_id": str(uuid4())}
|
|
assert BindingDAO._binding_instance(row, 7) == row["binding_instance_id"]
|
|
assert BindingDAO._binding_instance(row, 8) != row["binding_instance_id"]
|
|
row["status"] = 0
|
|
assert BindingDAO._binding_instance(row, 7) != row["binding_instance_id"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_iccid_normalization_and_conflict(state):
|
|
service, conn = state
|
|
assert normalize_iccid(ICCID.lower()) == ICCID
|
|
for invalid in [None, 1234, "123456789012345", "x" * 23, "#" * 20]: assert normalize_iccid(invalid) is None
|
|
await service.record_iccid(DEVICE, ICCID.lower())
|
|
with pytest.raises(TrafficError): await service.record_iccid(DEVICE, "898604E91123D1334999")
|
|
assert conn.execute("SELECT iccid FROM device_settings").fetchone()[0] == ICCID
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mqtt_006_failure_and_iccid_failure(monkeypatch):
|
|
from handlers.mqtt_handler import TalkingQMQTTService
|
|
calls = []
|
|
logs = []
|
|
monkeypatch.setattr("handlers.mqtt_handler.logger.info", lambda *args: logs.append(args[-1]))
|
|
async def broken_iccid(*args): raise RuntimeError("database unavailable")
|
|
async def finalize(**kwargs):
|
|
calls.append(kwargs)
|
|
return None
|
|
monkeypatch.setattr("banban.service.joint_binding.joint_binding_service.record_iccid", broken_iccid)
|
|
monkeypatch.setattr("handlers.mqtt_handler.binding_service.finalize_nfc_bind", finalize)
|
|
mqtt = TalkingQMQTTService({"broker": "localhost"})
|
|
await mqtt._handle_bind_response(DEVICE, {"status": "failed", "data": {"uuid": "CARD"}})
|
|
await mqtt._handle_bind_response(DEVICE, {"status": "success", "data": None})
|
|
assert not calls
|
|
await mqtt._handle_bind_response(DEVICE, {"status": "success", "data": {"uuid": "CARD", "iccid": ICCID}})
|
|
await mqtt._handle_bind_response(DEVICE, {"status": "success", "data": {"uuid": "CARD"}})
|
|
assert len(calls) == 2
|
|
assert any("ICCID present=True, valid=True" in line for line in logs)
|
|
assert any("ICCID present=False, valid=False" in line for line in logs)
|
|
assert not any("ICCID persisted" in line for line in logs)
|
|
async def saved_iccid(*args): return None
|
|
monkeypatch.setattr("banban.service.joint_binding.joint_binding_service.record_iccid", saved_iccid)
|
|
await mqtt._handle_bind_response(DEVICE, {"status": "success", "data": {"uuid": "CARD", "iccid": ICCID}})
|
|
assert any("ICCID persisted" in line for line in logs)
|
|
assert not any(ICCID in line for line in logs)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("body", [None, {"code": "OK", "data": {}},
|
|
{"code": "OK", "data": {"targetAppId": "wrong", "handoffToken": "secret", "expiresIn": 300}},
|
|
{"code": "OK", "data": {"targetAppId": "wx56e1cfc52aab630f", "handoffToken": "secret", "expiresIn": True}}])
|
|
async def test_upstream_malformed_reply_is_not_success(monkeypatch, body):
|
|
real_client = httpx.AsyncClient
|
|
transport = httpx.MockTransport(lambda req: httpx.Response(200, json=body))
|
|
monkeypatch.setattr(traffic_handoff.settings, "banban_handoff_service_token", "test-service-token")
|
|
monkeypatch.setattr(traffic_handoff.httpx, "AsyncClient", lambda **kwargs: real_client(transport=transport, **kwargs))
|
|
with pytest.raises(TrafficError): await traffic_handoff.create_handoff(ICCID, "banban:test", "test-request")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_router_timeout_returns_retryable_error(monkeypatch):
|
|
from banban.routers.joint_binding import call
|
|
from fastapi import HTTPException
|
|
real_wait_for = asyncio.wait_for
|
|
async def short_wait(coro, timeout):
|
|
assert timeout == 8
|
|
return await real_wait_for(coro, timeout=0.001)
|
|
monkeypatch.setattr(asyncio, "wait_for", short_wait)
|
|
with pytest.raises(HTTPException) as failure: await call(asyncio.sleep(10))
|
|
assert failure.value.status_code == 503
|
|
assert failure.value.detail["code"] == "OPERATION_TIMEOUT"
|