tboudreaux/MeanField_SERiF
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
668fd5cc6d60aeb7df371eee1c12ca3ab4718e22
MeanField_SERiF/tests/utils/dependencies/observers.cpp
Emily Boudreaux cb20faeb3d test(tests): added test suite 
The test suite is extensive. Note that LLMs were used to aid in the development of the test suite. All LLM generated code was manually reviewed.
2026-11-09 13:53:17 -05:00

379 lines
18 KiB
C++
Raw Blame History

#include "serif/utils/dependencies/dependencies.hpp"
#include "serif/utils/dependencies/timing.hpp"
#include "serif/utils/error/errors.hpp"
#include <catch2/catch_test_macros.hpp>
#include <expected>
#include <memory>
#include <stdexcept>
#include <string>
#include <utility>
#include <vector>
namespace {
namespace dependencies = serif::utils::dependencies;
using Outcome = dependencies::NodeOutcome;
struct Source {};
struct Offset {};
struct Shared {};
struct Total {};
struct Unrequested {};
/// Simulate exceptional application failures through the library's error entry point.
void fail_observed_work() {
serif::utils::errors::serif_error<std::runtime_error, serif::utils::errors::SERiFErrorCode::INVALID_ARGUMENT>("observed work failed");
}
}
TEST_CASE("Graph observers follow nested node scopes and declared input order only during execution", "[unit][utils][dependencies][observers]") {
int source = 2;
int offset = 3;
std::vector<int> events;
events.reserve(128);
const auto context = dependencies::make_dependency_context(dependencies::borrow<Source>(source), dependencies::borrow<Offset>(offset));
auto graph = dependencies::make_computation_graph(
dependencies::make_computation<Total, dependencies::InputList<Offset, Shared>>([&events](int shift, int value) {
events.push_back(70);
return value + shift;
}),
dependencies::make_computation<Shared, dependencies::InputList<Source>>([&events](int value) {
events.push_back(30);
return 2 * value;
}),
dependencies::make_computation<Unrequested, dependencies::InputList<Source>>([&events](int value) {
events.push_back(99);
return value;
})
);
auto observers = dependencies::make_observers(
dependencies::observe_node<Shared>([&events] {
events.push_back(10);
return std::make_unique<int>(40);
}, [&events](std::unique_ptr<int>& state, Outcome outcome) noexcept {
events.push_back(*state);
CHECK(outcome == Outcome::completed);
}),
dependencies::observe_node<Shared>([&events] {
events.push_back(11);
return 41;
}, [&events](int& state, Outcome outcome) noexcept {
events.push_back(state);
CHECK(outcome == Outcome::completed);
}),
dependencies::observe_node<Total>([&events] {
events.push_back(50);
return 80;
}, [&events](int& state, Outcome outcome) noexcept {
events.push_back(state);
CHECK(outcome == Outcome::completed);
}),
dependencies::observe_edge<Shared, Total>([&events](const int&) { events.push_back(61); }),
dependencies::observe_edge<Offset, Total>([&events](const int&) { events.push_back(60); }),
dependencies::observe_edge<Source, Shared>([&events](const int&) { events.push_back(20); }),
dependencies::observe_edge<Source, Shared>([&events](const int&) { events.push_back(21); })
);
auto cache = dependencies::make_computation_cache(std::move(graph), context, std::move(observers));
REQUIRE(cache.try_prepare(context, dependencies::RequestedResults<Total>{}));
CHECK(events == std::vector<int>{10, 11, 20, 21, 30, 41, 40, 50, 60, 61, 70, 80});
CHECK(cache.get<Total>() == 7);
const auto first = cache.share<Shared>();
events.clear();
REQUIRE(cache.try_prepare(context, dependencies::RequestedResults<Total>{}));
REQUIRE(cache.try_prepare(context, dependencies::RequestedResults<>{}));
CHECK(events.empty());
source = 4;
REQUIRE(cache.try_prepare(context, dependencies::RequestedResults<Shared>{}));
CHECK(events == std::vector<int>{10, 11, 20, 21, 30, 41, 40});
CHECK_FALSE(cache.contains<Total>());
CHECK(*first == 4);
events.clear();
REQUIRE(cache.try_prepare(context, dependencies::RequestedResults<Total>{}));
CHECK(events == std::vector<int>{50, 60, 61, 70, 80});
CHECK(cache.get<Total>() == 11);
events.clear();
REQUIRE(cache.try_prepare(context, dependencies::RequestedResults<Total>{}, dependencies::PreparationMode::rebuild));
CHECK(events == std::vector<int>{10, 11, 20, 21, 30, 41, 40, 50, 60, 61, 70, 80});
events.clear();
cache.invalidate();
REQUIRE(cache.try_prepare(context, dependencies::RequestedResults<Shared>{}));
CHECK(events == std::vector<int>{10, 11, 20, 21, 30, 41, 40});
}
TEST_CASE("Node outcomes describe attempted computation while cache publication remains transactional", "[unit][utils][dependencies][observers]") {
int source = 2;
std::vector<std::pair<int, Outcome>> outcomes;
outcomes.reserve(32);
const auto context = dependencies::make_dependency_context(dependencies::borrow<Source>(source));
auto graph = dependencies::make_computation_graph(
dependencies::make_computation<Shared, dependencies::InputList<Source>>([](int value) { return value; }),
dependencies::make_computation<Total, dependencies::InputList<Shared>>([](int value) -> std::expected<int, std::string> {
if (value == -2) {
fail_observed_work();
}
if (value < 0) {
return std::unexpected("negative input");
}
return 2 * value;
}),
dependencies::make_computation<Unrequested, dependencies::InputList<Total>>([](int value) { return value + 1; })
);
auto observers = dependencies::make_observers(
dependencies::observe_node<Shared>([] { return 1; }, [&outcomes](int& state, Outcome outcome) noexcept { outcomes.emplace_back(state, outcome); }),
dependencies::observe_node<Total>([] { return 2; }, [&outcomes](int& state, Outcome outcome) noexcept { outcomes.emplace_back(state, outcome); }),
dependencies::observe_node<Unrequested>([] { return 3; }, [&outcomes](int& state, Outcome outcome) noexcept { outcomes.emplace_back(state, outcome); })
);
auto cache = dependencies::make_computation_cache(std::move(graph), context, std::move(observers));
REQUIRE(cache.try_prepare(context));
const auto accepted = cache.share<Shared>();
CHECK(outcomes == std::vector<std::pair<int, Outcome>>{{1, Outcome::completed}, {2, Outcome::completed}, {3, Outcome::completed}});
outcomes.clear();
SECTION("A typed rejection completes earlier nodes but does not publish them") {
source = -1;
const auto rejected = cache.try_prepare(context);
REQUIRE_FALSE(rejected);
CHECK(rejected.error().error == "negative input");
CHECK(outcomes == std::vector<std::pair<int, Outcome>>{{1, Outcome::completed}, {2, Outcome::rejected}});
}
SECTION("An escaping exception closes the active node and propagates") {
source = -2;
CHECK_THROWS_AS(cache.try_prepare(context), std::runtime_error);
CHECK(outcomes == std::vector<std::pair<int, Outcome>>{{1, Outcome::completed}, {2, Outcome::exception}});
}
CHECK_FALSE(cache.ready());
CHECK_FALSE(cache.contains<Shared>());
CHECK(*accepted == 2);
outcomes.clear();
source = 2;
const auto recovered = cache.try_prepare(context);
REQUIRE(recovered);
CHECK(recovered->reused_count() == 3);
CHECK(cache.share<Shared>() == accepted);
CHECK(outcomes.empty());
}
TEST_CASE("Observer exceptions close only scopes whose before callback completed", "[unit][utils][dependencies][observers]") {
bool fail_before = false;
int computations = 0;
std::vector<int> events;
events.reserve(16);
const auto context = dependencies::make_dependency_context(dependencies::own<Source>(2));
auto graph = dependencies::make_computation_graph(
dependencies::make_computation<Total, dependencies::InputList<Source>>([&computations](int value) {
++computations;
return value;
})
);
auto observers = dependencies::make_observers(
dependencies::observe_node<Total>([&events] {
events.push_back(1);
return std::make_unique<int>(-1);
}, [&events](std::unique_ptr<int>& state, Outcome outcome) noexcept {
events.push_back(*state);
CHECK(outcome == Outcome::exception);
}),
dependencies::observe_node<Total>([&events, &fail_before] {
events.push_back(2);
if (fail_before) {
fail_observed_work();
}
return -2;
}, [&events](int& state, Outcome outcome) noexcept {
events.push_back(state);
CHECK(outcome == Outcome::exception);
}),
dependencies::observe_edge<Source, Total>([&events](const int&) {
events.push_back(3);
fail_observed_work();
})
);
auto cache = dependencies::make_computation_cache(std::move(graph), context, std::move(observers));
SECTION("A failed before callback has no matching after callback") {
fail_before = true;
CHECK_THROWS_AS(cache.try_prepare(context), std::runtime_error);
CHECK(events == std::vector<int>{1, 2, -1});
}
SECTION("An edge failure unwinds every entered node scope") {
CHECK_THROWS_AS(cache.try_prepare(context), std::runtime_error);
CHECK(events == std::vector<int>{1, 2, 3, -2, -1});
}
CHECK(computations == 0);
CHECK_FALSE(cache.ready());
}
TEST_CASE("Observers follow moved caches and read owning edges without copying move-only results", "[unit][utils][dependencies][observers]") {
using Value = std::unique_ptr<int>;
using Inputs = dependencies::InputList<Source>;
using Snapshot = std::shared_ptr<const Value>;
int source = 4;
const Value* observed = nullptr;
std::vector<std::pair<int, Outcome>> outcomes;
outcomes.reserve(16);
const auto context = dependencies::make_dependency_context(dependencies::borrow<Source>(source));
auto graph = dependencies::make_computation_graph(
dependencies::make_computation<Shared, Inputs>([](int value) -> std::expected<Value, std::string> {
if (value < 0) {
return std::unexpected("negative input");
}
return std::make_unique<int>(value);
}),
dependencies::make_owning_computation<Total, dependencies::InputList<Shared>>([](Snapshot value) { return value; })
);
auto observers = dependencies::make_observers(
dependencies::observe_node<Shared>([calls = std::make_unique<int>(0)] { return ++*calls; },
[&outcomes](int& invocation, Outcome outcome) noexcept { outcomes.emplace_back(invocation, outcome); }),
dependencies::observe_edge<Shared, Total>([&observed](const Value& value) { observed = &value; })
);
auto cache = dependencies::make_computation_cache(std::move(graph), context, std::move(observers));
REQUIRE(cache.try_prepare(context));
const Snapshot first = cache.share<Shared>();
CHECK(observed == first.get());
CHECK(cache.get<Total>() == first);
REQUIRE(cache.try_prepare(context));
CHECK(outcomes == std::vector<std::pair<int, Outcome>>{{1, Outcome::completed}});
auto moved = std::move(cache);
CHECK_FALSE(cache.ready());
source = 9;
REQUIRE(moved.try_prepare(context));
const Snapshot accepted = moved.share<Shared>();
CHECK(observed == accepted.get());
CHECK(**first == 4);
CHECK(**accepted == 9);
source = -1;
REQUIRE_FALSE(moved.try_prepare(context));
CHECK(observed == accepted.get()); // The owning consumer and its edge never ran for the rejected candidate.
source = 9;
REQUIRE(moved.try_prepare(context));
CHECK(moved.share<Shared>() == accepted);
REQUIRE(moved.try_prepare(context, dependencies::RequestedResults<Total>{}, dependencies::PreparationMode::rebuild));
CHECK(observed == moved.share<Shared>().get());
CHECK(outcomes == std::vector<std::pair<int, Outcome>>{{1, Outcome::completed}, {2, Outcome::completed}, {3, Outcome::rejected}, {4, Outcome::completed}});
}
TEST_CASE("A generic node observer composes with selected scopes without observing cache hits", "[unit][utils][dependencies][observers][timing]") {
int source = 3;
std::vector<int> events;
events.reserve(32);
const auto context = dependencies::make_dependency_context(dependencies::borrow<Source>(source));
auto graph = dependencies::make_computation_graph(
dependencies::make_computation<Shared, dependencies::InputList<Source>>([](int value) { return value * 2; }),
dependencies::make_computation<Total, dependencies::InputList<Shared>>([](int value) { return value + 1; }));
auto observers = dependencies::make_observers(
dependencies::observe_nodes([&events]<typename Key>(std::type_identity<Key>) {
const int identity = std::same_as<Key, Shared> ? 1 : 2;
events.push_back(identity);
return identity;
}, [&events](int& identity, Outcome outcome) noexcept {
CHECK(outcome == Outcome::completed);
events.push_back(-identity);
}),
dependencies::observe_node<Shared>([&events] { events.push_back(3); return 3; },
[&events](int& identity, Outcome) noexcept { events.push_back(-identity); }));
auto cache = dependencies::make_computation_cache(std::move(graph), context, std::move(observers));
REQUIRE(cache.try_prepare(context));
CHECK(events == std::vector<int>{1, 3, -3, -1, 2, -2});
CHECK(cache.get<Total>() == 7);
events.clear();
REQUIRE(cache.try_prepare(context));
REQUIRE(cache.try_prepare(context, dependencies::RequestedResults<>{}));
CHECK(events.empty());
}
TEST_CASE("Graph timing counts attempted work and preserves rollback with a deterministic clock", "[unit][utils][dependencies][observers][timing]") {
using TimePoint = std::chrono::steady_clock::time_point;
TimePoint now{};
int clock_reads = 0;
int source = 2;
const auto context = dependencies::make_dependency_context(dependencies::borrow<Source>(source));
auto graph = dependencies::make_computation_graph(
dependencies::make_computation<Total, dependencies::InputList<Shared>>([&now](int value) -> std::expected<int, std::string> {
now += std::chrono::seconds{3};
if (value == -2) { fail_observed_work(); }
if (value < 0) { return std::unexpected("negative input"); }
return 2 * value;
}),
dependencies::make_computation<Shared, dependencies::InputList<Source>>([&now](int value) {
now += std::chrono::seconds{2};
return value;
}),
dependencies::make_computation<Unrequested, dependencies::InputList<Source>>([&now](int value) {
now += std::chrono::seconds{7};
return value;
}));
dependencies::GraphTiming timing;
const auto factory = dependencies::time_graph_nodes(timing, [&]() noexcept { ++clock_reads; return now; });
auto observers = factory(std::type_identity<decltype(graph)>{});
auto cache = dependencies::make_computation_cache(std::move(graph), context, std::move(observers));
REQUIRE(timing.runs().size() == 1);
const auto& nodes = timing.runs()[0]->nodes;
REQUIRE(nodes.size() == 3);
CHECK(nodes[0].index == 0);
CHECK(nodes[1].index == 1);
CHECK(nodes[0].key_name != nodes[1].key_name);
REQUIRE(cache.try_prepare(context, dependencies::RequestedResults<Total>{}));
const auto accepted = cache.share<Shared>();
CHECK(nodes[0].seconds == 3.0);
CHECK(nodes[1].seconds == 2.0);
CHECK(nodes[0].calls == 1);
CHECK(nodes[0].completed == 1);
CHECK(nodes[2].calls == 0);
REQUIRE(cache.try_prepare(context, dependencies::RequestedResults<Total>{}));
CHECK(clock_reads == 4);
source = -1;
const auto rejected = cache.try_prepare(context, dependencies::RequestedResults<Total>{});
REQUIRE_FALSE(rejected);
CHECK(rejected.error().computation_index == 0);
CHECK(nodes[0].rejected == 1);
CHECK(nodes[1].completed == 2);
CHECK(*accepted == 2);
source = 2;
REQUIRE(cache.try_prepare(context, dependencies::RequestedResults<Total>{}));
CHECK(clock_reads == 8);
CHECK(cache.share<Shared>() == accepted);
auto moved = std::move(cache);
source = -2;
CHECK_THROWS_AS(moved.try_prepare(context, dependencies::RequestedResults<Total>{}), std::runtime_error);
CHECK(nodes[0].exceptions == 1);
CHECK(nodes[0].calls == 3);
CHECK(nodes[0].seconds == 9.0);
CHECK(nodes[0].maximum_seconds == 3.0);
CHECK(nodes[1].completed == 3);
CHECK(nodes[2].calls == 0);
CHECK(clock_reads == 12);
source = 2;
REQUIRE(moved.try_prepare(context, dependencies::RequestedResults<Total>{}));
CHECK(clock_reads == 12);
REQUIRE(moved.try_prepare(context, dependencies::RequestedResults<Unrequested>{}));
CHECK(nodes[2].calls == 1);
CHECK(nodes[2].seconds == 7.0);
CHECK(clock_reads == 14);
}
TEST_CASE("Timing observers own their records after the collector is released", "[unit][utils][dependencies][observers][timing]") {
const auto context = dependencies::make_dependency_context();
auto graph = dependencies::make_computation_graph(dependencies::make_computation<Total, dependencies::InputList<>>([] { return 7; }));
std::shared_ptr<const dependencies::GraphTimingRun> run;
auto observers = [&] {
dependencies::GraphTiming timing;
auto result = dependencies::time_graph_nodes(timing)(std::type_identity<decltype(graph)>{});
run = timing.runs()[0];
return result;
}();
auto cache = dependencies::make_computation_cache(std::move(graph), context, std::move(observers));
REQUIRE(cache.try_prepare(context));
CHECK(cache.get<Total>() == 7);
CHECK(run->nodes[0].calls == 1);
CHECK(run->nodes[0].completed == 1);
CHECK(run->nodes[0].seconds >= 0.0);
}
Reference in New Issue View Git Blame Copy Permalink