188 lines
7.6 KiB
Python
188 lines
7.6 KiB
Python
"""Tests for Balabit Mouse Dynamics Challenge data adapter."""
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
def test_module_exports():
|
|
"""The adapter module must export the public functions used by CLI."""
|
|
from ai_mouse.data_adapters import balabit
|
|
assert hasattr(balabit, "parse_session_csv")
|
|
assert hasattr(balabit, "segment_by_clicks")
|
|
assert hasattr(balabit, "filter_segments")
|
|
assert hasattr(balabit, "process_session")
|
|
assert hasattr(balabit, "MouseEvent")
|
|
assert hasattr(balabit, "Segment")
|
|
|
|
|
|
def test_mouse_event_dataclass():
|
|
"""MouseEvent has expected fields."""
|
|
from ai_mouse.data_adapters.balabit import MouseEvent
|
|
e = MouseEvent(t_ms=100, button="NoButton", state="Move", x=300, y=400)
|
|
assert e.t_ms == 100
|
|
assert e.state == "Move"
|
|
assert e.x == 300
|
|
|
|
|
|
def test_segment_dataclass():
|
|
"""Segment has expected fields."""
|
|
from ai_mouse.data_adapters.balabit import MouseEvent, Segment
|
|
events = [MouseEvent(t_ms=0, button="NoButton", state="Move", x=10, y=20)]
|
|
s = Segment(events=events, click_x=100, click_y=200, click_t_ms=500, session_id="user1_s1")
|
|
assert s.events == events
|
|
assert s.click_x == 100
|
|
assert s.session_id == "user1_s1"
|
|
|
|
|
|
def _write_csv(path: Path, rows: list[str]) -> None:
|
|
"""Helper to write a Balabit-format CSV with header."""
|
|
header = "record timestamp,client timestamp,button,state,x,y"
|
|
path.write_text(header + "\n" + "\n".join(rows) + "\n", encoding="utf-8")
|
|
|
|
|
|
class TestParseSessionCsv:
|
|
def test_parses_basic_rows(self, tmp_path):
|
|
from ai_mouse.data_adapters.balabit import parse_session_csv
|
|
csv = tmp_path / "session_1"
|
|
_write_csv(csv, [
|
|
"1500000000.000,0.000,NoButton,Move,100,200",
|
|
"1500000000.050,0.050,NoButton,Move,110,210",
|
|
"1500000000.100,0.100,Left,Pressed,120,220",
|
|
])
|
|
events = parse_session_csv(csv)
|
|
assert len(events) == 3
|
|
assert events[0].t_ms == 0
|
|
assert events[0].state == "Move"
|
|
assert events[0].x == 100
|
|
assert events[2].t_ms == 100
|
|
assert events[2].state == "Pressed"
|
|
assert events[2].button == "Left"
|
|
|
|
def test_handles_float_timestamps(self, tmp_path):
|
|
"""Client timestamps are floats in seconds; we convert to int ms."""
|
|
from ai_mouse.data_adapters.balabit import parse_session_csv
|
|
csv = tmp_path / "session_2"
|
|
_write_csv(csv, [
|
|
"0,1.234,NoButton,Move,50,60",
|
|
"0,1.250,NoButton,Move,55,65",
|
|
])
|
|
events = parse_session_csv(csv)
|
|
assert events[0].t_ms == 1234
|
|
assert events[1].t_ms == 1250
|
|
|
|
def test_skips_malformed_rows(self, tmp_path):
|
|
"""Rows with bad data are logged and skipped, not raised."""
|
|
from ai_mouse.data_adapters.balabit import parse_session_csv
|
|
csv = tmp_path / "session_3"
|
|
_write_csv(csv, [
|
|
"0,0.000,NoButton,Move,100,200",
|
|
"BROKEN_ROW",
|
|
"0,abc,NoButton,Move,100,200", # bad timestamp
|
|
"0,0.100,NoButton,Move,150,250",
|
|
])
|
|
events = parse_session_csv(csv)
|
|
assert len(events) == 2
|
|
assert events[0].x == 100
|
|
assert events[1].x == 150
|
|
|
|
def test_returns_empty_list_for_empty_file(self, tmp_path):
|
|
from ai_mouse.data_adapters.balabit import parse_session_csv
|
|
csv = tmp_path / "session_4"
|
|
csv.write_text("record timestamp,client timestamp,button,state,x,y\n", encoding="utf-8")
|
|
events = parse_session_csv(csv)
|
|
assert events == []
|
|
|
|
|
|
class TestSegmentByClicks:
|
|
def _make_event(self, t_ms: int, state: str, x: int, y: int, button: str = "NoButton"):
|
|
from ai_mouse.data_adapters.balabit import MouseEvent
|
|
return MouseEvent(t_ms=t_ms, button=button, state=state, x=x, y=y)
|
|
|
|
def test_one_click_one_segment(self):
|
|
from ai_mouse.data_adapters.balabit import segment_by_clicks
|
|
events = [
|
|
self._make_event(0, "Move", 10, 20),
|
|
self._make_event(100, "Move", 50, 60),
|
|
self._make_event(500, "Move", 100, 100),
|
|
self._make_event(600, "Pressed", 110, 110, button="Left"),
|
|
]
|
|
segments = segment_by_clicks(events, window_ms=1200, session_id="test_s1")
|
|
assert len(segments) == 1
|
|
seg = segments[0]
|
|
assert seg.click_x == 110
|
|
assert seg.click_y == 110
|
|
assert seg.click_t_ms == 600
|
|
assert len(seg.events) == 3
|
|
assert seg.session_id == "test_s1"
|
|
|
|
def test_window_excludes_old_events(self):
|
|
"""Move events earlier than (click_t - window_ms) are dropped."""
|
|
from ai_mouse.data_adapters.balabit import segment_by_clicks
|
|
events = [
|
|
self._make_event(0, "Move", 10, 20), # too old
|
|
self._make_event(100, "Move", 20, 30), # too old
|
|
self._make_event(900, "Move", 30, 40), # in window
|
|
self._make_event(1000, "Pressed", 40, 50, button="Left"),
|
|
]
|
|
segments = segment_by_clicks(events, window_ms=200, session_id="s")
|
|
assert len(segments) == 1
|
|
assert len(segments[0].events) == 1
|
|
assert segments[0].events[0].t_ms == 900
|
|
|
|
def test_multiple_clicks_multiple_segments(self):
|
|
from ai_mouse.data_adapters.balabit import segment_by_clicks
|
|
events = [
|
|
self._make_event(0, "Move", 10, 20),
|
|
self._make_event(100, "Pressed", 50, 50, button="Left"),
|
|
self._make_event(200, "Released", 50, 50, button="Left"),
|
|
self._make_event(300, "Move", 60, 60),
|
|
self._make_event(400, "Move", 70, 70),
|
|
self._make_event(500, "Pressed", 80, 80, button="Left"),
|
|
]
|
|
segments = segment_by_clicks(events, window_ms=1200, session_id="s")
|
|
assert len(segments) == 2
|
|
assert segments[0].click_x == 50
|
|
assert segments[1].click_x == 80
|
|
# Second segment's events must not include the first Pressed
|
|
for e in segments[1].events:
|
|
assert e.state == "Move"
|
|
|
|
def test_skips_pressed_with_non_left_button(self):
|
|
"""Right-clicks and wheel-clicks don't anchor segments (only Left)."""
|
|
from ai_mouse.data_adapters.balabit import segment_by_clicks
|
|
events = [
|
|
self._make_event(0, "Move", 10, 20),
|
|
self._make_event(100, "Pressed", 50, 50, button="Right"), # ignored
|
|
self._make_event(200, "Move", 60, 60),
|
|
self._make_event(300, "Pressed", 70, 70, button="Left"), # anchor
|
|
]
|
|
segments = segment_by_clicks(events, window_ms=1200, session_id="s")
|
|
assert len(segments) == 1
|
|
assert segments[0].click_x == 70
|
|
|
|
def test_no_clicks_returns_empty(self):
|
|
from ai_mouse.data_adapters.balabit import segment_by_clicks
|
|
events = [
|
|
self._make_event(0, "Move", 10, 20),
|
|
self._make_event(100, "Move", 20, 30),
|
|
]
|
|
segments = segment_by_clicks(events, window_ms=1200, session_id="s")
|
|
assert segments == []
|
|
|
|
def test_excludes_drag_events(self):
|
|
"""Drag events are not Move; segment should only include Move."""
|
|
from ai_mouse.data_adapters.balabit import segment_by_clicks
|
|
events = [
|
|
self._make_event(0, "Move", 10, 20),
|
|
self._make_event(100, "Drag", 30, 40), # not Move
|
|
self._make_event(200, "Move", 50, 60),
|
|
self._make_event(300, "Pressed", 70, 80, button="Left"),
|
|
]
|
|
segments = segment_by_clicks(events, window_ms=1200, session_id="s")
|
|
assert len(segments) == 1
|
|
# Drag event should not appear in seg.events
|
|
assert all(e.state == "Move" for e in segments[0].events)
|
|
assert len(segments[0].events) == 2
|