Commit 89d4b346 authored by Waleed Akbar's avatar Waleed Akbar
Browse files

Update .gitignore to exclude coverage directory and add unit tests for... 

Update .gitignore to exclude coverage directory and add unit tests for real-device streaming validation (NET CONF collector)
parent 8f00c312

.gitignore

+1 −0
Original line number Diff line number Diff line
@@ -174,6 +174,7 @@ cython_debug/ 
# Other

/tmp

.github

coverage/



# Sqlite

*.db

src/telemetry/backend/tests/netconf_oc/test_unit_NetconfOCcollector.py

+51 −0
Original line number Diff line number Diff line
@@ -205,3 +205,54 @@ def test_subscribe_get_state_unsub(connected_collector): 


    # Cleanup: job has expired by now; call unsub without asserting True

    col.UnsubscribeState(sub_id)





# ===========================================================================

# Test #3 -- Real-device streaming validation (skipped when USE_REAL_DEVICE=False)

#

# Connects to the real NETCONF device, subscribes for both supported KPIs

# on _CHANNEL, and collects a short burst of samples (interval=5 s, duration=20 s).

# Validates that at least one sample arrives per KPI and that each value is

# a finite float within a physically plausible range.

# ===========================================================================

@pytest.mark.skipif(not USE_REAL_DEVICE, reason="Requires a real NETCONF device (USE_REAL_DEVICE=False)")

def test_real_device_stream(connected_collector):

    import time as _time

    import math

    col = connected_collector



    LOGGER.info("[REAL] Connected to %s:%s", _REAL_ADDRESS, _REAL_PORT)



    sub_power = "sub-real-power"

    t0        = _time.time()



    results = col.SubscribeState([

        (sub_power, {"endpoint": _CHANNEL, "kpi": int(KPI.RECEIVED_POWER)}, 20.0, 5.0),

    ])

    assert results == [True], f"SubscribeState failed: {results}"



    power_samples = []



    for ts, key, val in col.GetState(duration=20, blocking=True):

        elapsed = _time.time() - t0

        LOGGER.info(

            "[REAL][GET] t=+%5.1fs  channel=%s  val=%.6g  ts=%.3f",

            elapsed, key, val, ts,

        )

        power_samples.append(val)



    elapsed_total = _time.time() - t0

    LOGGER.info(

        "[REAL][DONE] elapsed=%.1fs  power_samples=%d",

        elapsed_total, len(power_samples),

    )



    # At least one input-power sample must have arrived

    assert len(power_samples) >= 1, "No input-power samples received from device"



    # input-power: must be a finite float (typical range -40 .. 0 dBm)

    for v in power_samples:

        assert isinstance(v, float), f"Power value is not float: {v!r}"

        assert math.isfinite(v),     f"Power value is not finite: {v}"



    col.UnsubscribeState(sub_power)