-- Self-test for the conformance runner.
--
-- The Google conformance suite is an external binary
-- (`conformance_test_runner`) we can't reasonably bundle here, so this
-- test stands in for it: drives our runner with crafted
-- `ConformanceRequest` cases and asserts well-formed
-- `ConformanceResponse` output.
--
-- Two layers:
--   1. `core` group — calls `cmd.conformance.core.handle_request` directly.
--      Covers every dispatch arm without paying subprocess cost.
--   2. `subprocess` group — actually pipes framed bytes through
--      `tarantool cmd/conformance-runner.lua`. Covers the length-prefixed
--      framing and the read-until-EOF loop.

local t           = require('luatest')
local fio         = require('fio')
local core        = require('cmd.conformance.core')
local conformance = require('full.conformance.conformance_pb')
local proto3      = require('full.protobuf_test_messages.proto3.test_messages_proto3_pb')

local PROTOBUF = conformance.WireFormat.PROTOBUF
local JSON     = conformance.WireFormat.JSON
local TEXT     = conformance.WireFormat.TEXT_FORMAT

local PROTO3_NAME = 'protobuf_test_messages.proto3.TestAllTypesProto3'

local function encode_req(t_)
    return conformance.ConformanceRequest_encode(t_)
end

local function decode_resp(bytes)
    return conformance.ConformanceResponse_decode(bytes)
end

-- ---------------------------------------------------------------------------
-- 1. Direct dispatch (no subprocess)
-- ---------------------------------------------------------------------------

local core_g = t.group('conformance.core')

core_g.test_failureset_preflight = function()
    -- The conformance runner asks for a FailureSet up front. Empty payload,
    -- empty FailureSet response is the canonical answer.
    local req = encode_req({
        protobuf_payload = '',
        requested_output_format = PROTOBUF,
        message_type = 'conformance.FailureSet',
    })
    local resp = decode_resp(core.handle_request(req))
    t.assert_equals(resp.protobuf_payload, '')
    t.assert_equals(resp.skipped, nil)
    t.assert_equals(resp.parse_error, nil)
    t.assert_equals(resp.runtime_error, nil)
end

core_g.test_pb_to_pb_roundtrip = function()
    -- Encode a TestAllTypesProto3 ourselves, ask the runner to round-trip
    -- it through pb->pb, assert byte-identical output. Since our encoder
    -- is deterministic for non-map fields, the output bytes must match
    -- input bytes exactly.
    local input = proto3.TestAllTypesProto3_encode({
        optional_int32  = 42,
        optional_string = 'hello',
        repeated_int32  = {1, 2, 3, 4},
    })
    local resp = decode_resp(core.handle_request(encode_req({
        protobuf_payload = input,
        requested_output_format = PROTOBUF,
        message_type = PROTO3_NAME,
    })))
    t.assert_equals(resp.protobuf_payload, input)
end

core_g.test_pb_to_json = function()
    local input = proto3.TestAllTypesProto3_encode({
        optional_int32 = 7,
        optional_string = 'world',
    })
    local resp = decode_resp(core.handle_request(encode_req({
        protobuf_payload = input,
        requested_output_format = JSON,
        message_type = PROTO3_NAME,
    })))
    t.assert_str_contains(resp.json_payload, '"optionalInt32":7')
    t.assert_str_contains(resp.json_payload, '"optionalString":"world"')
end

core_g.test_json_to_pb = function()
    local resp = decode_resp(core.handle_request(encode_req({
        json_payload = [[{"optionalInt32": 9, "optionalString": "abc"}]],
        requested_output_format = PROTOBUF,
        message_type = PROTO3_NAME,
    })))
    t.assert_not(resp.parse_error, resp.parse_error)
    t.assert_not(resp.runtime_error, resp.runtime_error)
    local decoded = proto3.TestAllTypesProto3_decode(resp.protobuf_payload)
    t.assert_equals(decoded.optional_int32, 9)
    t.assert_equals(decoded.optional_string, 'abc')
end

core_g.test_parse_error_on_malformed_protobuf = function()
    -- A truncated varint should produce a parse_error, not a crash.
    local resp = decode_resp(core.handle_request(encode_req({
        protobuf_payload = '\x08',  -- tag(1, VARINT), no value
        requested_output_format = PROTOBUF,
        message_type = PROTO3_NAME,
    })))
    t.assert_not_equals(resp.parse_error, nil)
end

core_g.test_parse_error_on_malformed_json = function()
    local resp = decode_resp(core.handle_request(encode_req({
        json_payload = '{not valid json',
        requested_output_format = PROTOBUF,
        message_type = PROTO3_NAME,
    })))
    t.assert_not_equals(resp.parse_error, nil)
end

core_g.test_unsupported_message_type_skipped = function()
    -- proto2 / editions test message types are intentionally unsupported.
    local resp = decode_resp(core.handle_request(encode_req({
        protobuf_payload = '',
        requested_output_format = PROTOBUF,
        message_type = 'protobuf_test_messages.proto2.TestAllTypesProto2',
    })))
    t.assert_str_contains(resp.skipped or '', 'unsupported message type')
end

core_g.test_text_format_skipped = function()
    local input = proto3.TestAllTypesProto3_encode({optional_int32 = 1})
    local resp = decode_resp(core.handle_request(encode_req({
        protobuf_payload = input,
        requested_output_format = TEXT,
        message_type = PROTO3_NAME,
    })))
    t.assert_str_contains(resp.skipped or '', 'jspb/text')
end

