[PATCH BlueZ v5 3/3] test-functional: Cover BAP metadata reconfiguration

@@ -67,17 +67,14 @@ test_bap_unicast_transport_created
 	1. Start `bluetoothctl` with the scripts on both hosts.
 	2. Pair over LE: ``scan on`` on the initiator, ``advertise on`` on
 	   the acceptor, then ``pair``.
-	3. Initiator: ``endpoint.config <remote endpoint>
-	   /local/endpoint/ep0 16_2_1``, using the remote PAC Sink endpoint
-	   exposed once the services are resolved.
+	3. Wait for automatic stream configuration through the registered
+	   endpoint's ``SelectProperties`` method.
 	4. Initiator: ``transport.show <transport>`` for each transport.
 
 :Expected:
 	1. ``Endpoint /local/endpoint/ep0 registered`` on both hosts.
 	2. ``Pairing successful``.
-	3. The remote endpoint appears as
-	   ``Endpoint /org/bluez/hci0/dev_XX/pac_sinkN``, and configuring it
-	   creates one transport per location on *both* hosts.
+	3. One transport per location is created on *both* hosts.
 	4. Each transport of the initiator reports the local PAC Source
 	   UUID (``00002bcb-...``), ``Codec: 0x06`` for LC3, ``Device:``
 	   pointing at the acceptor device object, and ``State: idle``.

@@ -108,6 +105,35 @@ test_bap_unicast_transport_acquire
 	be established, as `bluetoothd` sets up the ISO listener on its own
 	when the stream is enabled.
 
+test_bap_unicast_reconfigure_metadata[empty|media]
+------------------------------------------------
+
+:Setup: As above, with both transports already created and ATT MTU 64.
+
+:Steps:
+	1. Create two custom presets from the existing codec configurations
+	   and QoS, with empty or Media streaming-context metadata. Call
+	   ``MediaEndpoint1.ClearConfiguration`` on the remote endpoint.
+	2. Wait for completion and check that both old transports are gone.
+	3. Issue ``endpoint.config`` for both presets without waiting
+	   between requests.
+	4. Acquire both replacement transports.
+
+:Expected: Both configuration requests succeed. Two replacement transports
+	retain their codec configurations and the requested metadata. Both
+	acquisitions succeed and the transports become active.
+
+:Notes: MTU 64 meets BAP's minimum. It fits the two-ASE Codec Configuration
+	response and one Codec Configured notification together, but not the
+	second ASE notification. This exercises the interval in which the
+	initiator has received a successful response but is still waiting
+	for a fresh Configured state before starting QoS.
+
+	Release uses D-Bus directly: bluetoothctl has no command for
+	``ClearConfiguration``. This test covers the initiator's playback
+	streams; it does not exercise microphone streams or simultaneous
+	playback and capture.
+
 BROADCAST
 =========
 

@@ -4,13 +4,21 @@
 Tests for BAP (LE Audio) using bluetoothctl in VM instances
 """
 
-import re
+import threading
 import warnings
 
+import dbus
 import pytest
 
-from pytest_bluezenv import Bluetoothd, Pexpect, find_exe, host_config
-from pytest_bluezenv.utils import bluez_src_dir
+from pytest_bluezenv import (
+    Bluetoothd,
+    Pexpect,
+    find_exe,
+    get_dbus,
+    host_config,
+    mainloop_wrap,
+)
+from pytest_bluezenv.utils import DEFAULT_TIMEOUT, bluez_src_dir
 
 pytestmark = [pytest.mark.vm]
 

@@ -24,8 +32,6 @@ KernelExperimental = true
 ControllerMode = le
 """
 
-PRESET = "16_2_1"
-
 TRANSPORT_RE = r"Transport (/org/bluez/\S+/fd\d+)"
 
 

@@ -214,12 +220,7 @@ def unicast_hosts(hosts):
 
     pair_le(host0, initiator, host1, acceptor)
 
-    # Remote PAC Sink endpoint is exposed once services are resolved
-    _, m = initiator.expect(r"Endpoint (/org/bluez/\S+/pac_sink\d+)")
-    remote = m[0].decode("utf-8")
-
-    initiator.send(f"endpoint.config {remote} /local/endpoint/ep0 {PRESET}\n")
-
+    # SelectProperties configures the streams automatically after pairing.
     yield host0, host1, initiator, acceptor
 
 

@@ -274,6 +275,122 @@ def test_bap_unicast_transport_acquire(unicast_hosts):
     )
 
 
