57 lines
5.4 KiB
TypeScript
57 lines
5.4 KiB
TypeScript
import assert from "node:assert/strict"
|
|
import { createServer } from "node:http"
|
|
import Fastify from "fastify"
|
|
import { pool, closePool } from "../db/index.js"
|
|
import { initDatabase } from "../db/bootstrap.js"
|
|
import trafficRoutes from "../routes/traffic.js"
|
|
|
|
const database=process.env.MMAPP_OWN_SQL_DB
|
|
if(!database) console.log("traffic live SQL: skipped; explicit owned database required")
|
|
else {
|
|
assert(/^mm_accept_166_\d+$/.test(database));assert.equal(new URL(process.env.DATABASE_URL!).pathname,"/"+database)
|
|
assert.equal((await pool.query("SELECT current_database() AS name")).rows[0].name,database)
|
|
const prefix="own-live179-"+Date.now()+"-"
|
|
let serverId:number|undefined,mode="zero",nativeRequests=0,cases=0,two=false
|
|
const point=(name:string)=>({name,"rx-bits-per-second":"0","tx-bits-per-second":"0"})
|
|
const http=createServer(async(req,res)=>{
|
|
nativeRequests++;res.setHeader("Content-Type","application/json")
|
|
if(req.method==="GET")return res.end(JSON.stringify((two?["ether1","ether2"]:["ether1"]).map(name=>({name,running:"true",disabled:"false"}))))
|
|
let data="";for await(const chunk of req)data+=chunk;const name=JSON.parse(data).interface
|
|
if(mode==="auth"||mode==="fallback-deny"&&name==="ether2"){res.statusCode=401;return res.end('{"message":"Unauthorized"}')}
|
|
if(mode.startsWith("fallback")&&name.includes(",")){res.statusCode=400;return res.end('{"message":"Unsupported joined interface"}')}
|
|
if(mode==="empty")return res.end("[]")
|
|
if(mode==="malformed")return res.end(JSON.stringify([{name,"rx-bits-per-second":"12bad","tx-bits-per-second":"0"}]))
|
|
if(mode==="partial")return res.end(JSON.stringify([point("ether1")]))
|
|
res.end(JSON.stringify(name.split(",").map(point)))
|
|
})
|
|
await new Promise<void>(resolve=>http.listen(0,"127.0.0.1",resolve));const address=http.address();assert(address&&typeof address==="object")
|
|
const app=Fastify();await app.register(trafficRoutes)
|
|
const streams=new Set<AbortController>()
|
|
async function open() {
|
|
const ac=new AbortController();streams.add(ac);const r=await fetch(app.listeningOrigin+`/traffic/servers/${serverId}/live`,{signal:ac.signal});assert.equal(r.status,200);assert(r.body)
|
|
const reader=r.body.getReader(),decoder=new TextDecoder();let buffer=""
|
|
return {ac, async next(){const timeout=setTimeout(()=>ac.abort(),8000);try{
|
|
for(;;){const end=buffer.indexOf("\n\n");if(end>=0){const block=buffer.slice(0,end);buffer=buffer.slice(end+2);const event=block.match(/^event: (.*)$/m)?.[1],data=block.match(/^data: (.*)$/m)?.[1];if(event&&data)return{event,data:JSON.parse(data)};continue}
|
|
const chunk=await reader.read();if(chunk.done)return null;buffer+=decoder.decode(chunk.value,{stream:true})}
|
|
}finally{clearTimeout(timeout)}},close(){ac.abort();streams.delete(ac)}}
|
|
}
|
|
try {
|
|
await initDatabase();await app.listen({host:"127.0.0.1",port:0})
|
|
serverId=Number((await pool.query("INSERT INTO servers(name,host,port,use_ssl,enabled,username,password,updated_at) VALUES($1,'127.0.0.1',$2,false,true,'fixture','fixture',$3) RETURNING id",[prefix+"server",address.port,new Date(Date.now()-3_600_000).toISOString()])).rows[0].id)
|
|
let stream=await open(),event=await stream.next();assert.equal(event?.event,"sample");assert.equal(event?.data.rxMbps,0);assert.equal(event?.data.timestampKind,"backend-receipt");stream.close();cases++
|
|
for(const kind of ["empty","malformed","partial"]){mode=kind;two=kind==="partial";stream=await open();event=await stream.next();assert.equal(event?.event,"error");stream.close();cases++}
|
|
mode="fallback";two=true;stream=await open();event=await stream.next();assert.equal(event?.event,"sample");assert.equal(event?.data.rxMbps,0);stream.close();cases++
|
|
mode="fallback-deny";stream=await open();event=await stream.next();assert.equal(event?.event,"error");assert.match(event?.data.error,/401/);assert.equal(await stream.next(),null);stream.close();cases++
|
|
mode="auth";two=false;stream=await open();event=await stream.next();assert.equal(event?.event,"error");assert.equal(await stream.next(),null);stream.close();cases++
|
|
mode="zero";stream=await open();await stream.next();const beforeDisable=nativeRequests
|
|
await pool.query("UPDATE servers SET enabled=false WHERE id=$1",[serverId]);event=await stream.next();assert.equal(event?.event,"error");assert.match(event?.data.error,/выключено/);assert.equal(await stream.next(),null);assert.equal(nativeRequests,beforeDisable);stream.close();cases++
|
|
const beforeInitial=nativeRequests;const disabled=await app.inject({method:"GET",url:`/traffic/servers/${serverId}/live`});assert.equal(disabled.statusCode,404);assert.equal(nativeRequests,beforeInitial);cases++
|
|
await pool.query("UPDATE servers SET enabled=true WHERE id=$1",[serverId]);stream=await open();await stream.next();const beforeConfig=nativeRequests
|
|
await pool.query("UPDATE servers SET updated_at=$1 WHERE id=$2",[new Date(Date.now()+1000).toISOString(),serverId]);event=await stream.next();assert.match(event?.data.error,/Конфигурация/);assert.equal(await stream.next(),null);assert.equal(nativeRequests,beforeConfig);stream.close();cases++
|
|
console.log(`traffic live owned SQL: ${cases} cases; real TCP SSE/HTTP/zero/empty/malformed/partial/fallback/auth/disabled/config passed`)
|
|
}finally{
|
|
for(const ac of streams)ac.abort();await app.close();await new Promise<void>(resolve=>http.close(()=>resolve()))
|
|
if(serverId!==undefined)await pool.query("DELETE FROM servers WHERE id=$1",[serverId]);await closePool()
|
|
}
|
|
}
|