10个设备同时崩溃，你只测了1台：物联网测试的致命陷阱_千锋教育

import paho.mqtt.client as mqtt
import threading
import time

def simulate_device(device_id):
    client = mqtt.Client(f"device_{device_id}")
    client.connect("iot-broker.com", 1883, 60)
    for i in range(10):
        client.publish(f"device/{device_id}/data", f"temp={i}")
        time.sleep(0.1)
    client.disconnect()

# 并发100个设备
threads = []
for i in range(100):
    t = threading.Thread(target=simulate_device, args=(i,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()
print("100设备并发测试完成")

import asyncio
import aiohttp
import paho.mqtt.client as mqtt
from aiocoap import *

async def coap_request():
    protocol = await Context.create_client_context()
    request = Message(code=GET, uri="coap://iot-server/temperature")
    response = await protocol.request(request).response
    return response.payload

def mqtt_publish():
    client = mqtt.Client("tester")
    client.connect("iot-broker.com", 1883, 60)
    client.publish("sensor/temp", "25.5")
    client.disconnect()

async def http_request():
    async with aiohttp.ClientSession() as session:
        async with session.get("http://iot-server/api/status") as resp:
            return await resp.text()

# 并发测试
async def main():
    tasks = [
        coap_request(),
        http_request(),
        asyncio.to_thread(mqtt_publish)
    ]
    results = await asyncio.gather(*tasks)
    print(f"CoAP结果: {results[0]}")
    print(f"HTTP结果: {results[1]}")
    print(f"MQTT结果: {results[2]}")

asyncio.run(main())

import time
import threading

class DeviceState:
    def __init__(self, device_id):
        self.device_id = device_id
        self.state = "offline"
        self.version = 0
        self.lock = threading.Lock()

    def update_state(self, new_state):
        with self.lock:
            if new_state != self.state:
                self.state = new_state
                self.version += 1
                print(f"设备{self.device_id} 状态更新: {new_state} (版本{self.version})")

    def sync_with_cloud(self, cloud_state, cloud_version):
        with self.lock:
            if cloud_version > self.version:
                self.state = cloud_state
                self.version = cloud_version
                print(f"设备{self.device_id} 同步云端状态: {cloud_state}")

# 测试状态同步
device = DeviceState("device_001")
device.update_state("online")
device.sync_with_cloud("offline", 2)  # 云端版本更高，强制同步

class EdgeProcessor:
    def __init__(self, algorithm_version="v2.0"):
        self.version = algorithm_version

    def process_local(self, data):
        # 本地处理逻辑
        return f"本地处理(v{self.version}): {data}"

class CloudProcessor:
    def __init__(self, algorithm_version="v2.0"):
        self.version = algorithm_version

    def process_cloud(self, data):
        # 云端处理逻辑，必须与本地一致
        return f"云端处理(v{self.version}): {data}"

# 一致性测试
edge = EdgeProcessor("v2.0")
cloud = CloudProcessor("v2.0")

test_data = "温度:25.5"
result_local = edge.process_local(test_data)
result_cloud = cloud.process_cloud(test_data)

assert result_local == result_cloud, f"一致性失败: {result_local} != {result_cloud}"
print("边缘计算一致性测试通过")

import paho.mqtt.client as mqtt
import ssl

# 安全连接测试
def test_secure_mqtt():
    client = mqtt.Client("secure_tester")
    client.tls_set(ca_certs="ca.crt", 
                   certfile="client.crt", 
                   keyfile="client.key",
                   tls_version=ssl.PROTOCOL_TLSv1_2)
    client.connect("iot-broker.com", 8883, 60)
    client.publish("sensor/temp", "25.5", qos=2)
    print("安全MQTT连接测试通过")

# 不安全连接测试（应被拒绝）
def test_insecure_mqtt():
    client = mqtt.Client("insecure_tester")
    try:
        client.connect("iot-broker.com", 1883, 60)
        client.publish("sensor/temp", "25.5", qos=2)
        print("不安全MQTT连接测试通过（异常）")
    except Exception as e:
        print(f"不安全连接被拒绝：{e}")

test_secure_mqtt()
test_insecure_mqtt()