firmware/test/test_stream_api/test_main.cpp

#include "MeshTypes.h"
#include "SerialConsole.h"
#include "TestUtil.h"
#include "configuration.h"
#include "mesh-pb-constants.h"
#include "mesh/MeshService.h"
#include "mesh/StreamAPI.h"
#include "mesh/StreamFrameWriter.h"
#include <algorithm>
#include <cstdarg>
#include <cstdint>
#include <deque>
#include <limits>
#include <unity.h>
#include <vector>

/// Output-only stream whose write quotas deterministically simulate backpressure.
class ScriptedStream : public Stream
{
  public:
    /// Report that no input bytes are queued.
    int available() override { return 0; }
    /// Return end-of-input for the output-only stream.
    int read() override { return -1; }
    /// Return end-of-input without consuming data.
    int peek() override { return -1; }
    /// Report the configured output capacity.
    int availableForWrite() override { return availableCapacity; }

    /// Route single-byte writes through the quota-aware buffer writer.
    size_t write(uint8_t value) override { return write(&value, 1); }

    /// Accept at most the next scripted quota and capture accepted bytes.
    size_t write(const uint8_t *buffer, size_t size) override
    {
        requestedLengths.push_back(size);
        size_t quota = size;
        if (!writeQuotas.empty()) {
            quota = writeQuotas.front();
            writeQuotas.pop_front();
        }
        size_t accepted = std::min(quota, size);
        output.insert(output.end(), buffer, buffer + accepted);
        return accepted;
    }

    /// Record flush calls without changing captured output.
    void flush() override { flushCount++; }

    /// Set the maximum bytes accepted by the next write call.
    void queueWrite(size_t quota) { writeQuotas.push_back(quota); }

    int availableCapacity = std::numeric_limits<int>::max();
    unsigned flushCount = 0;
    std::deque<size_t> writeQuotas;
    std::vector<size_t> requestedLengths;
    std::vector<uint8_t> output;
};

/// Print sink that records bytes emitted by the real SerialConsole.
class RecordingPrint : public Print
{
  public:
    /// Capture one output byte.
    size_t write(uint8_t value) override
    {
        output.push_back(value);
        return 1;
    }

    std::vector<uint8_t> output;
};

/// Installs a MeshService for a test and restores the previous global service.
class ScopedMeshService
{
  public:
    /// Install the scoped service.
    ScopedMeshService() : previous(service) { service = &instance; }
    /// Restore the prior service after StreamAPI fixtures are destroyed.
    ~ScopedMeshService() { service = previous; }

  private:
    MeshService instance;
    MeshService *previous;
};

/// Exposes generic StreamAPI hooks and records frame-write behavior.
class StreamAPITestShim : public StreamAPI
{
  public:
    /// Construct the shim over a scripted stream.
    explicit StreamAPITestShim(Stream *stream) : StreamAPI(stream) {}

    /// Keep connection-timeout handling inactive during tests.
    bool checkIsConnected() override { return true; }

    /// Invoke the generic transport implementation rather than this shim's capture hook.
    bool writeBaseFrame(uint8_t *buf, size_t len, bool bestEffort = false) { return StreamAPI::writeFrame(buf, len, bestEffort); }

    bool finishReady = true;
    bool allowWrite = true;
    unsigned finishCalls = 0;
    unsigned frameWriteCalls = 0;
    unsigned failureCalls = 0;
    size_t failedFrameLen = 0;
    size_t failedWrittenLen = 0;
    std::vector<uint8_t> capturedPayload;

  protected:
    /// Record the pending-frame gate and return its configured state.
    bool finishPendingFrame() override
    {
        finishCalls++;
        return finishReady;
    }

    /// Apply the configured generic write-readiness result.
    bool canWriteFrame(size_t) override { return allowWrite; }

    /// Capture generic short-write failure metadata.
    void onFrameWriteFailed(size_t frameLen, size_t writtenLen) override
    {
        failureCalls++;
        failedFrameLen = frameLen;
        failedWrittenLen = writtenLen;
    }

    /// Capture one encoded PhoneAPI payload without writing it.
    bool writeFrame(uint8_t *buf, size_t len, bool bestEffort) override
    {
        (void)bestEffort;
        frameWriteCalls++;
        capturedPayload.assign(buf + 4, buf + 4 + len);
        return false;
    }
};

/// Exposes framed-log hooks and records best-effort writes.
class LogHookStreamAPI : public StreamAPI
{
  public:
    /// Construct the log shim over a scripted stream.
    explicit LogHookStreamAPI(Stream *stream) : StreamAPI(stream) {}

    /// Keep connection-timeout handling inactive during tests.
    bool checkIsConnected() override { return true; }

