bigbrother/net_alerter/test_net_alerter.py

#!/usr/bin/env python3
"""
Test suite for net_alerter.py — alert formatting fixes and false positive suppression.
"""

import sys
import threading
import time
from pathlib import Path
from unittest.mock import patch, MagicMock
import pytest

# Add parent dir to path so we can import net_alerter
sys.path.insert(0, str(Path(__file__).parent))

import net_alerter



def cleanup_flap_state():
    """Reset flap detection state between tests."""
    net_alerter._flap_window.clear()
    net_alerter._interface_recovering = False
    if net_alerter._recovery_timer:
        net_alerter._recovery_timer.cancel()
        net_alerter._recovery_timer = None


@pytest.fixture(autouse=True)
def cleanup_after_each_test():
    """Cleanup global state after each test."""
    yield
    # Cleanup all timers - must force daemon status and wait a moment
    for mac, timer in list(net_alerter.departure_timers.items()):
        if timer.is_alive():
            timer.cancel()
        # Wait a tiny bit for timer thread to notice cancellation
        time.sleep(0.01)
    net_alerter.departure_timers.clear()
    net_alerter.known_devices.clear()
    net_alerter.last_departed_time.clear()
    net_alerter._churn_tracker.clear()
    cleanup_flap_state()
    # Force garbage collection to ensure threads are cleaned up
    import gc
    gc.collect()

def test_hostname_dedup_when_hostname_equals_ip():
    """Test that hostname is not repeated when DNS fails (hostname == IP)."""
    # Simulate a device where hostname lookup returned the IP address itself
    mac = "88:a2:9e:8e:f2:90"
    ip = "10.0.0.0"
    hostname = ip  # This is what happens when reverse DNS fails

    # Format the alert using internal logic
    msg = net_alerter.format_arrival(hostname, ip, "Apple", mac)

    # Should show IP only once
    assert msg == "[NET] ARRIVED: 10.0.0.0 [Apple] MAC:88:a2:9e:8e:f2:90"
    assert msg.count("10.0.0.0") == 1, "IP should appear exactly once"


def test_hostname_shown_when_different_from_ip():
    """Test that hostname is shown separately when it differs from IP."""
    cleanup_flap_state()
    mac = "88:a2:9e:8e:f2:90"
    ip = "10.0.0.0"
    hostname = "mydevice.local"

    # Format the alert
    msg = net_alerter.format_arrival(hostname, ip, "Apple", mac)

    # Should show both hostname and IP
    assert msg == "[NET] ARRIVED: mydevice.local (10.0.0.0) [Apple] MAC:88:a2:9e:8e:f2:90"
    assert "mydevice.local" in msg
    assert "10.0.0.0" in msg


def test_oui_lookup_with_inline_dict():
    """Test that OUI lookup returns vendor for known MACs."""
    cleanup_flap_state()
    # These are real OUI prefixes
    test_cases = [
        ("88:a2:9e", "Apple"),  # Apple
        ("00:1a:7d", "Apple"),  # Apple
        ("18:5f:3f", "Apple"),  # Apple
        ("a4:c3:f0", "Apple"),  # Apple
        ("28:87:ba", "Google"),  # Google
        ("54:27:58", "Google"),  # Google
        ("34:15:13", "Amazon"),  # Amazon
        ("b8:27:eb", "Raspberry Pi"),  # Raspberry Pi
    ]

    for mac_prefix, expected_vendor in test_cases:
        vendor = net_alerter.lookup_oui(mac_prefix + ":00:00:00")
        assert vendor == expected_vendor, f"Expected {expected_vendor} for {mac_prefix}, got {vendor}"


def test_oui_lookup_unknown_vendor():
    """Test that unknown OUI returns 'unknown'."""
    cleanup_flap_state()
    vendor = net_alerter.lookup_oui("aa:bb:cc:dd:ee:ff")
    assert vendor == "unknown", f"Expected 'unknown' for unknown OUI, got {vendor}"


