first commit

linux/tests/test_p2p.py

@@ -0,0 +1,120 @@
+"""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()