File: test_ortc.py

package info (click to toggle)
python-aiortc 1.11.0-1
  • links: PTS, VCS
  • area: main
  • in suites: sid, trixie
  • size: 3,052 kB
  • sloc: python: 20,294; javascript: 321; makefile: 21; sh: 1
file content (78 lines) | stat: -rw-r--r-- 2,201 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
import asyncio
import os
from unittest import TestCase, skipUnless

from aiortc import (
    RTCCertificate,
    RTCDtlsTransport,
    RTCIceGatherer,
    RTCIceTransport,
    RTCSctpTransport,
)

from .utils import asynctest


async def start_dtls_pair(ice_a, ice_b):
    dtls_a = RTCDtlsTransport(ice_a, [RTCCertificate.generateCertificate()])
    dtls_b = RTCDtlsTransport(ice_b, [RTCCertificate.generateCertificate()])

    await asyncio.gather(
        dtls_a.start(dtls_b.getLocalParameters()),
        dtls_b.start(dtls_a.getLocalParameters()),
    )

    return dtls_a, dtls_b


async def start_ice_pair():
    ice_a = RTCIceTransport(gatherer=RTCIceGatherer())
    ice_b = RTCIceTransport(gatherer=RTCIceGatherer())

    await asyncio.gather(ice_a.iceGatherer.gather(), ice_b.iceGatherer.gather())

    for candidate in ice_b.iceGatherer.getLocalCandidates():
        await ice_a.addRemoteCandidate(candidate)
    for candidate in ice_a.iceGatherer.getLocalCandidates():
        await ice_b.addRemoteCandidate(candidate)
    await asyncio.gather(
        ice_a.start(ice_b.iceGatherer.getLocalParameters()),
        ice_b.start(ice_a.iceGatherer.getLocalParameters()),
    )

    return ice_a, ice_b


async def start_sctp_pair(dtls_a, dtls_b):
    sctp_a = RTCSctpTransport(dtls_a)
    sctp_b = RTCSctpTransport(dtls_b)

    await asyncio.gather(
        sctp_a.start(sctp_b.getCapabilities(), sctp_b.port),
        sctp_b.start(sctp_a.getCapabilities(), sctp_a.port),
    )

    return sctp_a, sctp_b


@skipUnless('EXTENDED_TESTING' in os.environ, 'test requires network access')
class OrtcTest(TestCase):
    @asynctest
    async def test_sctp(self):
        # start ICE transports
        ice_a, ice_b = await start_ice_pair()

        # start DTLS transports
        dtls_a, dtls_b = await start_dtls_pair(ice_a, ice_b)

        # start SCTP transports
        sctp_a, sctp_b = await start_sctp_pair(dtls_a, dtls_b)

        # stop SCTP transports
        await asyncio.gather(sctp_a.stop(), sctp_b.stop())

        # stop DTLS transports
        await asyncio.gather(dtls_a.stop(), dtls_b.stop())

        # stop ICE transports
        await asyncio.gather(ice_a.stop(), ice_b.stop())