def test_departure_format_dedup():
    """Test that departure message also deduplicates hostname and IP."""
    cleanup_flap_state()
    mac = "88:a2:9e:8e:f2:90"
    ip = "10.0.0.0"
    hostname = ip
    duration = "5m 30s"

    msg = net_alerter.format_departure(hostname, ip, "Apple", mac, duration)
    assert msg == "[NET] DEPARTED: 10.0.0.0 [Apple] MAC:88:a2:9e:8e:f2:90 — present 5m 30s"
    assert msg.count("10.0.0.0") == 1


def test_infrastructure_ips_never_alert():
    """Test that infrastructure IPs (gateway, broadcast, self) don't trigger alerts."""
    cleanup_flap_state()
    net_alerter.known_devices.clear()
    net_alerter.infrastructure_ips.add("10.0.0.0")  # gateway

    alert_calls = []
    original_send_alert = net_alerter.send_alert
    net_alerter.send_alert = lambda msg: alert_calls.append(msg)

    try:
        mac = "aa:bb:cc:dd:ee:ff"
        ip = "10.0.0.0"

        net_alerter.on_arrival(mac, ip, "gateway")

        assert len(alert_calls) == 0, f"Infrastructure IP should not alert, but got: {alert_calls}"
    finally:
        net_alerter.send_alert = original_send_alert


def test_departure_debounce_15min():
    """Test that departure alerts are debounced for 15 minutes."""
    cleanup_flap_state()
    net_alerter.known_devices.clear()
    net_alerter.departure_timers.clear()

    alert_calls = []
    original_send_alert = net_alerter.send_alert
    net_alerter.send_alert = lambda msg: alert_calls.append(msg)

    try:
        mac = "aa:bb:cc:dd:ee:ff"

        net_alerter.known_devices[mac] = {
            "ip": "10.0.0.0",
            "hostname": "testhost",
            "vendor": "Test",
            "first_seen": time.time(),
            "last_seen": time.time()
        }

        net_alerter.on_departure(mac)

        assert len(alert_calls) == 0, f"Departure should be debounced, but got alert: {alert_calls}"

        assert mac in net_alerter.departure_timers, "Departure timer should be created"

        net_alerter.departure_timers[mac].cancel()
        del net_alerter.departure_timers[mac]
    finally:
        net_alerter.send_alert = original_send_alert


def test_re_arrival_within_5min_suppresses_alert():
    """Test that device re-arrival within 5 min of departure suppresses ARRIVED alert."""
    cleanup_flap_state()
    net_alerter.known_devices.clear()
    net_alerter.departure_timers.clear()
    net_alerter.last_departed_time.clear()

    alert_calls = []
    original_send_alert = net_alerter.send_alert
    net_alerter.send_alert = lambda msg: alert_calls.append(msg)

    try:
        mac = "aa:bb:cc:dd:ee:ff"
        ip = "10.0.0.0"

        net_alerter.known_devices[mac] = {
            "ip": ip,
            "hostname": "testhost",
            "vendor": "Test",
            "first_seen": time.time(),
            "last_seen": time.time()
        }

        net_alerter.on_departure(mac)
        net_alerter.last_departed_time[mac] = time.time()
        net_alerter.on_arrival(mac, ip, "testhost")

        # Check that ARRIVED alerts are suppressed (occupancy alerts are allowed)
        arrived_alerts = [a for a in alert_calls if "ARRIVED" in a]
        assert len(arrived_alerts) == 0, f"Re-arrival within 5 min should be suppressed, but got: {arrived_alerts}"

        if mac in net_alerter.departure_timers:
            net_alerter.departure_timers[mac].cancel()
    finally:
        net_alerter.send_alert = original_send_alert


