test: cover guided placement and malleability simulator

"""Check movement accounting without starting daemons or touching user data."""

import math

import os

from pathlib import Path

import re

import shutil

import subprocess

import pytest

@pytest.fixture

def simulator(test_workspace):

    for directory in test_workspace.bindirs:

        candidate = Path(directory) / "gkfs_malleability_sim"

        if candidate.is_file() and os.access(candidate, os.X_OK):

            return candidate

    candidate = shutil.which("gkfs_malleability_sim")

    assert candidate, "gkfs_malleability_sim must be built and installed"

    return Path(candidate)

def run_simulator(simulator, *arguments):

    environment = os.environ.copy()

    # Keep the subprocess independent of distributor settings on the test host.

    for key in list(environment):

        if key.startswith("GKFS_RANDOM_SLICING_"):

            del environment[key]

    return subprocess.run(

        [str(simulator), *map(str, arguments)], capture_output=True, text=True,

        env=environment, timeout=30,

    )

@pytest.mark.parametrize("strategy,cutshift", [

    ("simple_hash", False), ("random_slicing", False),

    ("random_slicing", True),

])

@pytest.mark.parametrize("old_nodes,new_nodes", [(4, 2), (2, 4), (3, 3)])

def test_simulator_movement_accounting(simulator, strategy, cutshift,

                                       old_nodes, new_nodes):

    arguments = ["--strategy", strategy, "--old-nodes", old_nodes,

                 "--new-nodes", new_nodes, "--files", 7,

                 "--file-size", 2051, "--chunk-size", 1024]

    if cutshift:

        arguments.append("--cutshift")

    result = run_simulator(simulator, *arguments)

    assert result.returncode == 0, result.stderr

    values = dict(line.split("=", 1) for line in result.stdout.splitlines()

                  if "=" in line and not line.startswith(" "))

    assert values["strategy"] == strategy

    assert values["cutshift"] == ("ON" if cutshift else "OFF")

    assert int(values["old_nodes"]) == old_nodes

    assert int(values["new_nodes"]) == new_nodes

    assert int(values["total_chunks"]) == 21

    assert int(values["total_bytes"]) == 14357

    chunks, size = int(values["moved_chunks"]), int(values["moved_bytes"])

    assert 0 <= chunks <= 21

    assert 0 <= size <= 14357

    assert math.isclose(float(values["movement_ratio"]), size / 14357,

                        rel_tol=1e-5, abs_tol=1e-6)

    sources, targets = result.stdout.split("per_source_moved\n")[1].split(

        "per_target_received\n")

    for section, nodes in [(sources, old_nodes), (targets, new_nodes)]:

        hosts = re.findall(r"host (\d+): chunks=(\d+) bytes=(\d+)", section)

        assert [int(host) for host, _, _ in hosts] == list(range(nodes))

        assert sum(int(count) for _, count, _ in hosts) == chunks

        assert sum(int(count) for _, _, count in hosts) == size

    if old_nodes == new_nodes:

        assert chunks == size == 0

    else:

        assert chunks > 0

    if new_nodes < old_nodes:

        assert (int(values["moved_from_removed_bytes"]) +

                int(values["moved_from_survivor_bytes"])) == size

        assert (int(values["moved_from_removed_chunks"]) +

                int(values["moved_from_survivor_chunks"])) == chunks

        assert float(values["shrink_lower_bound_ratio"]) == 0.5

        survivor_bytes = int(values["moved_from_survivor_bytes"])

        assert math.isclose(

            float(values["extra_survivor_movement_ratio"]),

            survivor_bytes / 14357, rel_tol=1e-5, abs_tol=1e-6,

        )

        if strategy == "random_slicing" and cutshift:

            assert survivor_bytes == 0

@pytest.mark.parametrize("old_nodes,new_nodes", [(4, 2), (2, 4)])

def test_simulator_comparison_matrix(simulator, old_nodes, new_nodes):

    result = run_simulator(simulator, "--old-nodes", old_nodes,

                           "--new-nodes", new_nodes, "--files", 3,

                           "--file-size", 1025, "--chunk-size", 1024)

    assert result.returncode == 0, result.stderr

    assert "strategy comparison" in result.stdout

    rows = [line.split() for line in result.stdout.splitlines()

            if line.startswith(("simple_hash ", "random_slicing "))]

    assert [(row[0], row[1]) for row in rows] == [

        ("simple_hash", "OFF"), ("random_slicing", "OFF"),

        ("random_slicing", "ON"),

    ]

    for row in rows:

        assert 0 <= int(row[2]) <= 3075

        assert 0 <= int(row[5]) <= 6

        assert 0 <= float(row[6]) <= 100

        assert float(row[7]) >= 0

@pytest.mark.parametrize("arguments", [

    ["--old-nodes", "0"], ["--new-nodes", "0"], ["--files", "0"],

    ["--file-size", "0"], ["--chunk-size", "0"],

    ["--strategy", "unknown"], ["--old-nodes", "not-a-number"],

])

