aos/util/config_validator_lib.cc - RealtimeRoboticsGroup/test - Gitiles

 #include "aos/util/config_validator_lib.h"

 #include <algorithm>
 #include <chrono>
 #include <cstdlib>
 #include <initializer_list>
 #include <map>
 #include <memory>
 #include <ostream>
 #include <set>
 #include <string>
 #include <string_view>
 #include <utility>
 #include <vector>

 #include "absl/flags/declare.h"
 #include "absl/log/check.h"
 #include "absl/log/log.h"
 #include "flatbuffers/buffer.h"
 #include "flatbuffers/detached_buffer.h"
 #include "flatbuffers/string.h"
 #include "flatbuffers/vector.h"
 #include "gtest/gtest.h"

 #include "aos/events/event_loop.h"
 #include "aos/events/logging/log_reader.h"
 #include "aos/events/logging/logfile_sorting.h"
 #include "aos/events/logging/logfile_utils.h"
 #include "aos/events/simulated_event_loop.h"
 #include "aos/flatbuffers/builder.h"
 #include "aos/flatbuffers/static_vector.h"
 #include "aos/json_to_flatbuffer.h"
 #include "aos/network/remote_message_generated.h"
 #include "aos/network/timestamp_channel.h"
 #include "aos/testing/tmpdir.h"
 #include "aos/util/config_validator_config_static.h"
 #include "aos/util/file.h"
 #include "aos/util/simulation_logger.h"

 ABSL_DECLARE_FLAG(bool, validate_timestamp_logger_nodes);

 namespace aos::util {

 namespace {
 void RunSimulationAndExit(const aos::Configuration *config) {
   aos::SimulatedEventLoopFactory factory(config);

   factory.RunFor(std::chrono::seconds(1));

   std::exit(EXIT_SUCCESS);
 }

 // Checks if either the node is in the specified list of node names or if the
 // list is empty (in which case it is treated as matching all nodes).
 bool NodeInList(
     const flatbuffers::Vector<flatbuffers::Offset<flatbuffers::String>> *list,
     const aos::Node *node) {
   if (list == nullptr || list->size() == 0) {
     return true;
   }
   for (const flatbuffers::String *name : *list) {
     if (name->string_view() == node->name()->string_view()) {
       return true;
     }
   }
   return false;
 }

 }  // namespace

 void ConfigIsValid(const aos::Configuration *config,
                    const ConfigValidatorConfig *validation_config_raw) {
   ASSERT_TRUE(config->has_channels())
       << "An AOS config must have channels. If you have a valid use-case for "
          "channels with no channels, please write a design proposal.";

   aos::fbs::Builder<ConfigValidatorConfigStatic> validation_config;
   CHECK(validation_config->FromFlatbuffer(validation_config_raw));

   if (validation_config_raw->has_logging() &&
       validation_config_raw->logging()->validate_individual_node_loggers() &&
       configuration::MultiNode(config)) {
     if (!validation_config->logging()->has_logger_sets()) {
       validation_config->mutable_logging()->add_logger_sets();
     }
     auto logger_sets =
         validation_config->mutable_logging()->mutable_logger_sets();
     for (const aos::Node *node : configuration::GetNodes(config)) {
       CHECK(logger_sets->reserve(logger_sets->size() + 1));
       auto logger_set = logger_sets->emplace_back();
       CHECK(logger_set->add_loggers()->FromFlatbuffer({node->name()->str()}));
       CHECK(logger_set->add_replay_nodes()->FromFlatbuffer(
           {node->name()->str()}));
     }
   }

   // First, we do some sanity checks--these are likely to indicate a malformed
   // config, and so catching them early with a clear error message is likely to
   // help.

   // The set of all channels that are required by the channels that are
   // configured--these are the remote timestamp channels that *must* be present,
   // and ideally there are no other channels present.
   std::set<const Channel *> required_timestamp_channels;
   // The set of all channels that *look* like remote timestamp channels. This
   // may include channels that are improperly configured and thus have typos &
   // aren't actually going to do anything at runtime.
   std::set<const Channel *> configured_timestamp_channels;
   bool validation_failed = false;
   for (size_t channel_index = 0; channel_index < config->channels()->size();
        ++channel_index) {
     const aos::Channel *channel = config->channels()->Get(channel_index);
     ASSERT_TRUE(channel->has_name()) << "All AOS channels must have a name.";
     ASSERT_TRUE(channel->has_type()) << "All AOS channels must have a type.";

     const bool channel_looks_like_remote_message_channel =
         channel->type()->string_view() ==
         message_bridge::RemoteMessage::GetFullyQualifiedName();

     const bool check_for_not_logged_channels =
         !validation_config->has_logging() ||
         validation_config->AsFlatbuffer().logging()->all_channels_logged();
     const bool channel_is_not_logged =
         channel->logger() == aos::LoggerConfig::NOT_LOGGED;
     if (check_for_not_logged_channels) {
       if (channel_looks_like_remote_message_channel != channel_is_not_logged) {
         LOG(WARNING)
             << "Channel " << configuration::StrippedChannelToString(channel)
             << " is " << EnumNameLoggerConfig(channel->logger()) << " but "
             << (channel_looks_like_remote_message_channel ? "is" : "is not")
             << " a remote timestamp channel. This is almost certainly wrong.";
         validation_failed = true;
       }
     }

     if (channel_looks_like_remote_message_channel) {
       configured_timestamp_channels.insert(channel);
     } else {
       if (channel->has_destination_nodes()) {
         // TODO(james): Technically the timestamp finder should receive a
         // non-empty application name. However, there are no known users that
         // care at this moment.
         message_bridge::ChannelTimestampFinder timestamp_finder(
             config, "",
             configuration::GetNode(config,
                                    channel->source_node()->string_view()));
         for (const Connection *connection : *channel->destination_nodes()) {
           switch (connection->timestamp_logger()) {
             case LoggerConfig::NOT_LOGGED:
             case LoggerConfig::LOCAL_LOGGER:
               if (connection->has_timestamp_logger_nodes()) {
                 LOG(WARNING)
                     << "Connections that are "
                     << EnumNameLoggerConfig(connection->timestamp_logger())
                     << " should not have remote timestamp logger nodes "
                        "populated. This is for the connection to "
                     << connection->name()->string_view() << " on "
                     << configuration::StrippedChannelToString(channel);
                 validation_failed = true;
               }
               break;
             case LoggerConfig::REMOTE_LOGGER:
             case LoggerConfig::LOCAL_AND_REMOTE_LOGGER:
               if (!connection->has_timestamp_logger_nodes() ||
                   connection->timestamp_logger_nodes()->size() != 1 ||
                   connection->timestamp_logger_nodes()->Get(0)->string_view() !=
                       channel->source_node()->string_view()) {
                 LOG(WARNING)
                     << "Connections that are "
                     << EnumNameLoggerConfig(connection->timestamp_logger())
                     << " should have exactly 1 remote timestamp logger node "
                        "populated, and that node should be the source_node ("
                     << channel->source_node()->string_view()
                     << "). This is for the connection to "
                     << connection->name()->string_view() << " on "
                     << configuration::StrippedChannelToString(channel);
                 validation_failed = true;
               }
               // TODO(james): This will be overly noisy, as it ends up
               // CHECK-failing.
               const Channel *found_channel =
                   timestamp_finder.ForChannel(channel, connection);
               CHECK(found_channel != nullptr);
               required_timestamp_channels.insert(found_channel);
               break;
           }
         }
       }
     }
   }

   // Check that all of the things that look like timestamp channels are indeed
   // required.
   // Note: Because ForChannel() will die if a required channel is not present,
   // we do not do a separate check that all the required channels exist.
   for (const auto &channel : configured_timestamp_channels) {
     if (required_timestamp_channels.count(channel) == 0) {
       LOG(WARNING) << "Timestamp channel "
                    << configuration::StrippedChannelToString(channel)
                    << " was specified in the config but is not used.";
       validation_failed = true;
     }
   }

   if (validation_failed) {
     FAIL() << "Remote timestamp linting failed.";
     return;
   }

   // Because the most common way for simulation to fail involves it dying, force
   // it to fail in a slightly more controlled manner.
   ASSERT_EXIT(RunSimulationAndExit(config),
               ::testing::ExitedWithCode(EXIT_SUCCESS), "");

   if (!validation_config->has_logging() || !configuration::MultiNode(config)) {
     return;
   }

   // We will run all the logger configs in two modes:
   // 1) We don't send any data on any non-infrastructure channels; this confirms
   //    that the logs are readable in the absence of any user applications being
   //    present.
   // 2) We confirm that we can generate a good logfile that actually has data
   //    on every channel (some checks in the LogReader may not get hit if there
   //    is no data on a given channel).
   const std::string log_path = aos::testing::TestTmpDir() + "/logs/";
   for (const bool send_data_on_channels : {false, true}) {
     SCOPED_TRACE(send_data_on_channels);
     // Single nodes (multi-nodes with node count = 1) will not produce readable
     // logs in the absense of data.
     if (!send_data_on_channels && (configuration::NodesCount(config) == 1u)) {
       continue;
     }
     // Send timing report when we are sending data.
     const bool do_skip_timing_report = !send_data_on_channels;
     for (const LoggerNodeSetValidationStatic &logger_set :
          *validation_config->logging()->logger_sets()) {
       SCOPED_TRACE(aos::FlatbufferToJson(&logger_set.AsFlatbuffer()));
       aos::SimulatedEventLoopFactory factory(config);
       std::vector<std::unique_ptr<LoggerState>> loggers;
       if (logger_set.has_loggers() && logger_set.loggers()->size() > 0) {
         std::vector<std::string> logger_nodes;
         for (const auto &node : *logger_set.loggers()) {
           logger_nodes.push_back(node.str());
         }
         loggers = MakeLoggersForNodes(&factory, logger_nodes, log_path,
                                       do_skip_timing_report);
       } else {
         loggers =
             MakeLoggersForAllNodes(&factory, log_path, do_skip_timing_report);
       }

       std::vector<std::unique_ptr<EventLoop>> test_loops;
       std::map<std::string, std::vector<std::unique_ptr<RawSender>>>
           test_senders;

       if (send_data_on_channels) {
         // Make a sender on every non-infrastructure channel on every node
         // (including channels that may not be observable by the current logger
         // set).
         for (const aos::Node *node : configuration::GetNodes(config)) {
           test_loops.emplace_back(factory.MakeEventLoop("", node));
           for (const aos::Channel *channel : *config->channels()) {
             // TODO(james): Make a more sophisticated check for "infrastructure"
             // channels than just looking for a "/aos" in the channel--we don't
             // accidentally want to spam nonsense data onto any timestamp
             // channels, though.
             if (configuration::ChannelIsSendableOnNode(channel, node) &&
                 channel->name()->str().find("/aos") == std::string::npos &&
                 channel->logger() != LoggerConfig::NOT_LOGGED) {
               test_senders[node->name()->str()].emplace_back(
                   test_loops.back()->MakeRawSender(channel));
               RawSender *sender =
                   test_senders[node->name()->str()].back().get();
               test_loops.back()->OnRun([sender, channel]() {
                 flatbuffers::DetachedBuffer buffer =
                     JsonToFlatbuffer("{}", channel->schema());
                 sender->CheckOk(sender->Send(buffer.data(), buffer.size()));
               });
             }
           }
         }
       }

       factory.RunFor(std::chrono::seconds(2));

       // Get all of the loggers to close before trying to read the logfiles.
       loggers.clear();

       // Confirm that we can read the log, and that if we put data in it that we
       // can find data on all the nodes that the user cares about.
       logger::LogReader reader(logger::SortParts(logger::FindLogs(log_path)));
       SimulatedEventLoopFactory replay_factory(reader.configuration());
       reader.RegisterWithoutStarting(&replay_factory);

       // Find every channel we deliberately sent data on, and if it is for a
       // node that we care about, confirm that we get it during replay.
       std::vector<std::unique_ptr<EventLoop>> replay_loops;
       for (const aos::Node *node :
            configuration::GetNodes(replay_factory.configuration())) {
         // If the user doesn't care about this node, don't check it.
         if (!NodeInList(logger_set.has_replay_nodes()
                             ? logger_set.replay_nodes()->AsFlatbufferVector()
                             : nullptr,
                         node)) {
           continue;
         }
         replay_loops.emplace_back(replay_factory.MakeEventLoop("", node));
       }

       std::vector<std::pair<const aos::Node *, std::unique_ptr<RawFetcher>>>
           fetchers;
       for (const auto &node_senders : test_senders) {
         for (const auto &sender : node_senders.second) {
           for (auto &loop : replay_loops) {
             if (configuration::ChannelIsReadableOnNode(sender->channel(),
                                                        loop->node())) {
               fetchers.push_back(std::make_pair(
                   loop->node(),
                   loop->MakeRawFetcher(configuration::GetChannel(
                       replay_factory.configuration(), sender->channel(),
                       loop->name(), loop->node()))));
             }
           }
         }
       }

       replay_factory.Run();

       for (auto &pair : fetchers) {
         EXPECT_TRUE(pair.second->Fetch())
             << "Failed to log or replay any data on "
             << configuration::StrippedChannelToString(pair.second->channel())
             << " reading from " << logger::MaybeNodeName(pair.first)
             << " with source node "
             << (pair.second->channel()->has_source_node()
                     ? pair.second->channel()->source_node()->string_view()
                     : "")
             << ".";
       }

       reader.Deregister();

       // Clean up the logs.
       UnlinkRecursive(log_path);
     }
   }
 }

 }  // namespace aos::util
	#include "aos/util/config_validator_lib.h"

	#include <algorithm>
	#include <chrono>
	#include <cstdlib>
	#include <initializer_list>
	#include <map>
	#include <memory>
	#include <ostream>
	#include <set>
	#include <string>
	#include <string_view>
	#include <utility>
	#include <vector>

	#include "absl/flags/declare.h"
	#include "absl/log/check.h"
	#include "absl/log/log.h"
	#include "flatbuffers/buffer.h"
	#include "flatbuffers/detached_buffer.h"
	#include "flatbuffers/string.h"
	#include "flatbuffers/vector.h"
	#include "gtest/gtest.h"

	#include "aos/events/event_loop.h"
	#include "aos/events/logging/log_reader.h"
	#include "aos/events/logging/logfile_sorting.h"
	#include "aos/events/logging/logfile_utils.h"
	#include "aos/events/simulated_event_loop.h"
	#include "aos/flatbuffers/builder.h"
	#include "aos/flatbuffers/static_vector.h"
	#include "aos/json_to_flatbuffer.h"
	#include "aos/network/remote_message_generated.h"
	#include "aos/network/timestamp_channel.h"
	#include "aos/testing/tmpdir.h"
	#include "aos/util/config_validator_config_static.h"
	#include "aos/util/file.h"
	#include "aos/util/simulation_logger.h"

	ABSL_DECLARE_FLAG(bool, validate_timestamp_logger_nodes);

	namespace aos::util {

	namespace {
	void RunSimulationAndExit(const aos::Configuration *config) {
	aos::SimulatedEventLoopFactory factory(config);

	factory.RunFor(std::chrono::seconds(1));

	std::exit(EXIT_SUCCESS);
	}

	// Checks if either the node is in the specified list of node names or if the
	// list is empty (in which case it is treated as matching all nodes).
	bool NodeInList(
	const flatbuffers::Vector<flatbuffers::Offset<flatbuffers::String>> *list,
	const aos::Node *node) {
	if (list == nullptr \|\| list->size() == 0) {
	return true;
	}
	for (const flatbuffers::String name : list) {
	if (name->string_view() == node->name()->string_view()) {
	return true;
	}
	}
	return false;
	}

	} // namespace

	void ConfigIsValid(const aos::Configuration *config,
	const ConfigValidatorConfig *validation_config_raw) {
	ASSERT_TRUE(config->has_channels())
	<< "An AOS config must have channels. If you have a valid use-case for "
	"channels with no channels, please write a design proposal.";

	aos::fbs::Builder<ConfigValidatorConfigStatic> validation_config;
	CHECK(validation_config->FromFlatbuffer(validation_config_raw));

	if (validation_config_raw->has_logging() &&
	validation_config_raw->logging()->validate_individual_node_loggers() &&
	configuration::MultiNode(config)) {
	if (!validation_config->logging()->has_logger_sets()) {
	validation_config->mutable_logging()->add_logger_sets();
	}
	auto logger_sets =
	validation_config->mutable_logging()->mutable_logger_sets();
	for (const aos::Node *node : configuration::GetNodes(config)) {
	CHECK(logger_sets->reserve(logger_sets->size() + 1));
	auto logger_set = logger_sets->emplace_back();
	CHECK(logger_set->add_loggers()->FromFlatbuffer({node->name()->str()}));
	CHECK(logger_set->add_replay_nodes()->FromFlatbuffer(
	{node->name()->str()}));
	}
	}

	// First, we do some sanity checks--these are likely to indicate a malformed
	// config, and so catching them early with a clear error message is likely to
	// help.

	// The set of all channels that are required by the channels that are
	// configured--these are the remote timestamp channels that must be present,
	// and ideally there are no other channels present.
	std::set<const Channel *> required_timestamp_channels;
	// The set of all channels that look like remote timestamp channels. This
	// may include channels that are improperly configured and thus have typos &
	// aren't actually going to do anything at runtime.
	std::set<const Channel *> configured_timestamp_channels;
	bool validation_failed = false;
	for (size_t channel_index = 0; channel_index < config->channels()->size();
	++channel_index) {
	const aos::Channel *channel = config->channels()->Get(channel_index);
	ASSERT_TRUE(channel->has_name()) << "All AOS channels must have a name.";
	ASSERT_TRUE(channel->has_type()) << "All AOS channels must have a type.";

	const bool channel_looks_like_remote_message_channel =
	channel->type()->string_view() ==
	message_bridge::RemoteMessage::GetFullyQualifiedName();

	const bool check_for_not_logged_channels =
	!validation_config->has_logging() \|\|
	validation_config->AsFlatbuffer().logging()->all_channels_logged();
	const bool channel_is_not_logged =
	channel->logger() == aos::LoggerConfig::NOT_LOGGED;
	if (check_for_not_logged_channels) {
	if (channel_looks_like_remote_message_channel != channel_is_not_logged) {
	LOG(WARNING)
	<< "Channel " << configuration::StrippedChannelToString(channel)
	<< " is " << EnumNameLoggerConfig(channel->logger()) << " but "
	<< (channel_looks_like_remote_message_channel ? "is" : "is not")
	<< " a remote timestamp channel. This is almost certainly wrong.";
	validation_failed = true;
	}
	}

	if (channel_looks_like_remote_message_channel) {
	configured_timestamp_channels.insert(channel);
	} else {
	if (channel->has_destination_nodes()) {
	// TODO(james): Technically the timestamp finder should receive a
	// non-empty application name. However, there are no known users that
	// care at this moment.
	message_bridge::ChannelTimestampFinder timestamp_finder(
	config, "",
	configuration::GetNode(config,
	channel->source_node()->string_view()));
	for (const Connection connection : channel->destination_nodes()) {
	switch (connection->timestamp_logger()) {
	case LoggerConfig::NOT_LOGGED:
	case LoggerConfig::LOCAL_LOGGER:
	if (connection->has_timestamp_logger_nodes()) {
	LOG(WARNING)
	<< "Connections that are "
	<< EnumNameLoggerConfig(connection->timestamp_logger())
	<< " should not have remote timestamp logger nodes "
	"populated. This is for the connection to "
	<< connection->name()->string_view() << " on "
	<< configuration::StrippedChannelToString(channel);
	validation_failed = true;
	}
	break;
	case LoggerConfig::REMOTE_LOGGER:
	case LoggerConfig::LOCAL_AND_REMOTE_LOGGER:
	if (!connection->has_timestamp_logger_nodes() \|\|
	connection->timestamp_logger_nodes()->size() != 1 \|\|
	connection->timestamp_logger_nodes()->Get(0)->string_view() !=
	channel->source_node()->string_view()) {
	LOG(WARNING)
	<< "Connections that are "
	<< EnumNameLoggerConfig(connection->timestamp_logger())
	<< " should have exactly 1 remote timestamp logger node "
	"populated, and that node should be the source_node ("
	<< channel->source_node()->string_view()
	<< "). This is for the connection to "
	<< connection->name()->string_view() << " on "
	<< configuration::StrippedChannelToString(channel);
	validation_failed = true;
	}
	// TODO(james): This will be overly noisy, as it ends up
	// CHECK-failing.
	const Channel *found_channel =
	timestamp_finder.ForChannel(channel, connection);
	CHECK(found_channel != nullptr);
	required_timestamp_channels.insert(found_channel);
	break;
	}
	}
	}
	}
	}

	// Check that all of the things that look like timestamp channels are indeed
	// required.
	// Note: Because ForChannel() will die if a required channel is not present,
	// we do not do a separate check that all the required channels exist.
	for (const auto &channel : configured_timestamp_channels) {
	if (required_timestamp_channels.count(channel) == 0) {
	LOG(WARNING) << "Timestamp channel "
	<< configuration::StrippedChannelToString(channel)
	<< " was specified in the config but is not used.";
	validation_failed = true;
	}
	}

	if (validation_failed) {
	FAIL() << "Remote timestamp linting failed.";
	return;
	}

	// Because the most common way for simulation to fail involves it dying, force
	// it to fail in a slightly more controlled manner.
	ASSERT_EXIT(RunSimulationAndExit(config),
	::testing::ExitedWithCode(EXIT_SUCCESS), "");

	if (!validation_config->has_logging() \|\| !configuration::MultiNode(config)) {
	return;
	}

	// We will run all the logger configs in two modes:
	// 1) We don't send any data on any non-infrastructure channels; this confirms
	// that the logs are readable in the absence of any user applications being
	// present.
	// 2) We confirm that we can generate a good logfile that actually has data
	// on every channel (some checks in the LogReader may not get hit if there
	// is no data on a given channel).
	const std::string log_path = aos::testing::TestTmpDir() + "/logs/";
	for (const bool send_data_on_channels : {false, true}) {
	SCOPED_TRACE(send_data_on_channels);
	// Single nodes (multi-nodes with node count = 1) will not produce readable
	// logs in the absense of data.
	if (!send_data_on_channels && (configuration::NodesCount(config) == 1u)) {
	continue;
	}
	// Send timing report when we are sending data.
	const bool do_skip_timing_report = !send_data_on_channels;
	for (const LoggerNodeSetValidationStatic &logger_set :
	*validation_config->logging()->logger_sets()) {
	SCOPED_TRACE(aos::FlatbufferToJson(&logger_set.AsFlatbuffer()));
	aos::SimulatedEventLoopFactory factory(config);
	std::vector<std::unique_ptr<LoggerState>> loggers;
	if (logger_set.has_loggers() && logger_set.loggers()->size() > 0) {
	std::vector<std::string> logger_nodes;
	for (const auto &node : *logger_set.loggers()) {
	logger_nodes.push_back(node.str());
	}
	loggers = MakeLoggersForNodes(&factory, logger_nodes, log_path,
	do_skip_timing_report);
	} else {
	loggers =
	MakeLoggersForAllNodes(&factory, log_path, do_skip_timing_report);
	}

	std::vector<std::unique_ptr<EventLoop>> test_loops;
	std::map<std::string, std::vector<std::unique_ptr<RawSender>>>
	test_senders;

	if (send_data_on_channels) {
	// Make a sender on every non-infrastructure channel on every node
	// (including channels that may not be observable by the current logger
	// set).
	for (const aos::Node *node : configuration::GetNodes(config)) {
	test_loops.emplace_back(factory.MakeEventLoop("", node));
	for (const aos::Channel channel : config->channels()) {
	// TODO(james): Make a more sophisticated check for "infrastructure"
	// channels than just looking for a "/aos" in the channel--we don't
	// accidentally want to spam nonsense data onto any timestamp
	// channels, though.
	if (configuration::ChannelIsSendableOnNode(channel, node) &&
	channel->name()->str().find("/aos") == std::string::npos &&
	channel->logger() != LoggerConfig::NOT_LOGGED) {
	test_senders[node->name()->str()].emplace_back(
	test_loops.back()->MakeRawSender(channel));
	RawSender *sender =
	test_senders[node->name()->str()].back().get();
	test_loops.back()->OnRun([sender, channel]() {
	flatbuffers::DetachedBuffer buffer =
	JsonToFlatbuffer("{}", channel->schema());
	sender->CheckOk(sender->Send(buffer.data(), buffer.size()));
	});
	}
	}
	}
	}

	factory.RunFor(std::chrono::seconds(2));

	// Get all of the loggers to close before trying to read the logfiles.
	loggers.clear();

	// Confirm that we can read the log, and that if we put data in it that we
	// can find data on all the nodes that the user cares about.
	logger::LogReader reader(logger::SortParts(logger::FindLogs(log_path)));
	SimulatedEventLoopFactory replay_factory(reader.configuration());
	reader.RegisterWithoutStarting(&replay_factory);

	// Find every channel we deliberately sent data on, and if it is for a
	// node that we care about, confirm that we get it during replay.
	std::vector<std::unique_ptr<EventLoop>> replay_loops;
	for (const aos::Node *node :
	configuration::GetNodes(replay_factory.configuration())) {
	// If the user doesn't care about this node, don't check it.
	if (!NodeInList(logger_set.has_replay_nodes()
	? logger_set.replay_nodes()->AsFlatbufferVector()
	: nullptr,
	node)) {
	continue;
	}
	replay_loops.emplace_back(replay_factory.MakeEventLoop("", node));
	}

	std::vector<std::pair<const aos::Node *, std::unique_ptr<RawFetcher>>>
	fetchers;
	for (const auto &node_senders : test_senders) {
	for (const auto &sender : node_senders.second) {
	for (auto &loop : replay_loops) {
	if (configuration::ChannelIsReadableOnNode(sender->channel(),
	loop->node())) {
	fetchers.push_back(std::make_pair(
	loop->node(),
	loop->MakeRawFetcher(configuration::GetChannel(
	replay_factory.configuration(), sender->channel(),
	loop->name(), loop->node()))));
	}
	}
	}
	}

	replay_factory.Run();

	for (auto &pair : fetchers) {
	EXPECT_TRUE(pair.second->Fetch())
	<< "Failed to log or replay any data on "
	<< configuration::StrippedChannelToString(pair.second->channel())
	<< " reading from " << logger::MaybeNodeName(pair.first)
	<< " with source node "
	<< (pair.second->channel()->has_source_node()
	? pair.second->channel()->source_node()->string_view()
	: "")
	<< ".";
	}

	reader.Deregister();

	// Clean up the logs.
	UnlinkRecursive(log_path);
	}
	}
	}

	} // namespace aos::util