test_new_vs_census_semantics.py

"""验证 wifimap new (first_seen 视角) vs wifimap census (last_seen 视角) 的语义边界。

四种设备状态组合（按 first_seen 和 last_seen 相对窗口位置）:

    | first_seen 在窗口外 | first_seen 在窗口内
----|---------------------|---------------------
    | 老员工电脑回来       | 新客户刚来还在
last| 仅 census           | new + census
seen|---------------------|---------------------
在窗 | 老员工电脑离开       | 早上访客已走
口外 | 都没有              | 仅 new

注: 这些查询是 OR 关系，不是 AND，意思是两个视角看同一 DB.
"""
from datetime import datetime, timedelta

import pytest

from wifimap.census import aggregate_census
from wifimap.db import Database, init_schema
from wifimap.newdev import find_new_devices


# 用 fixed 时刻 "现在" = 2026-05-14 12:00:00 作为参考
NOW = datetime(2026, 5, 14, 12, 0, 0)


def _add_device(conn, mac, first_seen_dt, last_seen_dt, vendor="Apple"):
    """Helper: 用底层 INSERT 写设备，绕过 record_sighting 的 upsert 逻辑.

    record_sighting 会自动设 first_seen = last_seen = seen_at,
    我们要分别控制两个字段.
    """
    conn.execute(
        """INSERT INTO devices (mac, first_seen, last_seen, vendor, is_transient)
           VALUES (?, ?, ?, ?, 0)""",
        (
            mac.lower(),
            first_seen_dt.strftime("%Y-%m-%d %H:%M:%S"),
            last_seen_dt.strftime("%Y-%m-%d %H:%M:%S"),
            vendor,
        ),
    )
    conn.commit()


@pytest.fixture
def four_corners(memory_conn):
    """Seed 4 设备覆盖 first_seen / last_seen 的 4 个象限."""
    init_schema(memory_conn)

    # 老员工电脑 + 现在在公司
    _add_device(
        memory_conn, "aa:01:00:00:00:01",
        first_seen_dt=NOW - timedelta(days=5),
        last_seen_dt=NOW - timedelta(seconds=30),
    )
    # 老员工电脑 + 已下班离开
    _add_device(
        memory_conn, "aa:01:00:00:00:02",
        first_seen_dt=NOW - timedelta(days=5),
        last_seen_dt=NOW - timedelta(hours=8),
    )
    # 新客户刚来 + 还在
    _add_device(
        memory_conn, "aa:01:00:00:00:03",
        first_seen_dt=NOW - timedelta(hours=1),
        last_seen_dt=NOW - timedelta(seconds=30),
    )
    # 早上访客 + 已走
    _add_device(
        memory_conn, "aa:01:00:00:00:04",
        first_seen_dt=NOW - timedelta(hours=12),
        last_seen_dt=NOW - timedelta(hours=8),
    )
    return memory_conn


def test_new_24h_includes_first_seen_within_24h(four_corners):
    """wifimap new --range 24h: 应该列出 first_seen 在过去 24h 内的所有设备."""
    since = (NOW - timedelta(hours=24)).strftime("%Y-%m-%d %H:%M:%S")
    rows = find_new_devices(four_corners, since=since)
    macs = {r["mac"] for r in rows}
    # 设备 03 (1h 前 first_seen) 和 04 (12h 前 first_seen) 都在窗口内
    # 设备 01 02 在 5 天前，在窗口外
    assert macs == {"aa:01:00:00:00:03", "aa:01:00:00:00:04"}


def test_census_60min_includes_last_seen_within_60min(four_corners):
    """wifimap census --window 60: 应该列出 last_seen 在过去 60 分钟内的设备."""
    since = (NOW - timedelta(minutes=60)).strftime("%Y-%m-%d %H:%M:%S")
    result = aggregate_census(four_corners, since=since)
    # 收集所有出现在结果里的 mac
    devices_query = list(four_corners.execute(
        "SELECT mac FROM devices WHERE last_seen >= ? AND is_transient = 0",
        (since,),
    ))
    macs = {r["mac"] for r in devices_query}
    # 设备 01 (30s 前 last_seen) 和 03 (30s 前 last_seen) 在线
    # 设备 02 (8h 前) 和 04 (8h 前) 不在线
    assert macs == {"aa:01:00:00:00:01", "aa:01:00:00:00:03"}
    assert result.total == 2


def test_overlap_device_in_both_new_and_census(four_corners):
    """设备 03 (新客户刚来 + 还在) 应该同时出现在 new 和 census."""
    since_24h = (NOW - timedelta(hours=24)).strftime("%Y-%m-%d %H:%M:%S")
    since_60min = (NOW - timedelta(minutes=60)).strftime("%Y-%m-%d %H:%M:%S")

    new_macs = {r["mac"] for r in find_new_devices(four_corners, since=since_24h)}
    census_macs = {
        r["mac"] for r in four_corners.execute(
            "SELECT mac FROM devices WHERE last_seen >= ? AND is_transient = 0",
            (since_60min,),
        )
    }

    # 设备 03 既是新设备 (在 24h 内) 又当前在线 (60min 内被扫到)
    assert "aa:01:00:00:00:03" in new_macs
    assert "aa:01:00:00:00:03" in census_macs