def test_simulator_rejects_invalid_options(simulator, arguments):

    result = run_simulator(simulator, *arguments)

    assert result.returncode != 0

    assert result.stderr

    assert "total_bytes=" not in result.stdout

 No newline at end of file

    ${CMAKE_CURRENT_LIST_DIR}/test_path.cpp

    ${CMAKE_CURRENT_LIST_DIR}/test_common_path.cpp

    ${CMAKE_CURRENT_LIST_DIR}/test_distributor.cpp

    ${CMAKE_CURRENT_LIST_DIR}/test_guided_placement.cpp

    ${CMAKE_CURRENT_LIST_DIR}/test_replica_placement.cpp

    ${CMAKE_CURRENT_LIST_DIR}/test_replica_migrator.cpp

    ${CMAKE_CURRENT_LIST_DIR}/test_replica_migration_adapter.cpp

/* SPDX-License-Identifier: GPL-3.0-or-later */

#include <catch2/catch_all.hpp>

#include <common/env.hpp>

#include <common/rpc/distributor.hpp>

#include "helpers/helpers.hpp"

#include <cstdlib>

#include <fstream>

#include <optional>

#include <string>

namespace {

struct saved_environment {

    const char* name;

    std::optional<std::string> value;

    explicit saved_environment(const char* variable) : name(variable) {

        if(const auto* previous = std::getenv(name)) {

            value = previous;

        }

        ::unsetenv(name);

    }

    ~saved_environment() {

        if(value) {

            ::setenv(name, value->c_str(), 1);

        } else {

            ::unsetenv(name);

        }

    }

};

struct guided_mapping {

    saved_environment primary{gkfs::env::GUIDED_DISTRIBUTION_PATH};

    saved_environment legacy{"LIBGKFS_GUIDED_DISTRIBUTION_PATH"};

    saved_environment oldest{"GKFS_GUIDED_PATH"};

    helpers::temporary_directory directory;

    const fs::path path = directory.dirname() / "mapping";

    explicit guided_mapping(const std::string& contents) {

        std::ofstream output(path);

        output << contents;

        output.close();

        REQUIRE(output.good());

        REQUIRE(::setenv(primary.name, path.c_str(), 1) == 0);

    }

};

const std::vector<gkfs::rpc::host_t> all_hosts{0, 1, 2, 3};

} // namespace

TEST_CASE("Guided placement uses same-depth prefixes with component boundaries",

          "[distributor][guided]") {

    guided_mapping mapping("#/alpha 0 2\n#/beta/ 0 1\n#/invalid 0 99\n");

    gkfs::rpc::GuidedDistributor guided(3, 4);

    gkfs::rpc::SimpleHashDistributor hashed(3, 4);

    REQUIRE(guided.localhost() == 3);

    REQUIRE(guided.hosts_size() == 4);

    REQUIRE(guided.locate_directory_metadata() == all_hosts);

    for(const auto& path : {"/alpha", "/alpha/file", "/alpha/sub/file"}) {

        REQUIRE(guided.locate_data(path, 7, 0) == 2);

        REQUIRE(guided.locate_file_metadata(path, 0) == 2);

        REQUIRE(guided.locate_directory_metadata(path) ==

                std::vector<gkfs::rpc::host_t>{2});

    }

    REQUIRE(guided.locate_data("/beta/file", 0, 0) == 1);

    REQUIRE(guided.locate_file_metadata("/beta/file", 0) == 1);

    REQUIRE(guided.locate_directory_metadata("/beta/") ==

            std::vector<gkfs::rpc::host_t>{1});

    for(const auto& path : {"/alphabet/file", "/missing/file", "/", "", "rel"}) {

        REQUIRE(guided.locate_data(path, 7, 0) ==

                hashed.locate_data(path, 7, 0));

        REQUIRE(guided.locate_file_metadata(path, 0) ==

                hashed.locate_file_metadata(path, 0));

        REQUIRE(guided.locate_directory_metadata(path) == all_hosts);

    }

    REQUIRE(guided.locate_data("/invalid/file", 7, 0) ==

            hashed.locate_file_metadata("/invalid/file", 0));

    REQUIRE(guided.locate_file_metadata("/invalid/file", 1) ==

            hashed.locate_file_metadata("/invalid/file", 1));

    REQUIRE(guided.locate_directory_metadata("/invalid") == all_hosts);

}

