water/server/tests/water_device_ingress_test.php

<?php
/** 预留设备入口测试：真实数据库事务回滚，不启用网关、不连接 Broker。 */
require dirname(__DIR__).'/vendor/autoload.php';
\Dotenv\Dotenv::createUnsafeMutable(dirname(__DIR__))->load();
\support\App::loadAllConfig(['route']);
date_default_timezone_set('Asia/Shanghai');
use app\service\water\{DeviceIngress,DeviceReport,DeviceProtocolAdapter};
use plugin\saiadmin\app\service\WaterBusiness;
use plugin\saiadmin\exception\ApiException;
use support\Db;
use plugin\saiadmin\app\model\water\WaterModel;

class TestWaterProtocol implements DeviceProtocolAdapter
{
    public function fromHttp(string $body,array $headers): ?array { return json_decode($body,true,512,JSON_THROW_ON_ERROR); }
    public function fromMqtt(string $topic,string $payload,array $properties): ?array { return $topic === 'test/water' ? $this->fromHttp($payload,[]) : null; }
}
$checks = 0;
function verify(bool $ok,string $message): void { global $checks; ++$checks; if (!$ok) throw new RuntimeException($message); }
function rejected(callable $fn,int $status): void { try { $fn(); } catch (ApiException $e) { verify($e->getCode()===$status,'Expected status '.$status); return; } throw new RuntimeException('Expected rejection'); }
$disabled = new DeviceIngress([]);
rejected(fn()=>$disabled->http('{}','',[]),503);
$disabled->mqtt(['topic'=>'test/water','payload'=>'invalid']);
verify(!config('water_device.http_enabled') && !config('water_device.mqtt_enabled') && config('water_device.adapter') === null,'Local device integrations stay disabled');
$response = (new \app\controller\WaterDeviceController())->report(new \support\Request("POST /iot/water/report HTTP/1.1\r\nHost: localhost\r\nContent-Length: 2\r\n\r\n{}"));
verify($response->getStatusCode()===503,'Reserved HTTP controller returns real 503');
\Webman\Event\BootStrap::start(null);
verify(count(\Webman\Event\Event::getListeners('mqtt.message')) > 0,'MQTT event extension is registered');
$token = str_repeat('test-only-',4);
$settings = ['http_enabled'=>true,'mqtt_enabled'=>true,'http_token'=>$token,'adapter'=>TestWaterProtocol::class,'max_payload_bytes'=>1024];
$ingress = new DeviceIngress($settings);
rejected(fn()=>$ingress->http('{}','Bearer wrong',[]),401);
rejected(fn()=>(new DeviceIngress(array_replace($settings,['http_token'=>''])))->http('{}','Bearer ',[]),503);
rejected(fn()=>(new DeviceIngress(array_replace($settings,['adapter'=>null])))->http('{}','Bearer '.$token,[]),503);
rejected(fn()=>$ingress->http(str_repeat('x',1025),'Bearer '.$token,[]),413);
$before = \plugin\saiadmin\app\model\water\WaterDevice::query()->count();
$service = new WaterBusiness();
$reporter = new DeviceReport();
$tag = 'INGRESS-TEST-'.bin2hex(random_bytes(5));
Db::beginTransaction();
try {
    $enterprise = $service->save('enterprise',['name'=>$tag,'contact'=>'测试','phone'=>'TEST','address'=>'回滚测试地址'],1)['id'];
    $sku = $service->save('sku',['name'=>$tag,'liters'=>18.9],1)['id'];
    $device = $service->save('device',['name'=>$tag,'sn'=>$tag,'enterprise_id'=>$enterprise,'sku_id'=>$sku],1)['id'];
    $get = fn()=>\plugin\saiadmin\app\model\water\WaterDevice::query()->where('id',$device)->first()?->toArray();
    $at = date('Y-m-d H:i:s',time()-10);
    $heartbeat = ['kind'=>'heartbeat','sn'=>$tag,'reported_at'=>$at];
    $result = $ingress->http(json_encode($heartbeat),'Bearer '.$token,[]);
    verify($result['updated'] && $get()['last_seen']===$at,'Authenticated normalized heartbeat updates last_seen');
    verify(!$reporter->accept($heartbeat)['updated'],'Repeated heartbeat does not refresh last_seen');
    verify(!$reporter->accept(array_replace($heartbeat,['reported_at'=>date('Y-m-d H:i:s',time()-20)]))['updated'],'Out-of-order heartbeat cannot move last_seen backwards');
    verify($service->listing('devices',['keyword'=>$tag])['data'][0]['connection_status']==='online','Heartbeat drives existing online display');
    foreach ([date('Y-m-d H:i:s',time()-600),date('Y-m-d H:i:s',time()+600),'2026-02-30 00:00:00'] as $invalid) rejected(fn()=>$reporter->accept(array_replace($heartbeat,['reported_at'=>$invalid])),422);
    rejected(fn()=>$reporter->accept(array_replace($heartbeat,['sn'=>$tag.'-missing'])),404);
    rejected(fn()=>$reporter->accept(array_replace($heartbeat,['kind'=>'exchange'])),422);
    $usage = array_replace($heartbeat,['kind'=>'usage','today_liters'=>12.5]);
    $ingress->mqtt(['topic'=>'test/water','payload'=>json_encode($usage)]);
    verify((float)$get()['today_liters']===12.5 && $get()['telemetry_date']===date('Y-m-d'),'MQTT adapter feeds same cumulative usage handler');
    verify(!$reporter->accept($usage)['updated'],'Duplicate snapshot is not summed');
    verify(!$reporter->accept(array_replace($usage,['today_liters'=>8]))['updated'],'Lower out-of-order snapshot is ignored');
    $reporter->accept(array_replace($usage,['today_liters'=>15]));
    verify((float)$get()['today_liters']===15.0,'Larger snapshot replaces current total');
    foreach ([-1,100000000,'bad',null] as $invalid) rejected(fn()=>$reporter->accept(array_replace($usage,['today_liters'=>$invalid])),422);
    \plugin\saiadmin\app\model\water\WaterDevice::query()->where('id',$device)->update(['telemetry_date'=>date('Y-m-d',strtotime('-1 day')),'today_liters'=>99]);
    $reporter->accept(array_replace($usage,['today_liters'=>0]));
    verify((float)$get()['today_liters']===0.0,'New day begins from new cumulative snapshot');
    $snapshot = $get();
    $ingress->mqtt(['topic'=>'unrelated/topic','payload'=>json_encode($usage)]);
    verify($get()===$snapshot,'Unrelated MQTT topic ignored');
    verify(!\plugin\saiadmin\app\model\water\WaterAccount::query()->where('enterprise_id',$enterprise)->count() && (int)\plugin\saiadmin\app\model\water\WaterSku::query()->where('id',$sku)->value('available')===0,'Telemetry never changes stock or entitlement');
    \plugin\saiadmin\app\model\water\WaterEnterprise::query()->where('id',$enterprise)->update(['status'=>2]);
    rejected(fn()=>$reporter->accept($heartbeat),422);
} finally { Db::rollback(); }
verify(\plugin\saiadmin\app\model\water\WaterDevice::query()->count()===$before,'Test devices rolled back');
echo "PASS: $checks device ingress checks; all database fixtures rolled back.\n";