core_g.test_empty_payload_decodes_as_empty_message = function()
    -- proto3 says empty bytes is a valid empty message.
    local resp = decode_resp(core.handle_request(encode_req({
        protobuf_payload = '',
        requested_output_format = PROTOBUF,
        message_type = PROTO3_NAME,
    })))
    t.assert_equals(resp.protobuf_payload, '')
end

core_g.test_wkt_timestamp_field_roundtrip = function()
    -- Exercise the WKT path: TestAllTypesProto3.optional_timestamp.
    -- We rely on our WKT Timestamp encoder/decoder.
    local input = proto3.TestAllTypesProto3_encode({
        optional_timestamp = require('datetime').new({timestamp = 1700000000}),
    })
    local resp = decode_resp(core.handle_request(encode_req({
        protobuf_payload = input,
        requested_output_format = PROTOBUF,
        message_type = PROTO3_NAME,
    })))
    t.assert_equals(resp.protobuf_payload, input)
end

-- ---------------------------------------------------------------------------
-- 2. Subprocess: stdin/stdout framing
-- ---------------------------------------------------------------------------
--
-- These tests verify only the framing wrapper in cmd/conformance-runner.lua;
-- the dispatch logic is fully covered by the `core` group above.

local sub_g = t.group('conformance.subprocess')

local REPO_ROOT = fio.abspath(fio.pathjoin(
    fio.dirname(debug.getinfo(1, 'S').source:sub(2)), '..'))
local RUNNER = fio.pathjoin(REPO_ROOT, 'cmd', 'conformance-runner.lua')

local function le32(n)
    return string.char(
        n % 256,
        math.floor(n / 256) % 256,
        math.floor(n / 65536) % 256,
        math.floor(n / 16777216) % 256)
end

local function read_le32(s, off)
    local b1, b2, b3, b4 = s:byte(off, off + 3)
    return b1 + b2 * 256 + b3 * 65536 + b4 * 16777216
end

-- Run the runner with `input_bytes` piped to stdin, return the raw stdout.
local function run_runner(input_bytes)
    local in_path  = os.tmpname()
    local out_path = os.tmpname()
    local err_path = os.tmpname()
    local fin = assert(io.open(in_path, 'wb'))
    fin:write(input_bytes); fin:close()

    local lua_path = os.getenv('LUA_PATH') or ''
    local cmd = string.format(
        'cd %q && LUA_PATH=%q tarantool cmd/conformance-runner.lua < %q > %q 2> %q',
        REPO_ROOT, lua_path, in_path, out_path, err_path)
    local rc = os.execute(cmd)

    local fout = assert(io.open(out_path, 'rb'))
    local out  = fout:read('*a'); fout:close()
    local ferr = io.open(err_path, 'r')
    local err  = ferr and ferr:read('*a') or ''
    if ferr then ferr:close() end
    os.remove(in_path); os.remove(out_path); os.remove(err_path)
    return rc, out, err
end

-- Parse a stream of length-prefixed responses out of `bytes`.
local function parse_framed(bytes)
    local out, off = {}, 1
    while off + 4 <= #bytes + 1 do
        local n = read_le32(bytes, off)
        off = off + 4
        if off + n - 1 > #bytes then break end
        out[#out + 1] = bytes:sub(off, off + n - 1)
        off = off + n
    end
    return out, off
end

sub_g.test_single_request_round_trip = function()
    local req = encode_req({
        protobuf_payload = '',
        requested_output_format = PROTOBUF,
        message_type = 'conformance.FailureSet',
    })
    local framed = le32(#req) .. req
    local rc, out, err = run_runner(framed)
    t.assert_equals(rc, 0, 'runner exited non-zero, stderr=' .. err)
    local resps = parse_framed(out)
    t.assert_equals(#resps, 1, 'expected 1 framed response, stderr=' .. err)
    local resp = decode_resp(resps[1])
    t.assert_equals(resp.protobuf_payload, '')
end

sub_g.test_multiple_requests_in_one_session = function()
    -- The conformance runner sends many requests over a single pipe; the
    -- script must loop until EOF rather than handle one and exit.
    local req1 = encode_req({
        protobuf_payload = '',
        requested_output_format = PROTOBUF,
        message_type = 'conformance.FailureSet',
    })
    local payload = proto3.TestAllTypesProto3_encode({optional_int32 = 17})
    local req2 = encode_req({
        protobuf_payload = payload,
        requested_output_format = PROTOBUF,
        message_type = PROTO3_NAME,
    })
    local req3 = encode_req({
        json_payload = [[{"optionalInt32": 99}]],
        requested_output_format = JSON,
        message_type = PROTO3_NAME,
    })

    local framed = le32(#req1) .. req1
                .. le32(#req2) .. req2
                .. le32(#req3) .. req3
    local rc, out, err = run_runner(framed)
    t.assert_equals(rc, 0, 'runner exited non-zero, stderr=' .. err)
    local resps = parse_framed(out)
    t.assert_equals(#resps, 3, 'expected 3 framed responses, stderr=' .. err)

    local r1 = decode_resp(resps[1])
    local r2 = decode_resp(resps[2])
    local r3 = decode_resp(resps[3])
    t.assert_equals(r1.protobuf_payload, '')
    t.assert_equals(r2.protobuf_payload, payload)
    t.assert_str_contains(r3.json_payload, '"optionalInt32":99')
end

sub_g.test_empty_stdin_clean_exit = function()
    local rc, out, err = run_runner('')
    t.assert_equals(rc, 0, 'runner exited non-zero, stderr=' .. err)
    t.assert_equals(out, '', 'unexpected output on empty stdin')
end