local CONSTANTS = require("spec.internal.constants")
local function tcp_server(port, opts)
local threads = require "llthreads2.ex"
opts = opts or {}
if CONSTANTS.TEST_COVERAGE_MODE == "true" then
opts.timeout = CONSTANTS.TEST_COVERAGE_TIMEOUT
end
local thread = threads.new({
function(port, opts)
local socket = require "socket"
local server = assert(socket.tcp())
server:settimeout(opts.timeout or 60)
assert(server:setoption("reuseaddr", true))
assert(server:bind("*", port))
assert(server:listen())
local line
local oks, fails = 0, 0
local handshake_done = false
local n = opts.requests or 1
for _ = 1, n + 1 do
local client, err
if opts.timeout then
client, err = server:accept()
if err == "timeout" then
line = "timeout"
break
else
assert(client, err)
end
else
client = assert(server:accept())
end
if opts.tls and handshake_done then
local ssl = require "spec.internal.ssl"
local params = {
mode = "server",
protocol = "any",
key = "spec/fixtures/kong_spec.key",
certificate = "spec/fixtures/kong_spec.crt",
}
client = ssl.wrap(client, params)
client:dohandshake()
end
line, err = client:receive()
if err == "closed" then
fails = fails + 1
else
if not handshake_done then
assert(line == "\\START")
client:send("\\OK\n")
handshake_done = true
else
if line == "@DIE@" then
client:send(string.format("%d:%d\n", oks, fails))
client:close()
break
end
oks = oks + 1
client:send((opts.prefix or "") .. line .. "\n")
end
client:close()
end
end
server:close()
return line
end
}, port, opts)
local thr = thread:start()
ngx.sleep(0.001)
local sock = ngx.socket.tcp()
sock:settimeout(0.01)
while true do
if sock:connect("localhost", port) then
sock:send("\\START\n")
local ok = sock:receive()
sock:close()
if ok == "\\OK" then
break
end
end
end
sock:close()
return thr
end
local function kill_tcp_server(port)
local sock = ngx.socket.tcp()
assert(sock:connect("localhost", port))
assert(sock:send("@DIE@\n"))
local str = assert(sock:receive())
assert(sock:close())
local oks, fails = str:match("(%d+):(%d+)")
return tonumber(oks), tonumber(fails)
end
local code_status = {
[200] = "OK",
[201] = "Created",
[202] = "Accepted",
[203] = "Non-Authoritative Information",
[204] = "No Content",
[205] = "Reset Content",
[206] = "Partial Content",
[207] = "Multi-Status",
[300] = "Multiple Choices",
[301] = "Moved Permanently",
[302] = "Found",
[303] = "See Other",
[304] = "Not Modified",
[305] = "Use Proxy",
[307] = "Temporary Redirect",
[308] = "Permanent Redirect",
[400] = "Bad Request",
[401] = "Unauthorized",
[402] = "Payment Required",
[403] = "Forbidden",
[404] = "Not Found",
[405] = "Method Not Allowed",
[406] = "Not Acceptable",
[407] = "Proxy Authentication Required",
[408] = "Request Timeout",
[409] = "Conflict",
[410] = "Gone",
[411] = "Length Required",
[412] = "Precondition Failed",
[413] = "Payload Too Large",
[414] = "URI Too Long",
[415] = "Unsupported Media Type",
[416] = "Range Not Satisfiable",
[417] = "Expectation Failed",
[418] = "I'm a teapot",
[422] = "Unprocessable Entity",
[423] = "Locked",
[424] = "Failed Dependency",
[426] = "Upgrade Required",
[428] = "Precondition Required",
[429] = "Too Many Requests",
[431] = "Request Header Fields Too Large",
[451] = "Unavailable For Legal Reasons",
[500] = "Internal Server Error",
[501] = "Not Implemented",
[502] = "Bad Gateway",
[503] = "Service Unavailable",
[504] = "Gateway Timeout",
[505] = "HTTP Version Not Supported",
[506] = "Variant Also Negotiates",
[507] = "Insufficient Storage",
[508] = "Loop Detected",
[510] = "Not Extended",
[511] = "Network Authentication Required",
}
local EMPTY = {}
local function handle_response(code, body, headers)
if not code then
code = 500
body = ""
headers = EMPTY
end
local head_str = ""
for k, v in pairs(headers or EMPTY) do
head_str = head_str .. k .. ": " .. v .. "\r\n"
end
return code .. " " .. code_status[code] .. " HTTP/1.1" .. "\r\n" ..
"Content-Length: " .. #body .. "\r\n" ..
"Connection: close\r\n" ..
head_str ..
"\r\n" ..
body
end
local function handle_request(client, response)
local lines = {}
local headers = {}
local line, err
local content_length
repeat
line, err = client:receive("*l")
if err then
return nil, err
else
local k, v = line:match("^([^:]+):%s*(.+)$")
if k then
headers[k] = v
if k:lower() == "content-length" then
content_length = tonumber(v)
end
end
table.insert(lines, line)
end
until line == ""
local method = lines[1]:match("^(%S+)%s+(%S+)%s+(%S+)$")
local method_lower = method:lower()
local body
if content_length then
body = client:receive(content_length)
elseif method_lower == "put" or method_lower == "post" then
body = client:receive("*a")
end
local response_str
local meta = getmetatable(response)
if type(response) == "function" or (meta and meta.__call) then
response_str = response(lines, body, headers)
elseif type(response) == "table" and response.code then
response_str = handle_response(response.code, response.body, response.headers)
elseif type(response) == "table" and response[1] then
response_str = handle_response(response[1], response[2], response[3])
elseif type(response) == "string" then
response_str = response
elseif response == nil then
response_str = "HTTP/1.1 200 OK\r\nConnection: close\r\n\r\n"
end
client:send(response_str)
return lines, body, headers
end
local function http_mock(port, opts)
local socket = require "socket"
local server = assert(socket.tcp())
if CONSTANTS.TEST_COVERAGE_MODE == "true" then
opts.timeout = CONSTANTS.TEST_COVERAGE_TIMEOUT
end
server:settimeout(opts and opts.timeout or 60)
assert(server:setoption('reuseaddr', true))
assert(server:bind("*", port))
assert(server:listen())
return coroutine.wrap(function(response, exit)
local lines, body, headers
while not exit do
local client, err = server:accept()
if err then
lines, body = false, err
else
lines, body, headers = handle_request(client, response)
client:close()
end
response, exit = coroutine.yield(lines, body, headers)
end
server:close()
return true
end)
end
local function udp_server(port, n, timeout)
local threads = require "llthreads2.ex"
if CONSTANTS.TEST_COVERAGE_MODE == "true" then
timeout = CONSTANTS.TEST_COVERAGE_TIMEOUT
end
local thread = threads.new({
function(port, n, timeout)
local socket = require "socket"
local server = assert(socket.udp())
server:settimeout(timeout or 360)
server:setoption("reuseaddr", true)
server:setsockname("127.0.0.1", port)
local err
local data = {}
local handshake_done = false
local i = 0
while i < n do
local pkt, rport
pkt, err, rport = server:receivefrom()
if not pkt then
break
end
if pkt == "KONG_UDP_HELLO" then
if not handshake_done then
handshake_done = true
server:sendto("KONG_UDP_READY", "127.0.0.1", rport)
end
else
i = i + 1
data[i] = pkt
err = nil
end
end
server:close()
return (n > 1 and data or data[1]), err
end
}, port or CONSTANTS.MOCK_UPSTREAM_PORT, n or 1, timeout)
thread:start()
local socket = require "socket"
local handshake = socket.udp()
handshake:settimeout(0.01)
handshake:setsockname("127.0.0.1", 0)
while true do
handshake:sendto("KONG_UDP_HELLO", "127.0.0.1", port)
local data = handshake:receive()
if data == "KONG_UDP_READY" then
break
end
end
handshake:close()
return thread
end
local is_echo_server_ready, get_echo_server_received_data, echo_server_reset
do
local shell = require("spec.internal.shell")
local cmd = require("spec.internal.cmd")
local msg_id = -1
local prefix_dir = "servroot"
function is_echo_server_ready()
local sock = ngx.socket.tcp()
sock:settimeout(0.1)
local retry = 0
local test_port = 8188
while true do
if sock:connect("localhost", test_port) then
sock:send("START\n")
local ok = sock:receive()
sock:close()
if ok == "START" then
return true
end
else
retry = retry + 1
if retry > 10 then
return false
end
end
end
end
function get_echo_server_received_data(expected, timeout)
if timeout == nil then
timeout = 0.5
end
local extract_cmd = "grep content_by_lua "..prefix_dir.."/logs/error.log | tail -1"
local _, _, log = assert(shell.exec(extract_cmd))
local pattern = "%*(%d+)%s.*received data: (.*)"
local cur_msg_id, data = string.match(log, pattern)
local t = 0.1
local time_acc = 0
while string.find(data, expected) == nil or cur_msg_id == msg_id do
ngx.sleep(t)
time_acc = time_acc + t
if time_acc >= timeout then
return "timeout"
end
_, _, log = assert(shell.exec(extract_cmd))
cur_msg_id, data = string.match(log, pattern)
end
msg_id = cur_msg_id
return data
end
function echo_server_reset()
cmd.stop_kong(prefix_dir)
msg_id = -1
end
end
return {
tcp_server = tcp_server,
kill_tcp_server = kill_tcp_server,
http_mock = http_mock,
udp_server = udp_server,
is_echo_server_ready = is_echo_server_ready,
echo_server_reset = echo_server_reset,
get_echo_server_received_data = get_echo_server_received_data,
}