WebRTC Connection Negotiator

Build a connection negotiator that establishes a WebRTC connection between two peers through proper SDP offer/answer exchange. The negotiator should handle the complete signaling flow from offer creation through answer acceptance.

Requirements

Your implementation should:

1. Create two peer connections representing the offering peer and answering peer
2. Execute the complete SDP offer/answer negotiation sequence
3. Properly exchange SDP descriptions between both peers
4. Ensure both peers reach a stable connection state after negotiation
5. Provide a way to verify the final connection states
6. Clean up resources when done

Implementation

Create a Python module with an async function negotiate_connection() that:

• Takes no parameters
• Creates two peer connections internally
• Executes the full offer/answer exchange:
  1. The offering peer creates an SDP offer
  2. The offering peer sets the offer as its local description
  3. The answering peer receives and sets the offer as its remote description
  4. The answering peer creates an SDP answer
  5. The answering peer sets the answer as its local description
  6. The offering peer receives and sets the answer as its remote description
• Returns a tuple containing the signaling states of both peers
• Properly closes both connections before returning

Test Cases

Test Case 1: Successful Negotiation @test

Input: Execute the negotiation function

Expected Output:

• Both peers reach "stable" signaling state
• Function returns ("stable", "stable")

Test File: test_negotiation.py

import pytest
from negotiation import negotiate_connection

@pytest.mark.asyncio
async def test_successful_negotiation():
    offering_state, answering_state = await negotiate_connection()
    assert offering_state == "stable"
    assert answering_state == "stable"

Test Case 2: No Errors During Exchange @test

Input: Execute the negotiation function

Expected Output:

• No exceptions are raised during the entire negotiation process

Test File: test_negotiation.py

import pytest
from negotiation import negotiate_connection

@pytest.mark.asyncio
async def test_no_errors():
    try:
        await negotiate_connection()
        success = True
    except Exception:
        success = False
    assert success is True

Dependencies { .dependencies }

aiortc { .dependency }

Provides WebRTC implementation for Python including peer connection management and SDP negotiation.