saphid--frame-control/tests/test_tracking.py

"""Tracking protocols and fake-Frame BlueZ lifecycle; no headset or strap needed."""
import importlib.util
import math
import os
from pathlib import Path
import socket
import stat
import struct
import tempfile
import unittest
from unittest.mock import Mock, patch

SPEC = importlib.util.spec_from_file_location("tracking", Path(__file__).resolve().parents[1] / "frame/tracking/tracking.py")
t = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(t)


class Protocols(unittest.TestCase):
    def test_hrs_formats(self):
        self.assertEqual(t.heart_rate(b"\x00\x48"), 72)
        self.assertEqual(t.heart_rate(b"\x01\x2c\x01"), 300)
        self.assertEqual(t.heart_rate(b"\x1e\x48\x01\x00\x00\x04\x00\x04"), 72)
        self.assertEqual(t.heart_rate(b"\x02\x48"), 72)  # contact not supported
        self.assertIsNone(t.heart_rate(b"\x04\x48"))  # no contact
        self.assertIsNone(t.heart_rate(b"\x00\x00"))

    def test_hrs_malformed(self):
        for packet in (b"", b"\x00", b"\x01\x48", b"\x08\x48\x00", b"\x10\x48",
                       b"\x10\x48\x00", b"\x00\x48\x01", b"\xe0\x48"):
            with self.subTest(packet=packet), self.assertRaises(ValueError):
                t.heart_rate(packet)

    def test_gaze_coordinates(self):
        self.assertEqual(t.gaze_angles([0, 0, 0, 1]), (0, 0))
        angle = math.radians(15)
        pitch, yaw = t.gaze_angles([math.sin(angle), 0, 0, math.cos(angle)])
        self.assertAlmostEqual(pitch, -30)  # OpenXR +X rotation looks up
        self.assertAlmostEqual(yaw, 0)
        pitch, yaw = t.gaze_angles([0, -math.sin(angle), 0, math.cos(angle)])
        self.assertAlmostEqual(pitch, 0)
        self.assertAlmostEqual(yaw, 30)  # right

    def test_invalid_gaze(self):
        for pose in ([0, 0, 0, 0], [math.nan, 0, 0, 1], [0, 0, 0], [0, 0, 0, math.inf]):
            with self.assertRaises(ValueError):
                t.gaze_angles(pose)

    def test_osc_wire(self):
        self.assertEqual(t.osc_message('/x', [72]), b'/x\0\0,i\0\0' + struct.pack('>i', 72))
        self.assertEqual(t.osc_message('/x', [1.0, -2.0]), b'/x\0\0,ff\0' + struct.pack('>ff', 1, -2))
        for address in ('x', '/x\0y', '/x y', '/x*'):
            with self.assertRaises(ValueError):
                t.osc_message(address, [1])

    def test_default_never_opens_socket(self):
        with patch.object(t.socket, 'socket') as create:
            osc = t.Osc()
            osc.send('/x', [72])
            osc.close()
            create.assert_not_called()

    def test_only_configured_endpoint(self):
        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as receiver:
            receiver.bind(('127.0.0.1', 0))
            receiver.settimeout(1)
            osc = t.Osc(receiver.getsockname())
            try:
                osc.send('/tracking/eye/CenterPitchYaw', [0.0, 30.0])
                self.assertEqual(receiver.recv(1024), t.osc_message('/tracking/eye/CenterPitchYaw', [0.0, 30.0]))
            finally:
                osc.close()

    def test_endpoint_validation(self):
        for endpoint in [('example.org', 9000), ('0.0.0.0', 9000), ('224.0.0.1', 9000), ('127.0.0.1', 0), ('::1', 65536)]:
            with self.assertRaises(ValueError):
                t.Osc(endpoint)

    def test_heart_staleness_contact_and_log(self):
        now = [0]
        osc = Mock()
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / 'session.csv'
            session = t.HeartSession(osc, '/hr', path, lambda: now[0])
            self.assertIsNone(session.current())
            session.notification(b'\x00\x48')
            self.assertEqual(session.current(), 72)
            osc.send.assert_called_once_with('/hr', [72])
            now[0] = 6
            self.assertIsNone(session.current())
            session.notification(b'\x04\x48')
            self.assertIsNone(session.current())
            self.assertEqual(osc.send.call_count, 1)
            session.close()
            self.assertEqual(path.read_text().splitlines()[0], 'unix_seconds,bpm')
            self.assertEqual(len(path.read_text().splitlines()), 2)
            if os.name != 'nt':
                self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o600)
            with self.assertRaises(FileExistsError):
                t.HeartSession(osc, '/hr', path)

    def test_no_log_by_default(self):
        with patch.object(t.os, 'open') as create:
            session = t.HeartSession(Mock(), '/hr')
            session.notification(b'\x00\x48')
            session.close()
            create.assert_not_called()


