docker_test_harness-cd2025/base.lua

local tests = {}
local tests_by_mod = {}

local TESTS_STATE_ENUM = {
    NOT_STARTED = "not_started",
    STARTED = "started",
    STARTED_PLAYERS = "started_players",
    DONE = "done"
}
local tests_state = TESTS_STATE_ENUM.NOT_STARTED

local tests_context = {}

local failed = 0

test_harness.set_context = function(mod, version_string)
    table.insert(tests_context, { mod = mod, version_string = version_string})
end

local register_test = function(mod, name, func, opts)
    local modnames = minetest.get_modnames()
    local is_mod = false
    for i=1,#modnames do
        if modnames[i]==mod then
            is_mod=true
            break
        end
    end
    assert(is_mod)
    assert(type(name) == "string")
    assert(func == nil or type(func) == "function")
    if not opts then
        opts = {}
    else
        opts = test_harness.table_copy(opts)
    end
    opts.mod = mod
    opts.name = name
    opts.func = func
    table.insert(tests, opts)
    local mod_test_list = tests_by_mod[mod] or {}
    tests_by_mod[mod] = mod_test_list
    table.insert(mod_test_list, opts)
end

test_harness.get_test_registrator = function(mod)
    return function(name, func, opts)
        register_test(mod, name, func, opts)
    end
end


---------------------
-- Helpers
---------------------
local vec = vector.new
local vecw = function(axis, n, base)
    local ret = vec(base)
    ret[axis] = n
    return ret
end
local pos2str = minetest.pos_to_string
local get_node = minetest.get_node
local set_node = minetest.set_node