    /// Encode a formatted log through StreamAPI's production log path.
    void emitTestLog(const char *format, ...)
    {
        va_list args;
        va_start(args, format);
        emitLogRecord(meshtastic_LogRecord_Level_INFO, "test", format, args);
        va_end(args);
    }

    bool allowLogEncoding = false;
    unsigned frameWriteCalls = 0;
    bool lastBestEffort = false;

  protected:
    /// Apply the configured log-encoding gate.
    bool canEncodeLogRecord() override { return allowLogEncoding; }

    /// Record whether the encoded log was marked best-effort.
    bool writeFrame(uint8_t *, size_t, bool bestEffort) override
    {
        frameWriteCalls++;
        lastBestEffort = bestEffort;
        return true;
    }
};

/// Assert byte-for-byte equality between expected and captured stream output.
static void assertBytesEqual(const std::vector<uint8_t> &expected, const std::vector<uint8_t> &actual)
{
    TEST_ASSERT_EQUAL_UINT(expected.size(), actual.size());
    TEST_ASSERT_EQUAL_UINT8_ARRAY(expected.data(), actual.data(), expected.size());
}

/// Verify retries append only the unwritten tail and reproduce the frame exactly once.
void test_frame_writer_continues_only_unwritten_tail()
{
    ScriptedStream stream;
    StreamFrameWriter writer;
    std::vector<uint8_t> frame = {0x94, 0xc3, 0x00, 0x06, 1, 2, 3, 4, 5, 6};
    stream.queueWrite(3);
    stream.queueWrite(2);
    stream.queueWrite(frame.size());

    TEST_ASSERT_FALSE(writer.writeFrame(stream, frame.data(), frame.size(), false));
    TEST_ASSERT_FALSE(writer.isIdle());
    TEST_ASSERT_FALSE(writer.finishPendingFrame(stream));
    TEST_ASSERT_FALSE(writer.isIdle());
    TEST_ASSERT_TRUE(writer.finishPendingFrame(stream));
    TEST_ASSERT_TRUE(writer.isIdle());

    std::vector<size_t> expectedRequests = {10, 7, 5};
    TEST_ASSERT_EQUAL_UINT(expectedRequests.size(), stream.requestedLengths.size());
    TEST_ASSERT_EQUAL_UINT64_ARRAY(expectedRequests.data(), stream.requestedLengths.data(), expectedRequests.size());
    assertBytesEqual(frame, stream.output);
    TEST_ASSERT_EQUAL_UINT(0, stream.flushCount);
}

/// Verify a replacement session receives a complete old frame before its new frame.
void test_frame_writer_completes_retained_tail_before_new_session_frame()
{
    ScriptedStream stream;
    StreamFrameWriter writer;
    std::vector<uint8_t> oldFrame = {0x94, 0xc3, 0x00, 0x03, 0xa1, 0xa2, 0xa3};
    std::vector<uint8_t> newFrame = {0x94, 0xc3, 0x00, 0x02, 0xb1, 0xb2};
    stream.queueWrite(3);
    stream.queueWrite(oldFrame.size());
    stream.queueWrite(newFrame.size());

    TEST_ASSERT_FALSE(writer.writeFrame(stream, oldFrame.data(), oldFrame.size(), false));
    TEST_ASSERT_FALSE(writer.isIdle());

    // A replacement client starts without discarding the accepted old prefix.
    TEST_ASSERT_TRUE(writer.writeFrame(stream, newFrame.data(), newFrame.size(), false));
    TEST_ASSERT_TRUE(writer.isIdle());

    std::vector<uint8_t> expected = oldFrame;
    expected.insert(expected.end(), newFrame.begin(), newFrame.end());
    assertBytesEqual(expected, stream.output);
}

/// Verify a required main frame remains ordered behind a partial log frame.
void test_frame_writer_defers_main_behind_partial_log()
{
    ScriptedStream stream;
    StreamFrameWriter writer;
    std::vector<uint8_t> logFrame = {0x94, 0xc3, 0x00, 0x02, 0xa1, 0xa2};
    std::vector<uint8_t> mainFrame = {0x94, 0xc3, 0x00, 0x03, 0xb1, 0xb2, 0xb3};
    stream.queueWrite(2);
    stream.queueWrite(0);
    stream.queueWrite(logFrame.size());
    stream.queueWrite(mainFrame.size());

    TEST_ASSERT_FALSE(writer.writeFrame(stream, logFrame.data(), logFrame.size(), true));
    TEST_ASSERT_FALSE(writer.isIdle());
    TEST_ASSERT_FALSE(writer.writeFrame(stream, mainFrame.data(), mainFrame.size(), false));
    TEST_ASSERT_FALSE(writer.isIdle());
    TEST_ASSERT_FALSE(writer.finishPendingFrame(stream));
    TEST_ASSERT_FALSE(writer.isIdle());
    TEST_ASSERT_TRUE(writer.finishPendingFrame(stream));
    TEST_ASSERT_TRUE(writer.isIdle());

    std::vector<uint8_t> expected = logFrame;
    expected.insert(expected.end(), mainFrame.begin(), mainFrame.end());
    assertBytesEqual(expected, stream.output);
    TEST_ASSERT_EQUAL_UINT(4, stream.requestedLengths.size());
    TEST_ASSERT_EQUAL_UINT(0, stream.flushCount);
}