def test_dhcp_renewal_dedup_30min():
    """Test that DHCP renewal from known device (< 30 min old) doesn't send ARRIVED alert."""
    cleanup_flap_state()
    net_alerter.known_devices.clear()
    net_alerter.departure_timers.clear()

    alert_calls = []
    original_send_alert = net_alerter.send_alert
    net_alerter.send_alert = lambda msg: alert_calls.append(msg)

    try:
        mac = "aa:bb:cc:dd:ee:ff"
        ip = "10.0.0.0"

        now = time.time()
        net_alerter.known_devices[mac] = {
            "ip": ip,
            "hostname": "testhost",
            "vendor": "Test",
            "first_seen": now - 600,
            "last_seen": now - 100,
        }

        net_alerter.on_arrival(mac, ip, "testhost")

        # Check that ARRIVED alerts are suppressed (occupancy alerts are allowed)
        arrived_alerts = [a for a in alert_calls if "ARRIVED" in a]
        assert len(arrived_alerts) == 0, f"DHCP renewal < 30 min should be deduped, but got: {arrived_alerts}"
    finally:
        net_alerter.send_alert = original_send_alert
        for timer in list(net_alerter.departure_timers.values()):
            timer.cancel()
        net_alerter.departure_timers.clear()


def test_re_arrival_cancels_departure_timer():
    """Regression test: device departs → timer starts → device re-arrives BEFORE timer fires."""
    cleanup_flap_state()
    net_alerter.known_devices.clear()
    net_alerter.departure_timers.clear()
    net_alerter.last_departed_time.clear()

    alert_calls = []
    original_send_alert = net_alerter.send_alert
    net_alerter.send_alert = lambda msg: alert_calls.append(msg)

    try:
        mac = "aa:bb:cc:dd:ee:ff"
        ip = "10.0.0.0"

        now = time.time()
        net_alerter.known_devices[mac] = {
            "ip": ip,
            "hostname": "testhost",
            "vendor": "Test",
            "first_seen": now,
            "last_seen": now
        }

        net_alerter.on_departure(mac)

        assert mac in net_alerter.departure_timers, "Departure timer should be created"
        timer_ref = net_alerter.departure_timers[mac]

        net_alerter.on_arrival(mac, ip, "testhost")

        assert mac not in net_alerter.departure_timers or not timer_ref.is_alive(), \
            "Departure timer should be cancelled after re-arrival"

        departure_alerts = [a for a in alert_calls if "DEPARTED" in a]
        assert len(departure_alerts) == 0, \
            f"No departure alert should be sent when device re-arrives before timer expires, got: {departure_alerts}"

    finally:
        for timer in net_alerter.departure_timers.values():
            timer.cancel()
        net_alerter.departure_timers.clear()
        net_alerter.send_alert = original_send_alert


def test_rapid_flap_no_duplicate_arrived_alerts():
    """Test rapid RTM_NEWNEIGH events don't generate duplicate ARRIVED alerts."""
    cleanup_flap_state()
    net_alerter.known_devices.clear()
    net_alerter.departure_timers.clear()
    net_alerter.last_departed_time.clear()

    alert_calls = []
    original_send_alert = net_alerter.send_alert
    net_alerter.send_alert = lambda msg: alert_calls.append(msg)

    try:
        mac = "aa:bb:cc:dd:ee:ff"
        ip = "10.0.0.0"

        now = time.time()
        net_alerter.known_devices[mac] = {
            "ip": ip,
            "hostname": "testhost",
            "vendor": "Test",
            "first_seen": now,
            "last_seen": now
        }

        alert_calls.clear()

        net_alerter.on_departure(mac)

        assert len([a for a in alert_calls if "DEPARTED" in a]) == 0

        for i in range(5):
            net_alerter.on_arrival(mac, ip, "testhost")

        arrived_alerts = [a for a in alert_calls if "ARRIVED" in a]
        assert len(arrived_alerts) == 0, \
            f"Rapid re-arrivals should not generate duplicate ARRIVED alerts, but got {len(arrived_alerts)}: {arrived_alerts}"

    finally:
        for timer in net_alerter.departure_timers.values():
            timer.cancel()
        net_alerter.departure_timers.clear()
        net_alerter.send_alert = original_send_alert