local charset = "qwertyuiopasdfghjklzxcvbnmQWERTYUIOPASDFGHJKLZXCVBNM1234567890"
math.randomseed(os.clock())
function test_harness.randomString(length)
    local ret = {}
    local r
    for _ = 1, length do
        r = math.random(1, #charset)
        table.insert(ret, charset:sub(r, r))
    end
    return table.concat(ret)
end

function test_harness.table_copy(t)
    local t2 = {}
    for k, v in pairs(t) do
        if type(v) == "table" then
            t2[k] = test_harness.table_copy(v)
        else
            t2[k] = v
        end
    end
    return t2
end

--- Copies and modifies positions `pos1` and `pos2` so that each component of
-- `pos1` is less than or equal to the corresponding component of `pos2`.
-- Returns the new positions.
function test_harness.sort_pos(pos1, pos2)
    pos1 = vector.copy(pos1)
    pos2 = vector.copy(pos2)
    if pos1.x > pos2.x then
        pos2.x, pos1.x = pos1.x, pos2.x
    end
    if pos1.y > pos2.y then
        pos2.y, pos1.y = pos1.y, pos2.y
    end
    if pos1.z > pos2.z then
        pos2.z, pos1.z = pos1.z, pos2.z
    end
    return pos1, pos2
end

--- Determines the volume of the region defined by positions `pos1` and `pos2`.
-- @return The volume.
function test_harness.volume(pos1, pos2)
    local pos1, pos2 = test_harness.sort_pos(pos1, pos2)
    return (pos2.x - pos1.x + 1) *
        (pos2.y - pos1.y + 1) *
        (pos2.z - pos1.z + 1)
end

---------------------
-- Nodes
---------------------
local air = "air"
rawset(_G, "testnode1", "")
rawset(_G, "testnode2", "")
rawset(_G, "testnode3", "")
-- Loads nodenames to use for tests
local function init_nodes()
    testnode1 = minetest.registered_aliases["mapgen_stone"]
    testnode2 = minetest.registered_aliases["mapgen_dirt"]
    testnode3 = minetest.registered_aliases["mapgen_cobble"] or minetest.registered_aliases["mapgen_dirt_with_grass"]
    assert(testnode1 and testnode2 and testnode3)
end
-- Writes repeating pattern into given area
rawset(_G, "place_pattern", function(pos1, pos2, pattern)
    local pos = vec()
    local node = { name = "" }
    local i = 1
    for z = pos1.z, pos2.z do
        pos.z = z
        for y = pos1.y, pos2.y do
            pos.y = y
            for x = pos1.x, pos2.x do
                pos.x = x
                node.name = pattern[i]
                set_node(pos, node)
                i = i % #pattern + 1
            end
        end
    end
end)


---------------------
-- Area management
---------------------
assert(minetest.get_mapgen_setting("mg_name") == "singlenode")
rawset(_G, "area", {})
do
    local areamin, areamax
    local off
    local c_air = minetest.get_content_id(air)
    local vbuffer = {}
    -- Assign a new area for use, will emerge and then call ready()
    area.assign = function(min, max, ready)
        areamin = min
        areamax = max
        minetest.emerge_area(min, max, function(bpos, action, remaining)
            assert(action ~= minetest.EMERGE_ERRORED)
            if remaining > 0 then return end
            minetest.after(0, function()
                area.clear()
                ready()
            end)
        end)
    end
    -- Reset area contents and state
    area.clear = function()
        if off and vector.equals(off, vec(0, 0, 0)) then
            return
        end
        local vmanip = minetest.get_voxel_manip(areamin, areamax)
        local vpos1, vpos2 = vmanip:get_emerged_area()
        local vcount = (vpos2.x - vpos1.x + 1) * (vpos2.y - vpos1.y + 1) * (vpos2.z - vpos1.z + 1)
        if #vbuffer ~= vcount then
            vbuffer = {}
            for i = 1, vcount do
                vbuffer[i] = c_air
            end
        end
        vmanip:set_data(vbuffer)
        vmanip:write_to_map()
        off = vec(0, 0, 0)
    end
    -- Returns an usable area [pos1, pos2] that does not overlap previous ones
    area.get = function(sizex, sizey, sizez)
        local size
        if sizey == nil and sizez == nil then
            size = vec(sizex, sizex, sizex)
        else
            size = vec(sizex, sizey, sizez)
        end
        local pos1 = vector.add(areamin, off)
        local pos2 = vector.subtract(vector.add(pos1, size), 1)
        if pos2.x > areamax.x or pos2.y > areamax.y or pos2.z > areamax.z then
            error("Internal failure: out of space")
        end
        off = vector.add(off, size)
        return pos1, pos2
    end
    -- Returns an axis and count (= n) relative to the last-requested area that is unoccupied
    area.dir = function(n)
        local pos1 = vector.add(areamin, off)
        if pos1.x + n <= areamax.x then
            off.x = off.x + n
            return "x", n
        elseif pos1.x + n <= areamax.y then
            off.y = off.y + n
            return "y", n
        elseif pos1.z + n <= areamax.z then
            off.z = off.z + n
            return "z", n
        end
        error("Internal failure: out of space")
    end
    -- Returns [XYZ] margin (list of pos pairs) of n around last-requested area
    -- (may actually be larger but doesn't matter)
    area.margin = function(n)
        local pos1, pos2 = area.get(n)
        return {
            { vec(areamin.x, areamin.y, pos1.z), pos2 }, -- X/Y
            { vec(areamin.x, pos1.y, areamin.z), pos2 }, -- X/Z
            { vec(pos1.x, areamin.y, areamin.z), pos2 }, -- Y/Z
        }
    end
end
-- Split an existing area into two non-overlapping [pos1, half1], [half2, pos2] parts; returns half1, half2
area.split = function(pos1, pos2)
    local axis
    if pos2.x - pos1.x >= 1 then
        axis = "x"
    elseif pos2.y - pos1.y >= 1 then
        axis = "y"
    elseif pos2.z - pos1.z >= 1 then
        axis = "z"
    else
        error("Internal failure: area too small to split")
    end
    local hspan = math.floor((pos2[axis] - pos1[axis] + 1) / 2)
    local half1 = vecw(axis, pos1[axis] + hspan - 1, pos2)
    local half2 = vecw(axis, pos1[axis] + hspan, pos2)
    return half1, half2
end


---------------------
-- Checks
---------------------
rawset(_G, "check", {})
-- Check that all nodes in [pos1, pos2] are the node(s) specified
check.filled = function(pos1, pos2, nodes)
    if type(nodes) == "string" then
        nodes = { nodes }
    end
    local _, counts = minetest.find_nodes_in_area(pos1, pos2, nodes)
    local total = test_harness.volume(pos1, pos2)
    local sum = 0
    for _, n in pairs(counts) do
        sum = sum + n
    end
    if sum ~= total then
        error((total - sum) .. " " .. table.concat(nodes, ",") .. " nodes missing in " ..
            pos2str(pos1) .. " -> " .. pos2str(pos2))
    end
end
-- Check that none of the nodes in [pos1, pos2] are the node(s) specified
check.not_filled = function(pos1, pos2, nodes)
    if type(nodes) == "string" then
        nodes = { nodes }
    end
    local _, counts = minetest.find_nodes_in_area(pos1, pos2, nodes)
    for nodename, n in pairs(counts) do
        if n ~= 0 then
            error(counts[nodename] .. " " .. nodename .. " nodes found in " ..
                pos2str(pos1) .. " -> " .. pos2str(pos2))
        end
    end
end
-- Check that all of the areas are only made of node(s) specified
check.filled2 = function(list, nodes)
    for _, pos in ipairs(list) do
        check.filled(pos[1], pos[2], nodes)
    end
end
-- Check that none of the areas contain the node(s) specified
check.not_filled2 = function(list, nodes)
    for _, pos in ipairs(list) do
        check.not_filled(pos[1], pos[2], nodes)
    end
end
-- Checks presence of a repeating pattern in [pos1, po2] (cf. place_pattern)
check.pattern = function(pos1, pos2, pattern)
    local pos = vec()
    local i = 1
    for z = pos1.z, pos2.z do
        pos.z = z
        for y = pos1.y, pos2.y do
            pos.y = y
            for x = pos1.x, pos2.x do
                pos.x = x
                local node = get_node(pos)
                if node.name ~= pattern[i] then
                    error(pattern[i] .. " not found at " .. pos2str(pos) .. " (i=" .. i .. ")")
                end
                i = i % #pattern + 1
            end
        end
    end
end

local request_shutdown = function()
    if failed == 0 then
        io.close(io.open(minetest.get_worldpath() .. "/tests_ok", "w"))
    end
    print("Requesting server shutdown")
    minetest.request_shutdown()
end


local all_in_table = function(names, names_list)
    for _, n in ipairs(names) do
        if not names_list[n] then
            return false
        end
    end
    return true
end

local display_tests_summary = function()
    for mod, tests_list in pairs(tests_by_mod) do
        print(string.format("%#70s", mod))
        for _, test in ipairs(tests_list) do
            if test.func == nil then
                local s = ":"..test.mod..":---- " .. test.name .. " "
                print(s .. string.rep("-", 60 - #s))
            elseif test.result ~= nil then
                print(string.format(":%s:%-60s %s", test.mod,test.name, test.result.ok and "pass" or "FAIL"))
                if not test.result.ok then
                    print("   " .. test.result.err)
                end
            else
                print(string.format(":%s:%-60s %s", test.mod, test.name, "No result"))
            end
        end
    end
    print("Tests done, " .. failed .. " tests failed.")
end

test_harness.dump = function(o)
    if type(o) == 'table' then
        local s = '{ '
        for k, v in pairs(o) do
            if type(k) ~= 'number' then k = '"' .. k .. '"' end
            s = s .. '[' .. k .. '] = ' .. test_harness.dump(v) .. ','
        end
        return s .. '} '
    else
        return tostring(o)
    end
end

local save_players = function(players)
    local players_data = {}
    for _, p in ipairs(players) do
        local player_obj = nil
        local player_name = nil
        if type(p) == "string" then
            player_obj = minetest.get_player_by_name(p)
            player_name = p
        else
            player_obj = p
            player_name = p:get_player_name()
        end
        players_data[player_name] = {
            position = player_obj:get_pos(),
            privs = test_harness.table_copy(minetest.get_player_privs(player_name))
        }
    end
    return players_data
end

local restore_players = function(players_data)
    for player_name, data in pairs(players_data) do
        local player = minetest.get_player_by_name(player_name)
        player:set_pos(data.position)
        minetest.set_player_privs(player_name, data.privs)
    end
end




local set_tests_done = function()
    tests_state = TESTS_STATE_ENUM.DONE
    print("All tests done, " .. failed .. " tests failed.")
    display_tests_summary()

    if minetest.settings:get_bool("test_harness_stop_server", true) then
        request_shutdown()
    end
end

local get_connected_player_names = function()
    local connected_player_names = {}
    for _, p in ipairs(minetest.get_connected_players()) do
        connected_player_names[p:get_player_name()] = true
    end
    return connected_player_names
end

local run_player_tests = function(list_player_tests)
    if tests_state == TESTS_STATE_ENUM.DONE then
        return
    end
    local connected_player_names = get_connected_player_names()

    local test_run_callback = function()
        for _, test in ipairs(list_player_tests) do
            if test.func ~= nil and test.result == nil then return end
        end
        set_tests_done()
    end

    for _, test in ipairs(list_player_tests) do
        if tests_state == TESTS_STATE_ENUM.DONE then
            break
        end
        if not test.result and
          test.func and
          test.players and
          next(test.players) and
          all_in_table(test.players, connected_player_names)
        then
            local wanted = vec(56, 56, 56)
            for x = 0, math.floor(wanted.x / 16) do
                for y = 0, math.floor(wanted.y / 16) do
                    for z = 0, math.floor(wanted.z / 16) do
                        assert(minetest.forceload_block(vec(x * 16, y * 16, z * 16), true, -1))
                    end
                end
            end
            area.assign(vec(0, 0, 0), wanted, function()
                area.clear()
                local player_data = save_players(test.players)
                local ok, err = pcall(test.func)
                test.result = { ok = ok, err = err }
                print(string.format(":%s:%-60s %s", test.mod, test.name, ok and "pass" or "FAIL"))
                if not ok then
                    print("   " .. err)
                    failed = failed + 1
                    if minetest.settings:get_bool("test_harness_failfast", false) then
                        if minetest.settings:get_bool("test_harness_stop_server", true) then
                            request_shutdown()
                        else
                            set_tests_done()
                        end
                    end
                end
                restore_players(player_data)
                test_run_callback()
            end)
        end
    end
end


---------------------
-- Main function
---------------------
local run_tests = function()
    tests_state = TESTS_STATE_ENUM.STARTED
    local simple_tests = {}
    local players_tests = {}
    do
        local nb_tests = 0
        local current_title = {}
        for _, test in ipairs(tests) do
            if not test.func then
                nb_tests = nb_tests + 1
                table.insert(current_title, test)
            elseif test.players and next(test.players) then
                for _, t in ipairs(current_title) do
                    table.insert(players_tests, t)
                end
                current_title = {}
                table.insert(players_tests, test)
            else
                for _, t in ipairs(current_title) do
                    table.insert(simple_tests, t)
                end
                current_title = {}
                table.insert(simple_tests, test)
            end
        end
        local v = minetest.get_version()


        print("Running " .. nb_tests .. " tests for:")
        for _,context in ipairs(tests_context) do
            print("  - "..context.mod .." - "..context.version_string)
        end
        print("on " .. v.project .. " " .. (v.hash or v.string))
    end

    init_nodes()

    -- emerge area from (0,0,0) ~ (56,56,56) and keep it loaded
    -- Note: making this area smaller speeds up tests
    local wanted = vec(56, 56, 56)
    for x = 0, math.floor(wanted.x / 16) do
        for y = 0, math.floor(wanted.y / 16) do
            for z = 0, math.floor(wanted.z / 16) do
                assert(minetest.forceload_block(vec(x * 16, y * 16, z * 16), true, -1))
            end
        end
    end
    failed = 0
    area.assign(vec(0, 0, 0), wanted, function()
        for _, test in ipairs(simple_tests) do
            if not test.func then
                local s = ":"..test.mod..":---- " .. test.name .. " "
                print(s .. string.rep("-", 60 - #s))
                test.result = { ok = true, err = "" }
            else
                area.clear()
                local ok, err = pcall(test.func)
                test.result = { ok = ok, err = err }
                print(string.format(":%s:%-60s %s", test.mod, test.name, ok and "pass" or "FAIL"))
                if not ok then
                    print("   " .. err)
                    failed = failed + 1
                    if minetest.settings:get_bool("test_harness_failfast", false) then
                        tests_state = TESTS_STATE_ENUM.DONE
                        break
                    end
                end
            end
        end

        print("Server tests done, " .. failed .. " tests failed.")
        if minetest.settings:get_bool("test_harness_stop_server", true) and next(players_tests) == nil then
            request_shutdown()
        end

        -- list of needed players
        local players_table = {}
        for _, t in ipairs(players_tests) do
            if t.players and next(t.players) then
                for _, p in ipairs(t.players) do
                    players_table[p] = true
                end
            end
        end
        local needed_playernames = {}
        for k, _ in pairs(players_table) do table.insert(needed_playernames, k) end

        for _, player_name in ipairs(needed_playernames) do
            print("registering player " .. player_name)
            minetest.set_player_password(player_name,
                minetest.get_password_hash(player_name,
                    minetest.settings:get("test_harness_test_players_password") or "test"))
        end

        -- launch test
        minetest.register_on_joinplayer(function(player)
            -- if player tests are started or done, do nothing
            if tests_state == TESTS_STATE_ENUM.DONE or tests_state == TESTS_STATE_ENUM.STARTED_PLAYERS then return end

            -- else check that all necessary players are connected before starting the tests (we have the list in players_table)
            local connected_player_names = get_connected_player_names()
            if all_in_table(needed_playernames, connected_player_names) then
                tests_state = TESTS_STATE_ENUM.STARTED_PLAYERS
                run_player_tests(players_tests)
            end
        end)

        minetest.register_on_leaveplayer(function(player, timeout)
            if tests_state ~= TESTS_STATE_ENUM.STARTED_PLAYERS then
                -- check that needed playernames is still in the list of connected players
                local connected_player_names = get_connected_player_names()
                if not all_in_table(needed_playernames, connected_player_names) then
                    tests_state = TESTS_STATE_ENUM.STARTED
                end
            end
        end)
    end)
end


-- for debug purposes
minetest.register_on_joinplayer(function(player)
    minetest.set_player_privs(player:get_player_name(),
        minetest.string_to_privs("fly,fast,noclip,basic_debug,debug,interact"))
end)
minetest.register_on_punchnode(function(pos, node, puncher)
    minetest.chat_send_player(puncher:get_player_name(), pos2str(pos))
end)

minetest.after(0, run_tests)