TEST_CASE("Guided directory placement fans out for conflicting nested rules",

          "[distributor][guided]") {

    guided_mapping mapping(

            "#/parent/child 0 1\n#/parent 0 2\n#/invalid/deep 0 99\n");

    gkfs::rpc::GuidedDistributor guided(0, 4);

    gkfs::rpc::SimpleHashDistributor hashed(0, 4);

    REQUIRE(guided.locate_data("/parent/child/file", 0, 0) == 1);

    REQUIRE(guided.locate_file_metadata("/parent/child/file", 0) == 1);

    REQUIRE(guided.locate_directory_metadata("/parent") == all_hosts);

    REQUIRE(guided.locate_directory_metadata("/parent/other") ==

            std::vector<gkfs::rpc::host_t>{2});

    REQUIRE(guided.locate_data("/parent/other/file", 5, 0) == 2);

    REQUIRE(guided.locate_file_metadata("/parent/other/file", 0) == 2);

    REQUIRE(guided.locate_data("/invalid/deep/file", 5, 0) ==

            hashed.locate_file_metadata("/invalid/deep/file", 0));

    REQUIRE(guided.locate_file_metadata("/invalid/deep/file", 1) ==

            hashed.locate_file_metadata("/invalid/deep/file", 1));

    REQUIRE(guided.locate_directory_metadata("/invalid/deep") == all_hosts);

    REQUIRE(guided.locate_directory_metadata("/unmapped") == all_hosts);

}

TEST_CASE("Guided chunk rules override prefixes and split directory targets",

          "[distributor][guided]") {

    guided_mapping mapping(

            "#/tree 0 2\n/tree/special 0 1\n/tree/special 1 1\n");

    gkfs::rpc::GuidedDistributor guided(0, 4);

    REQUIRE(guided.locate_data("/tree/special", 0, 0) == 1);

    REQUIRE(guided.locate_data("/tree/special", 2, 0) == 1);

    REQUIRE(guided.locate_file_metadata("/tree/special", 0) == 2);

    REQUIRE(guided.locate_directory_metadata("/tree") == all_hosts);

    REQUIRE(guided.locate_directory_metadata("/tree/other") ==

            std::vector<gkfs::rpc::host_t>{2});

}

TEST_CASE("Guided chunk mappings apply to descendants but not sibling names",

          "[distributor][guided]") {

    guided_mapping mapping("/mapped 0 1\n/mapped 1 1\n");

    gkfs::rpc::GuidedDistributor guided(0, 4);

    gkfs::rpc::SimpleHashDistributor hashed(0, 4);

    for(const auto& path : {"/mapped", "/mapped/child"}) {

        REQUIRE(guided.locate_data(path, 0, 0) == 1);

        REQUIRE(guided.locate_data(path, 2, 0) == 1);

        REQUIRE(guided.locate_data(path, 8, 0) ==

                hashed.locate_data(path, 8, 0));

        REQUIRE(guided.locate_file_metadata(path, 0) == 1);

    }

    REQUIRE(guided.locate_data("/mapped-other", 0, 0) ==

            hashed.locate_data("/mapped-other", 0, 0));

    REQUIRE(guided.locate_file_metadata("/mapped-other", 0) ==

            hashed.locate_file_metadata("/mapped-other", 0));

    REQUIRE(guided.locate_directory_metadata("/mapped") == all_hosts);

}

TEST_CASE("Guided mapping environment aliases and missing files fall back safely",

          "[distributor][guided]") {

    guided_mapping mapping("#/alias 0 2\n");

    const auto* variable = GENERATE("LIBGKFS_GUIDED_DISTRIBUTION_PATH",

                                    "GKFS_GUIDED_PATH");

    REQUIRE(::setenv(mapping.primary.name, "", 1) == 0);

    REQUIRE(::setenv(variable, mapping.path.c_str(), 1) == 0);

    gkfs::rpc::GuidedDistributor guided(0, 4);

    REQUIRE(guided.locate_data("/alias/file", 0, 0) == 2);

    REQUIRE(::setenv(mapping.primary.name,

                     (mapping.directory.dirname() / "absent").c_str(), 1) == 0);

    gkfs::rpc::GuidedDistributor fallback(0, 4);

    gkfs::rpc::SimpleHashDistributor hashed(0, 4);

    REQUIRE(fallback.locate_data("/alias/file", 3, 1) ==

            hashed.locate_data("/alias/file", 3, 1));

    REQUIRE(fallback.locate_file_metadata("/alias/file", 1) ==

            hashed.locate_file_metadata("/alias/file", 1));

    REQUIRE(fallback.locate_directory_metadata("/alias") == all_hosts);

}

TEST_CASE("Guided deep prefixes and host count updates preserve placement",

          "[distributor][guided]") {

    guided_mapping mapping("#/one/two 0 1\n");

    gkfs::rpc::GuidedDistributor guided(0, 4);

    REQUIRE(guided.locate_directory_metadata("/one") == all_hosts);

    REQUIRE(guided.locate_directory_metadata("/one/two/") ==

            std::vector<gkfs::rpc::host_t>{1});

    guided.hosts_size(2);

    REQUIRE(guided.hosts_size() == 2);

    REQUIRE(guided.locate_directory_metadata() ==

            std::vector<gkfs::rpc::host_t>{0, 1});

    REQUIRE(guided.locate_data("/one/two/file", 0, 3u, 0) == 1);

    REQUIRE(guided.hosts_size() == 3);

    REQUIRE(guided.locate_data("/one/two/file", 0, 3u, 0) == 1);

    REQUIRE(guided.locate_directory_metadata() ==

            std::vector<gkfs::rpc::host_t>{0, 1, 2});

}

 No newline at end of file