121 lines
4.2 KiB
Python
121 lines
4.2 KiB
Python
"""Prüft die Anmeldung als Wi-Fi-Display-Quelle gegen einen nachgebauten
|
|
wpa_supplicant. Genau dieser Schritt ist auf Android gesperrt."""
|
|
|
|
import os
|
|
import socket
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from wfd import p2p, wpa
|
|
|
|
|
|
class FakeSupplicant:
|
|
"""Nimmt Befehle auf einem UNIX-Socket entgegen, wie wpa_supplicant es tut."""
|
|
|
|
def __init__(self, path, answers=None):
|
|
self.path = path
|
|
self.answers = answers or {}
|
|
self.received = []
|
|
self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
|
|
self.sock.bind(path)
|
|
self.sock.settimeout(0.3)
|
|
self._stop = threading.Event()
|
|
self.thread = threading.Thread(target=self._serve, daemon=True)
|
|
self.thread.start()
|
|
|
|
def _serve(self):
|
|
while not self._stop.is_set():
|
|
try:
|
|
data, sender = self.sock.recvfrom(4096)
|
|
except socket.timeout:
|
|
continue
|
|
except OSError:
|
|
break
|
|
command = data.decode()
|
|
self.received.append(command)
|
|
reply = self.answers.get(command)
|
|
if reply is None:
|
|
for prefix, value in self.answers.items():
|
|
if command.startswith(prefix):
|
|
reply = value
|
|
break
|
|
self.sock.sendto((reply if reply is not None else "OK").encode(), sender)
|
|
|
|
def stop(self):
|
|
self._stop.set()
|
|
self.thread.join(1)
|
|
self.sock.close()
|
|
|
|
|
|
class WfdRegistrationTest(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.dir = tempfile.mkdtemp()
|
|
self.path = os.path.join(self.dir, "wlan0")
|
|
|
|
def test_announces_itself_as_source(self):
|
|
fake = FakeSupplicant(self.path)
|
|
try:
|
|
client = wpa.WpaClient(self.path)
|
|
manager = p2p.P2pManager(client)
|
|
self.assertTrue(manager.enable_wfd_source(control_port=7236))
|
|
client.close()
|
|
finally:
|
|
fake.stop()
|
|
|
|
self.assertIn("SET wifi_display 1", fake.received)
|
|
subelem = [c for c in fake.received if c.startswith("WFD_SUBELEM_SET")]
|
|
self.assertEqual(1, len(subelem), f"Erwartet ein WFD_SUBELEM_SET, bekam {fake.received}")
|
|
|
|
# Aufbau: "WFD_SUBELEM_SET 0 0006" + Info(4) + Port(4) + Durchsatz(4)
|
|
payload = subelem[0].split()[2]
|
|
self.assertTrue(payload.startswith("0006"), payload)
|
|
self.assertEqual(16, len(payload))
|
|
info = int(payload[4:8], 16)
|
|
port = int(payload[8:12], 16)
|
|
self.assertEqual(0b00, info & 0b11, "Gerätetyp muss 'Quelle' sein")
|
|
self.assertEqual(0b01, (info >> 4) & 0b11, "Sitzung muss als verfügbar gemeldet werden")
|
|
self.assertEqual(7236, port)
|
|
|
|
def test_reads_sink_announcement(self):
|
|
answers = {
|
|
"P2P_PEERS": "aa:bb:cc:dd:ee:ff",
|
|
"P2P_PEER": ("aa:bb:cc:dd:ee:ff\ndevice_name=[TV]Zimmer 2305\n"
|
|
"pri_dev_type=7-0050F204-1\nwfd_dev_info=0x00111c440032\n"),
|
|
}
|
|
fake = FakeSupplicant(self.path, answers)
|
|
try:
|
|
client = wpa.WpaClient(self.path)
|
|
peers = p2p.P2pManager(client).peers()
|
|
client.close()
|
|
finally:
|
|
fake.stop()
|
|
|
|
self.assertEqual(1, len(peers))
|
|
peer = peers[0]
|
|
self.assertEqual("[TV]Zimmer 2305", peer.name)
|
|
self.assertTrue(peer.is_sink, "Der Fernseher muss als Bildschirm erkannt werden")
|
|
self.assertTrue(peer.session_available, "Er wartet auf eine Verbindung")
|
|
self.assertEqual(7236, peer.control_port)
|
|
|
|
def test_connect_lets_tv_lead_the_group(self):
|
|
fake = FakeSupplicant(self.path)
|
|
try:
|
|
client = wpa.WpaClient(self.path)
|
|
peer = p2p.Peer("aa:bb:cc:dd:ee:ff", "TV", True, True, 7236)
|
|
p2p.P2pManager(client).connect(peer)
|
|
client.close()
|
|
finally:
|
|
fake.stop()
|
|
connect = [c for c in fake.received if c.startswith("P2P_CONNECT")][0]
|
|
self.assertIn("pbc", connect)
|
|
self.assertIn("go_intent=0", connect)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|