/** * Scenario 13: Cross-System Integration (IO + DataNode) * * Tests integration between IntraIO pub/sub system and IDataTree/IDataNode system. * Validates that modules can communicate via IO while sharing data via DataNode. */ #include "grove/JsonDataNode.h" #include "grove/JsonDataTree.h" #include "grove/IOFactory.h" #include "../helpers/TestMetrics.h" #include "../helpers/TestAssertions.h" #include "../helpers/TestReporter.h" #include #include #include #include #include #include using namespace grove; int main() { TestReporter reporter("Cross-System Integration Test"); TestMetrics metrics; std::cout << "================================================================================\n"; std::cout << "TEST: Cross-System Integration (IO + DataNode)\n"; std::cout << "================================================================================\n\n"; // === SETUP === std::cout << "Setup: Creating test directories...\n"; std::filesystem::create_directories("test_cross/config"); std::filesystem::create_directories("test_cross/data"); auto tree = std::make_unique("test_cross"); // Create IO instances auto configWatcherIO = IOFactory::create("intra", "ConfigWatcher"); auto playerIO = IOFactory::create("intra", "Player"); auto economyIO = IOFactory::create("intra", "Economy"); auto metricsIO = IOFactory::create("intra", "Metrics"); if (!configWatcherIO || !playerIO || !economyIO || !metricsIO) { std::cerr << "❌ Failed to create IO instances\n"; return 1; } // ======================================================================== // TEST 1: Config Hot-Reload → IO Broadcast // ======================================================================== std::cout << "\n=== TEST 1: Config Hot-Reload → IO Broadcast ===\n"; // Create initial config file nlohmann::json gameplayConfig = { {"difficulty", "normal"}, {"hpMultiplier", 1.0} }; std::ofstream configFile("test_cross/config/gameplay.json"); configFile << gameplayConfig.dump(2); configFile.close(); // Load config tree->loadConfigFile("gameplay.json"); // Player subscribes to config changes playerIO->subscribe("config:gameplay:changed"); // Setup reload callback for ConfigWatcher std::atomic configChangedEvents{0}; tree->onTreeReloaded([&]() { std::cout << " → Config reloaded, publishing event...\n"; auto data = std::make_unique("configChange", nlohmann::json{ {"config", "gameplay"}, {"timestamp", 12345} }); configWatcherIO->publish("config:gameplay:changed", std::move(data)); }); // Modify config file std::this_thread::sleep_for(std::chrono::milliseconds(100)); gameplayConfig["difficulty"] = "hard"; gameplayConfig["hpMultiplier"] = 1.5; std::ofstream configFile2("test_cross/config/gameplay.json"); configFile2 << gameplayConfig.dump(2); configFile2.close(); auto reloadStart = std::chrono::high_resolution_clock::now(); // Trigger reload if (tree->reloadIfChanged()) { std::cout << " Config was reloaded\n"; } std::this_thread::sleep_for(std::chrono::milliseconds(10)); // Check if player received message if (playerIO->hasMessages() > 0) { auto msg = playerIO->pullMessage(); configChangedEvents++; // Read new config from tree auto configRoot = tree->getConfigRoot(); auto gameplay = configRoot->getChild("gameplay"); if (gameplay) { std::string difficulty = gameplay->getString("difficulty"); double hpMult = gameplay->getDouble("hpMultiplier"); std::cout << " PlayerModule received config change: difficulty=" << difficulty << ", hpMult=" << hpMult << "\n"; ASSERT_EQ(difficulty, "hard", "Difficulty should be updated"); ASSERT_TRUE(std::abs(hpMult - 1.5) < 0.001, "HP multiplier should be updated"); } } auto reloadEnd = std::chrono::high_resolution_clock::now(); float reloadLatency = std::chrono::duration(reloadEnd - reloadStart).count(); std::cout << "Total latency (reload + publish + subscribe + read): " << reloadLatency << "ms\n"; ASSERT_LT(reloadLatency, 200.0f, "Total latency should be reasonable"); ASSERT_EQ(configChangedEvents.load(), 1, "Should receive exactly 1 config change event"); reporter.addMetric("config_reload_latency_ms", reloadLatency); reporter.addAssertion("config_hotreload_chain", true); std::cout << "✓ TEST 1 PASSED\n"; // ======================================================================== // TEST 2: State Persistence + Event Publishing // ======================================================================== std::cout << "\n=== TEST 2: State Persistence + Event Publishing ===\n"; auto dataRoot = tree->getDataRoot(); // Create player node auto player = std::make_unique("player", nlohmann::json::object()); auto profile = std::make_unique("profile", nlohmann::json{ {"name", "TestPlayer"}, {"level", 5}, {"gold", 1000} }); player->setChild("profile", std::move(profile)); dataRoot->setChild("player", std::move(player)); // Save to disk bool saved = tree->saveData(); ASSERT_TRUE(saved, "Should save data successfully"); std::cout << " Data saved to disk\n"; // Publish level up event auto levelUpData = std::make_unique("levelUp", nlohmann::json{ {"event", "level_up"}, {"newLevel", 6}, {"goldBonus", 500} }); playerIO->publish("player:level_up", std::move(levelUpData)); // Economy subscribes to player events economyIO->subscribe("player:*"); std::this_thread::sleep_for(std::chrono::milliseconds(10)); // Economy processes message int messagesReceived = 0; while (economyIO->hasMessages() > 0) { auto msg = economyIO->pullMessage(); messagesReceived++; std::cout << " EconomyModule received: " << msg.topic << "\n"; // Read player data from tree auto playerData = tree->getDataRoot()->getChild("player"); if (playerData) { auto profileData = playerData->getChild("profile"); if (profileData) { int gold = profileData->getInt("gold"); std::cout << " Player gold: " << gold << "\n"; ASSERT_EQ(gold, 1000, "Gold should match saved value"); } } } ASSERT_EQ(messagesReceived, 1, "Should receive 1 player event"); reporter.addAssertion("state_persistence_chain", true); std::cout << "✓ TEST 2 PASSED\n"; // ======================================================================== // TEST 3: Multi-Module State Synchronization // ======================================================================== std::cout << "\n=== TEST 3: Multi-Module State Synchronization ===\n"; int syncErrors = 0; for (int i = 0; i < 10; i++) { // Update gold in DataNode int goldValue = 1000 + i * 10; auto playerNode = tree->getDataRoot()->getChild("player"); if (playerNode) { auto profileNode = playerNode->getChild("profile"); if (profileNode) { profileNode->setInt("gold", goldValue); // Save back to tree playerNode->setChild("profile", std::move(profileNode)); tree->getDataRoot()->setChild("player", std::move(playerNode)); } } // Publish event with same value auto goldUpdate = std::make_unique("goldUpdate", nlohmann::json{ {"event", "gold_updated"}, {"gold", goldValue} }); playerIO->publish("player:gold:updated", std::move(goldUpdate)); std::this_thread::sleep_for(std::chrono::milliseconds(5)); // Economy verifies synchronization if (economyIO->hasMessages() > 0) { auto msg = economyIO->pullMessage(); int msgGold = msg.data->getInt("gold"); // Read from DataNode auto playerCheck = tree->getDataRoot()->getChild("player"); if (playerCheck) { auto profileCheck = playerCheck->getChild("profile"); if (profileCheck) { int dataGold = profileCheck->getInt("gold"); if (msgGold != dataGold) { std::cerr << " SYNC ERROR: msg=" << msgGold << " data=" << dataGold << "\n"; syncErrors++; } } } } } std::cout << "Synchronization errors: " << syncErrors << " / 10\n"; ASSERT_EQ(syncErrors, 0, "Should have zero synchronization errors"); reporter.addMetric("sync_errors", syncErrors); reporter.addAssertion("state_synchronization", syncErrors == 0); std::cout << "✓ TEST 3 PASSED\n"; // ======================================================================== // TEST 4: Runtime Metrics Collection // ======================================================================== std::cout << "\n=== TEST 4: Runtime Metrics Collection ===\n"; auto runtimeRoot = tree->getRuntimeRoot(); // Subscribe to metrics with low-frequency SubscriptionConfig metricsConfig; metricsConfig.replaceable = true; metricsConfig.batchInterval = 1000; // 1 second playerIO->subscribeLowFreq("metrics:*", metricsConfig); // Publish 20 metrics over 2 seconds for (int i = 0; i < 20; i++) { auto metricsData = std::make_unique("metrics", nlohmann::json{ {"fps", 60.0}, {"memory", 125000000 + i * 1000}, {"messageCount", i} }); metricsIO->publish("metrics:snapshot", std::move(metricsData)); std::this_thread::sleep_for(std::chrono::milliseconds(100)); } // Check batched messages std::this_thread::sleep_for(std::chrono::milliseconds(200)); int snapshotsReceived = 0; while (playerIO->hasMessages() > 0) { playerIO->pullMessage(); snapshotsReceived++; } std::cout << "Snapshots received: " << snapshotsReceived << " (expected ~2 due to batching)\n"; ASSERT_TRUE(snapshotsReceived >= 1 && snapshotsReceived <= 4, "Should receive batched snapshots"); // Verify runtime not persisted ASSERT_FALSE(std::filesystem::exists("test_cross/runtime"), "Runtime data should not be persisted"); reporter.addMetric("batched_snapshots", snapshotsReceived); reporter.addAssertion("runtime_metrics", true); std::cout << "✓ TEST 4 PASSED\n"; // ======================================================================== // TEST 5: Concurrent Access (IO + DataNode) // ======================================================================== std::cout << "\n=== TEST 5: Concurrent Access ===\n"; std::atomic running{true}; std::atomic publishCount{0}; std::atomic readCount{0}; std::atomic errors{0}; // Thread 1: Publish events std::thread pubThread([&]() { while (running) { try { auto data = std::make_unique("data", nlohmann::json{{"id", publishCount++}}); playerIO->publish("concurrent:test", std::move(data)); std::this_thread::sleep_for(std::chrono::milliseconds(10)); } catch (...) { errors++; } } }); // Thread 2: Read DataNode std::thread readThread([&]() { while (running) { try { auto playerData = tree->getDataRoot()->getChild("player"); if (playerData) { auto profileData = playerData->getChild("profile"); if (profileData) { int gold = profileData->getInt("gold", 0); readCount++; } } std::this_thread::sleep_for(std::chrono::milliseconds(20)); } catch (...) { errors++; } } }); // Run for 2 seconds std::this_thread::sleep_for(std::chrono::seconds(2)); running = false; pubThread.join(); readThread.join(); std::cout << "Concurrent test completed:\n"; std::cout << " Publishes: " << publishCount << "\n"; std::cout << " Reads: " << readCount << "\n"; std::cout << " Errors: " << errors << "\n"; ASSERT_EQ(errors.load(), 0, "Should have zero errors during concurrent access"); ASSERT_GT(publishCount.load(), 0, "Should have published messages"); ASSERT_GT(readCount.load(), 0, "Should have read data"); reporter.addMetric("concurrent_publishes", publishCount); reporter.addMetric("concurrent_reads", readCount); reporter.addMetric("concurrent_errors", errors); reporter.addAssertion("concurrent_access", errors == 0); std::cout << "✓ TEST 5 PASSED\n"; // ======================================================================== // CLEANUP // ======================================================================== std::filesystem::remove_all("test_cross"); // ======================================================================== // RAPPORT FINAL // ======================================================================== metrics.printReport(); reporter.printFinalReport(); return reporter.getExitCode(); }