def test_personal_device_oui_detection():
    """Test that Apple/Samsung/etc. OUI prefixes are detected as personal devices."""
    cleanup_flap_state()
    net_alerter.personal_devices.clear()

    test_cases = [
        ("88:a2:9e:ff:ff:ff", True),
        ("28:87:6d:ff:ff:ff", True),
        ("00:16:6b:ff:ff:ff", True),
        ("aa:bb:cc:dd:ee:ff", False),
    ]

    for mac, expected_is_personal in test_cases:
        result = net_alerter.is_personal_device(mac)
        assert result == expected_is_personal, f"Expected {expected_is_personal} for {mac}, got {result}"


def test_manual_personal_mac_config():
    """Test that NET_ALERTER_PERSONAL_MACS env var is parsed and devices are detected."""
    cleanup_flap_state()
    net_alerter.personal_devices.clear()

    import os
    original_env = os.environ.get("NET_ALERTER_PERSONAL_MACS")
    try:
        os.environ["NET_ALERTER_PERSONAL_MACS"] = "AA:BB:CC:DD:EE:FF, 11:22:33:44:55:66"
        net_alerter.load_personal_macs_from_env()

        assert "AABBCCDDEEFF" in net_alerter.personal_devices
        assert "112233445566" in net_alerter.personal_devices

        assert net_alerter.is_personal_device("aa:bb:cc:dd:ee:ff")
        assert net_alerter.is_personal_device("11:22:33:44:55:66")
    finally:
        net_alerter.personal_devices.clear()
        if original_env is not None:
            os.environ["NET_ALERTER_PERSONAL_MACS"] = original_env
        else:
            os.environ.pop("NET_ALERTER_PERSONAL_MACS", None)



def test_occupancy_vacant_to_occupied():
    """Test that VACANT → OCCUPIED transition fires occupancy alert on personal device arrival."""
    cleanup_flap_state()
    net_alerter.known_devices.clear()
    net_alerter.personal_devices.clear()
    net_alerter.location_occupancy = "UNKNOWN"

    alert_calls = []
    original_send_alert = net_alerter.send_alert
    net_alerter.send_alert = lambda msg: alert_calls.append(msg)

    try:
        net_alerter.location_occupancy = "VACANT"

        mac = "88:a2:9e:ff:ff:ff"
        ip = "10.0.0.0"
        net_alerter.on_arrival(mac, ip, "myphone")

        occupancy_alerts = [a for a in alert_calls if "Location:" in a]
        assert any("OCCUPIED" in a for a in occupancy_alerts), \
            f"Expected OCCUPIED alert on personal device arrival, got: {alert_calls}"
        assert net_alerter.location_occupancy == "OCCUPIED"
    finally:
        net_alerter.send_alert = original_send_alert


def test_occupancy_occupied_to_vacant():
    """Test that OCCUPIED → VACANT transition fires vacancy alert when last personal device departs."""
    cleanup_flap_state()
    net_alerter.known_devices.clear()
    net_alerter.personal_devices.clear()
    net_alerter.departure_timers.clear()
    net_alerter.location_occupancy = "UNKNOWN"

    alert_calls = []
    original_send_alert = net_alerter.send_alert
    net_alerter.send_alert = lambda msg: alert_calls.append(msg)

    try:
        mac = "88:a2:9e:ff:ff:ff"
        ip = "10.0.0.0"
        now = time.time()
        net_alerter.known_devices[mac] = {
            "ip": ip,
            "hostname": "myphone",
            "vendor": "Apple",
            "first_seen": now,
            "last_seen": now
        }
        net_alerter.location_occupancy = "OCCUPIED"

        alert_calls.clear()

        net_alerter.known_devices.pop(mac, None)
        net_alerter.update_occupancy_state()

        occupancy_alerts = [a for a in alert_calls if "Location:" in a]
        assert any("VACANT" in a for a in occupancy_alerts), \
            f"Expected VACANT alert when last personal device departs, got: {alert_calls}"
        assert net_alerter.location_occupancy == "VACANT"
    finally:
        net_alerter.send_alert = original_send_alert
        for timer in net_alerter.departure_timers.values():
            timer.cancel()
        net_alerter.departure_timers.clear()