+def clear_remote_transports(remote):
+    """Release the peer ASEs while retaining the connected GATT session."""
+    done = threading.Event()
+    errors = []
+
+    def failed(error):
+        errors.append(error)
+        done.set()
+
+    @mainloop_wrap
+    def clear():
+        dbus.Interface(
+            get_dbus().get_object("org.bluez", remote), "org.bluez.MediaEndpoint1"
+        ).ClearConfiguration(
+            dbus.ObjectPath(remote), reply_handler=done.set, error_handler=failed
+        )
+
+    clear()
+    assert done.wait(DEFAULT_TIMEOUT), "release did not complete"
+    if errors:
+        raise errors[0]
+
+
+@mainloop_wrap
+def remote_transport_properties(remote):
+    objects = dbus.Interface(
+        get_dbus().get_object("org.bluez", "/"),
+        "org.freedesktop.DBus.ObjectManager",
+    ).GetManagedObjects()
+    return {
+        str(path): props["org.bluez.MediaTransport1"]
+        for path, props in objects.items()
+        if path.startswith(remote + "/") and "org.bluez.MediaTransport1" in props
+    }
+
+
+def add_preset(ctl, name, props, metadata):
+    """Make a custom preset from a transport's codec configuration and QoS."""
+
+    def expect_reply(pattern):
+        failures = FAILURES + [r"(Invalid [^\r\n]*)", r"(No preset found)"]
+        idx, matches = ctl.expect(failures + [pattern], timeout=REPLY_TIMEOUT)
+        if idx < len(failures):
+            raise AssertionError(matches[0].decode("utf-8") if matches else "failed")
+
+    caps = " ".join(f"0x{byte:02x}" for byte in props["Configuration"])
+    meta = " ".join(f"0x{byte:02x}" for byte in metadata) or "no"
+    # dbus.Byte formats as a character unless converted to an ordinary int.
+    qos = {str(key): int(value) for key, value in props["QoS"].items()}
+    ctl.send(f'endpoint.presets /local/endpoint/ep0 {name} "{caps}"\n')
+    for prompt, value in [
+        ("Enter Target Latency", "Balance"),
+        ("Enter SDU Interval", qos["Interval"]),
+        ("Enter Framing", qos["Framing"]),
+        ("Enter PHY", qos["PHY"]),
+        ("Enter Max SDU", qos["SDU"]),
+        ("Enter RTN", qos["Retransmissions"]),
+        ("Enter Max Transport Latency", qos["Latency"]),
+        ("Enter Presentation Delay", qos["PresentationDelay"]),
+        ("Enter Metadata", meta),
+    ]:
+        expect_reply(prompt)
+        ctl.send(f"{value}\n")
+    ctl.send("version\n")
+    expect_reply(r"Version \d")
+
+
+# MTU 64 meets BAP's minimum and fits a two-ASE Codec Configuration response
+# with one Codec Configured notification, but not both ASE notifications.
+@host_config(
+    [Bluetoothd(conf=BAP_CONF + "\n[GATT]\nExchangeMTU = 64\n"), Pexpect()],
+    [Bluetoothd(conf=BAP_CONF), Pexpect()],
+)
+@pytest.mark.parametrize("metadata", [b"", b"\x03\x02\x04\x00"], ids=["empty", "media"])
+def test_bap_unicast_reconfigure_metadata(unicast_hosts, metadata):
+    host0, host1, initiator, acceptor = unicast_hosts
+    transports = expect_transports(initiator)
+    expect_transports(acceptor)
+    remote = transports[0].rsplit("/", 1)[0]
+    original = host0.call(remote_transport_properties, remote)
+    assert set(original) == set(transports)
+    presets = [f"metadata-{i}" for i in range(len(transports))]
+    for name, transport in zip(presets, transports):
+        add_preset(initiator, name, original[transport], metadata)
+
+    host0.call(clear_remote_transports, remote)
+    # Transport paths can be reused: check removal before recreating them.
+    assert not host0.call(remote_transport_properties, remote)
+
+    # PTY input permits both asynchronous requests without waiting for replies.
+    initiator.send(
+        "".join(
+            f"endpoint.config {remote} /local/endpoint/ep0 {name}\n" for name in presets
+        )
+    )
+    expect_all(initiator, [r"Endpoint /local/endpoint/ep0 configured"] * 2)
+    current = host0.call(remote_transport_properties, remote)
+    assert len(current) == len(original)
+    assert sorted(bytes(p["Configuration"]) for p in current.values()) == sorted(
+        bytes(p["Configuration"]) for p in original.values()
+    )
+    assert all(bytes(p["Metadata"]) == metadata for p in current.values())
+    left, right = sorted(current)
+    initiator.send(f"transport.acquire {left} {right}\n")
+    acquired = r"Acquire successful: fd \d+ MTU \d+:\d+"
+    expect_all(
+        initiator,
+        [
+            acquired,
+            acquired,
+            f"Transport {left} State: active",
+            f"Transport {right} State: active",
+        ],
+    )
+
+
 # Broadcast code used by the broadcast scripts, see BCAST_CODE in
 # client/player.c
 BCAST_CODE = (