def test_only_new_not_in_census(four_corners):
    """设备 04 (早上访客已走) 应该只在 new, 不在 census."""
    since_24h = (NOW - timedelta(hours=24)).strftime("%Y-%m-%d %H:%M:%S")
    since_60min = (NOW - timedelta(minutes=60)).strftime("%Y-%m-%d %H:%M:%S")

    new_macs = {r["mac"] for r in find_new_devices(four_corners, since=since_24h)}
    census_macs = {
        r["mac"] for r in four_corners.execute(
            "SELECT mac FROM devices WHERE last_seen >= ? AND is_transient = 0",
            (since_60min,),
        )
    }

    assert "aa:01:00:00:00:04" in new_macs       # 是新设备
    assert "aa:01:00:00:00:04" not in census_macs  # 但已离线


def test_only_census_not_in_new(four_corners):
    """设备 01 (老员工现在在公司) 应该只在 census, 不在 new."""
    since_24h = (NOW - timedelta(hours=24)).strftime("%Y-%m-%d %H:%M:%S")
    since_60min = (NOW - timedelta(minutes=60)).strftime("%Y-%m-%d %H:%M:%S")

    new_macs = {r["mac"] for r in find_new_devices(four_corners, since=since_24h)}
    census_macs = {
        r["mac"] for r in four_corners.execute(
            "SELECT mac FROM devices WHERE last_seen >= ? AND is_transient = 0",
            (since_60min,),
        )
    }

    assert "aa:01:00:00:00:01" not in new_macs   # 5 天前就有了，不是新
    assert "aa:01:00:00:00:01" in census_macs    # 但当前在线


def test_neither_new_nor_census(four_corners):
    """设备 02 (老员工已下班) 应该哪边都不在."""
    since_24h = (NOW - timedelta(hours=24)).strftime("%Y-%m-%d %H:%M:%S")
    since_60min = (NOW - timedelta(minutes=60)).strftime("%Y-%m-%d %H:%M:%S")

    new_macs = {r["mac"] for r in find_new_devices(four_corners, since=since_24h)}
    census_macs = {
        r["mac"] for r in four_corners.execute(
            "SELECT mac FROM devices WHERE last_seen >= ? AND is_transient = 0",
            (since_60min,),
        )
    }

    assert "aa:01:00:00:00:02" not in new_macs
    assert "aa:01:00:00:00:02" not in census_macs


def test_transient_excluded_from_new_by_default(memory_conn):
    """transient (随机 MAC) 默认不在 new 列表里."""
    init_schema(memory_conn)
    # 一个 transient 设备，刚出现
    memory_conn.execute(
        """INSERT INTO devices (mac, first_seen, last_seen, vendor, is_transient)
           VALUES (?, ?, ?, ?, 1)""",
        ("a6:11:22:33:44:55", NOW.strftime("%Y-%m-%d %H:%M:%S"),
         NOW.strftime("%Y-%m-%d %H:%M:%S"), None),
    )
    memory_conn.commit()

    since = (NOW - timedelta(hours=24)).strftime("%Y-%m-%d %H:%M:%S")
    rows = find_new_devices(memory_conn, since=since)
    assert rows == []

    rows_inc = find_new_devices(memory_conn, since=since, include_transient=True)
    assert len(rows_inc) == 1


def test_transient_excluded_from_census(memory_conn):
    """transient 默认不在 census 里."""
    init_schema(memory_conn)
    memory_conn.execute(
        """INSERT INTO devices (mac, first_seen, last_seen, vendor, is_transient)
           VALUES (?, ?, ?, ?, 1)""",
        ("a6:11:22:33:44:55", NOW.strftime("%Y-%m-%d %H:%M:%S"),
         NOW.strftime("%Y-%m-%d %H:%M:%S"), None),
    )
    memory_conn.commit()

    since = (NOW - timedelta(minutes=60)).strftime("%Y-%m-%d %H:%M:%S")
    result = aggregate_census(memory_conn, since=since)
    assert result.total == 0


def test_edge_case_device_exactly_at_boundary(memory_conn):
    """边界: first_seen == since 应该被 new 包含 (>=)."""
    init_schema(memory_conn)
    boundary = NOW - timedelta(hours=24)
    _add_device(
        memory_conn, "aa:01:00:00:00:99",
        first_seen_dt=boundary,
        last_seen_dt=boundary,
    )
    since = boundary.strftime("%Y-%m-%d %H:%M:%S")
    rows = find_new_devices(memory_conn, since=since)
    assert len(rows) == 1
    assert rows[0]["mac"] == "aa:01:00:00:00:99"


def test_range_filters_complete():
    """所有支持的 range 字符串都能解析."""
    from wifimap._cli_helpers import parse_time_range

    for s in ("today", "week", "month", "all", "24h", "7d", "30d"):
        start, label = parse_time_range(s)
        assert isinstance(start, datetime)
        assert label  # non-empty