class FakeBluez(unittest.TestCase):
    def setUp(self):
        self.device = '/org/bluez/hci0/dev_TEST'
        self.service = self.device + '/service1'
        self.char = self.service + '/char1'
        self.objects = {
            self.device: {t.DEVICE: {'Address': 'AA:BB:CC:DD:EE:FF', 'Connected': False, 'ServicesResolved': True}},
            self.service: {t.SERVICE: {'UUID': t.HRS, 'Device': self.device}},
            self.char: {t.CHARACTERISTIC: {'UUID': t.MEASUREMENT, 'Service': self.service, 'Flags': ['notify']}},
        }
        self.api = Mock()
        self.api.GetManagedObjects.side_effect = lambda: self.objects
        self.bus = Mock()
        self.interface = Mock(return_value=self.api)
        self.values = Mock()

    def reader(self):
        return t.BluezHeart(self.bus, self.interface, 'AA:BB:CC:DD:EE:FF', self.values)

    def test_subscribe_receive_and_cleanup(self):
        reader = self.reader()
        self.api.Connect.assert_called_once()
        self.assertTrue(reader.subscribe())
        self.api.StartNotify.assert_called_once()
        reader.changed(t.CHARACTERISTIC, {'Value': [0, 72]}, [], self.char)
        self.values.assert_called_once_with([0, 72])
        reader.changed(t.CHARACTERISTIC, {'Value': [0, 73]}, [], '/other/strap')
        self.assertEqual(self.values.call_count, 1)
        reader.close()
        self.api.StopNotify.assert_called_once()
        self.api.Disconnect.assert_called_once()
        self.bus.add_signal_receiver.return_value.remove.assert_called_once()

    def test_preserve_existing_connection(self):
        self.objects[self.device][t.DEVICE]['Connected'] = True
        reader = self.reader()
        reader.subscribe()
        reader.close()
        self.api.Connect.assert_not_called()
        self.api.Disconnect.assert_not_called()

    def test_only_selected_device_service(self):
        self.objects[self.service][t.SERVICE]['Device'] = '/other/device'
        reader = self.reader()
        try:
            with self.assertRaises(RuntimeError):
                reader.subscribe()
        finally:
            reader.close()
        self.api.StartNotify.assert_not_called()

    def test_wait_for_services(self):
        self.objects[self.device][t.DEVICE]['ServicesResolved'] = False
        reader = self.reader()
        self.assertFalse(reader.subscribe())
        reader.close()
        self.api.StartNotify.assert_not_called()
        self.api.StopNotify.assert_not_called()

    def test_disconnect_notification(self):
        reader = self.reader()
        reader.changed(t.DEVICE, {'Connected': 0}, [], self.device)  # dbus.Boolean behaves as int
        self.values.assert_called_once_with(None)
        reader.close()

    def test_failed_notify_cleans_connection(self):
        reader = self.reader()
        self.api.StartNotify.side_effect = RuntimeError('failure')
        with self.assertRaises(RuntimeError):
            reader.subscribe()
        reader.close()
        self.api.StopNotify.assert_not_called()
        self.api.Disconnect.assert_called_once()

    def test_unknown_device_does_not_connect_or_scan(self):
        self.objects.clear()
        with self.assertRaises(RuntimeError):
            self.reader()
        self.api.Connect.assert_not_called()
        self.api.StartDiscovery.assert_not_called()