/// Verify best-effort output starts only when the complete frame fits.
void test_frame_writer_rejects_best_effort_without_full_capacity()
{
    ScriptedStream stream;
    StreamFrameWriter writer;
    std::vector<uint8_t> frame = {0x94, 0xc3, 0x00, 0x02, 1, 2};
    stream.availableCapacity = frame.size() - 1;

    TEST_ASSERT_FALSE(writer.writeFrame(stream, frame.data(), frame.size(), true));
    TEST_ASSERT_TRUE(writer.isIdle());
    TEST_ASSERT_EQUAL_UINT(0, stream.requestedLengths.size());

    stream.availableCapacity = frame.size();
    TEST_ASSERT_TRUE(writer.writeFrame(stream, frame.data(), frame.size(), true));
    TEST_ASSERT_TRUE(writer.isIdle());
    TEST_ASSERT_EQUAL_UINT(1, stream.requestedLengths.size());
    assertBytesEqual(frame, stream.output);
}

/// Verify each zero-progress continuation makes one bounded write attempt.
void test_frame_writer_zero_progress_is_one_bounded_attempt()
{
    ScriptedStream stream;
    StreamFrameWriter writer;
    std::vector<uint8_t> frame = {0x94, 0xc3, 0x00, 0x02, 1, 2};
    stream.queueWrite(1);
    stream.queueWrite(0);
    stream.queueWrite(0);
    stream.queueWrite(frame.size());

    TEST_ASSERT_FALSE(writer.writeFrame(stream, frame.data(), frame.size(), false));
    TEST_ASSERT_EQUAL_UINT(1, stream.requestedLengths.size());
    TEST_ASSERT_FALSE(writer.finishPendingFrame(stream));
    TEST_ASSERT_EQUAL_UINT(2, stream.requestedLengths.size());
    TEST_ASSERT_FALSE(writer.finishPendingFrame(stream));
    TEST_ASSERT_EQUAL_UINT(3, stream.requestedLengths.size());
    TEST_ASSERT_TRUE(writer.finishPendingFrame(stream));
    TEST_ASSERT_EQUAL_UINT(4, stream.requestedLengths.size());
    assertBytesEqual(frame, stream.output);
}

/// Verify generic StreamAPI framing and successful-write flush behavior.
void test_stream_api_full_write_frames_and_flushes()
{
    ScopedMeshService scopedService;
    ScriptedStream stream;
    StreamAPITestShim api(&stream);
    uint8_t frame[7] = {0, 0, 0, 0, 0x11, 0x22, 0x33};

    TEST_ASSERT_TRUE(api.writeBaseFrame(frame, 3));

    std::vector<uint8_t> expected = {0x94, 0xc3, 0x00, 0x03, 0x11, 0x22, 0x33};
    assertBytesEqual(expected, stream.output);
    TEST_ASSERT_EQUAL_UINT(1, stream.requestedLengths.size());
    TEST_ASSERT_EQUAL_UINT(1, stream.flushCount);
    TEST_ASSERT_EQUAL_UINT(0, api.failureCalls);
}

/// Verify generic transports report short writes without flushing or retrying.
void test_stream_api_short_write_reports_failure_without_flush()
{
    ScopedMeshService scopedService;
    ScriptedStream stream;
    StreamAPITestShim api(&stream);
    uint8_t frame[7] = {0, 0, 0, 0, 0x11, 0x22, 0x33};
    stream.queueWrite(5);

    TEST_ASSERT_FALSE(api.writeBaseFrame(frame, 3));

    TEST_ASSERT_EQUAL_UINT(1, stream.requestedLengths.size());
    TEST_ASSERT_EQUAL_UINT(0, stream.flushCount);
    TEST_ASSERT_EQUAL_UINT(1, api.failureCalls);
    TEST_ASSERT_EQUAL_UINT(7, api.failedFrameLen);
    TEST_ASSERT_EQUAL_UINT(5, api.failedWrittenLen);
}