def test_occupancy_multiple_personal_devices():
    """Test that VACANT only when ALL personal devices are gone (not just one)."""
    cleanup_flap_state()
    net_alerter.known_devices.clear()
    net_alerter.personal_devices.clear()
    net_alerter.location_occupancy = "UNKNOWN"

    alert_calls = []
    original_send_alert = net_alerter.send_alert
    net_alerter.send_alert = lambda msg: alert_calls.append(msg)

    try:
        mac1 = "88:a2:9e:ff:ff:ff"
        mac2 = "28:87:6d:ff:ff:ff"
        now = time.time()

        net_alerter.location_occupancy = "VACANT"
        alert_calls.clear()

        net_alerter.on_arrival(mac1, "10.0.0.0", "phone1")
        occupancy_alerts = [a for a in alert_calls if "Location:" in a]
        assert any("OCCUPIED" in a for a in occupancy_alerts), "Should be OCCUPIED after first device"
        assert net_alerter.location_occupancy == "OCCUPIED"

        alert_calls.clear()

        net_alerter.on_arrival(mac2, "10.0.0.0", "phone2")
        occupancy_alerts = [a for a in alert_calls if "Location:" in a]
        assert len(occupancy_alerts) == 0, "Should not send occupancy alert when already OCCUPIED"
        assert net_alerter.location_occupancy == "OCCUPIED"

        alert_calls.clear()

        net_alerter.on_departure(mac1)
        net_alerter.known_devices[mac1]['departing'] = True

        net_alerter.update_occupancy_state()
        occupancy_alerts = [a for a in alert_calls if "Location:" in a]
        assert len(occupancy_alerts) == 0, "Should stay OCCUPIED when one device still present"

        alert_calls.clear()

        net_alerter.known_devices.pop(mac1, None)
        net_alerter.known_devices.pop(mac2, None)
        net_alerter.update_occupancy_state()

        occupancy_alerts = [a for a in alert_calls if "Location:" in a]
        assert any("VACANT" in a for a in occupancy_alerts), "Should be VACANT when all devices gone"
        assert net_alerter.location_occupancy == "VACANT"
    finally:
        net_alerter.send_alert = original_send_alert
        for timer in net_alerter.departure_timers.values():
            timer.cancel()
        net_alerter.departure_timers.clear()


if __name__ == "__main__":
    test_hostname_dedup_when_hostname_equals_ip()
    print("✓ test_hostname_dedup_when_hostname_equals_ip")

    test_hostname_shown_when_different_from_ip()
    print("✓ test_hostname_shown_when_different_from_ip")

    test_oui_lookup_with_inline_dict()
    print("✓ test_oui_lookup_with_inline_dict")

    test_oui_lookup_unknown_vendor()
    print("✓ test_oui_lookup_unknown_vendor")

    test_departure_format_dedup()
    print("✓ test_departure_format_dedup")

    test_infrastructure_ips_never_alert()
    print("✓ test_infrastructure_ips_never_alert")

    test_departure_debounce_15min()
    print("✓ test_departure_debounce_15min")

    test_re_arrival_within_5min_suppresses_alert()
    print("✓ test_re_arrival_within_5min_suppresses_alert")

    test_dhcp_renewal_dedup_30min()
    print("✓ test_dhcp_renewal_dedup_30min")

    test_re_arrival_cancels_departure_timer()
    print("✓ test_re_arrival_cancels_departure_timer")

    test_rapid_flap_no_duplicate_arrived_alerts()
    print("✓ test_rapid_flap_no_duplicate_arrived_alerts")

    test_personal_device_oui_detection()
    print("✓ test_personal_device_oui_detection")

    test_manual_personal_mac_config()
    print("✓ test_manual_personal_mac_config")

    test_occupancy_vacant_to_occupied()
    print("✓ test_occupancy_vacant_to_occupied")

    test_occupancy_occupied_to_vacant()
    print("✓ test_occupancy_occupied_to_vacant")

    test_occupancy_multiple_personal_devices()
    print("✓ test_occupancy_multiple_personal_devices")

    print("\nAll tests passed!")