/// Verify retained output blocks PhoneAPI from dequeuing the next payload.
void test_stream_api_finishes_pending_before_advancing_phone_api()
{
    ScopedMeshService scopedService;
    ScriptedStream stream;
    StreamAPITestShim api(&stream);
    api.sendConfigComplete();
    api.sendNotification(meshtastic_LogRecord_Level_WARNING, 42, "still queued");
    api.finishReady = false;

    api.runOncePart(nullptr, 0);
    TEST_ASSERT_EQUAL_UINT(1, api.finishCalls);
    TEST_ASSERT_EQUAL_UINT(0, api.frameWriteCalls);

    api.finishReady = true;
    api.runOncePart(nullptr, 0);
    TEST_ASSERT_EQUAL_UINT(2, api.finishCalls);
    TEST_ASSERT_EQUAL_UINT(1, api.frameWriteCalls);

    meshtastic_FromRadio decoded = meshtastic_FromRadio_init_zero;
    TEST_ASSERT_TRUE(
        pb_decode_from_bytes(api.capturedPayload.data(), api.capturedPayload.size(), &meshtastic_FromRadio_msg, &decoded));
    TEST_ASSERT_EQUAL_UINT(meshtastic_FromRadio_clientNotification_tag, decoded.which_payload_variant);
    TEST_ASSERT_EQUAL_UINT32(42, decoded.clientNotification.reply_id);
}

/// Verify framed logs honor the encoding gate and use best-effort writes.
void test_stream_api_gates_logs_and_marks_them_best_effort()
{
    ScopedMeshService scopedService;
    ScriptedStream stream;
    LogHookStreamAPI api(&stream);

    api.emitTestLog("blocked %u", 1U);
    TEST_ASSERT_EQUAL_UINT(0, api.frameWriteCalls);

    api.allowLogEncoding = true;
    api.emitTestLog("allowed %u", 2U);
    TEST_ASSERT_EQUAL_UINT(1, api.frameWriteCalls);
    TEST_ASSERT_TRUE(api.lastBestEffort);
}

/// Verify the real SerialConsole emits no unframed bytes in protobuf mode.
void test_serial_console_suppresses_raw_output_in_protobuf_mode()
{
    RecordingPrint sink;
    const bool oldHasLora = config.has_lora;
    const bool oldHasSecurity = config.has_security;
    const bool oldSerialEnabled = config.security.serial_enabled;
    const bool oldDebugLogApiEnabled = config.security.debug_log_api_enabled;

    config.has_lora = true;
    config.has_security = true;
    config.security.serial_enabled = true;
    config.security.debug_log_api_enabled = false;
    console->setDestination(&sink);

    console->write('A');
    const bool rawBeforeProtobuf = sink.output.size() == 1 && sink.output[0] == 'A';
    sink.output.clear();

    const uint8_t emptyToRadio = 0;
    console->handleToRadio(&emptyToRadio, 0);
    console->write('B');
    console->write('\n');
    console->log(MESHTASTIC_LOG_LEVEL_ERROR, "must stay framed");
    const bool emptyAfterProtobuf = sink.output.empty();

    console->setDestination(&Serial);
    config.has_lora = oldHasLora;
    config.has_security = oldHasSecurity;
    config.security.serial_enabled = oldSerialEnabled;
    config.security.debug_log_api_enabled = oldDebugLogApiEnabled;

    TEST_ASSERT_TRUE(rawBeforeProtobuf);
    TEST_ASSERT_TRUE(emptyAfterProtobuf);
}

/// Unity per-test setup; fixtures are local to each test.
void setUp(void) {}
/// Unity per-test teardown; fixtures clean themselves up.
void tearDown(void) {}

/// Initialize the native environment and run the stream regression suite.
void setup()
{
    initializeTestEnvironment();
    UNITY_BEGIN();
    RUN_TEST(test_frame_writer_continues_only_unwritten_tail);
    RUN_TEST(test_frame_writer_completes_retained_tail_before_new_session_frame);
    RUN_TEST(test_frame_writer_defers_main_behind_partial_log);
    RUN_TEST(test_frame_writer_rejects_best_effort_without_full_capacity);
    RUN_TEST(test_frame_writer_zero_progress_is_one_bounded_attempt);
    RUN_TEST(test_stream_api_full_write_frames_and_flushes);
    RUN_TEST(test_stream_api_short_write_reports_failure_without_flush);
    RUN_TEST(test_stream_api_finishes_pending_before_advancing_phone_api);
    RUN_TEST(test_stream_api_gates_logs_and_marks_them_best_effort);
    // usingProtobufs intentionally has no reset path, so this must run last.
    RUN_TEST(test_serial_console_suppresses_raw_output_in_protobuf_mode);
    exit(UNITY_END());
}

/// Unused Arduino loop required by the native Unity runner.
void loop() {}