From de44793481c70c356558fdc287dc071cb8af99e7 Mon Sep 17 00:00:00 2001 From: Nathan N <87051052+nat3058@users.noreply.github.com> Date: Fri, 27 Jun 2025 22:44:36 +0000 Subject: [PATCH 01/10] init testing --- dummy.txt | 0 1 file changed, 0 insertions(+), 0 deletions(-) create mode 100644 dummy.txt diff --git a/dummy.txt b/dummy.txt new file mode 100644 index 00000000..e69de29b From f91c42232e78786685a831699862e796eba97f4b Mon Sep 17 00:00:00 2001 From: Nathan N <87051052+nat3058@users.noreply.github.com> Date: Sat, 28 Jun 2025 16:39:35 +0000 Subject: [PATCH 02/10] More tests for certain edge cases --- livekit-api/tests/test_webhook.py | 164 ++++++++++++++++++++++++++++++ 1 file changed, 164 insertions(+) diff --git a/livekit-api/tests/test_webhook.py b/livekit-api/tests/test_webhook.py index fc8828e5..fcd64f58 100644 --- a/livekit-api/tests/test_webhook.py +++ b/livekit-api/tests/test_webhook.py @@ -3,6 +3,14 @@ import pytest # type: ignore from livekit.api import AccessToken, TokenVerifier, WebhookReceiver +from livekit.protocol.webhook import WebhookEvent # Keep this line +from livekit.protocol.models import ( # Added this import + Room, + ParticipantInfo, + TrackInfo, + TrackKind, + TrackSource, +) TEST_API_KEY = "myapikey" TEST_API_SECRET = "thiskeyistotallyunsafe" @@ -79,3 +87,159 @@ def test_invalid_body(): jwt = token.to_jwt() with pytest.raises(Exception): receiver.receive(body, jwt) + + +# --- ADDITIONAL TESTS START HERE --- + +# New test event: participant_connected +TEST_EVENT_PARTICIPANT_CONNECTED = """ +{ + "event": "participant_connected", + "room": { + "sid": "RM_hycBMAjmt6Ub", + "name": "Demo Room", + "emptyTimeout": 300, + "creationTime": "1692627281", + "numParticipants": 2 + }, + "participant": { + "sid": "PA_abcdefg", + "identity": "user123", + "state": 1, + "joinedAt": "1692985600", + "name": "User 1" + }, + "id": "EV_participant_connected_test", + "createdAt": "1692985600" +} +""" + +# New test event: track_published +TEST_EVENT_TRACK_PUBLISHED = """ +{ + "event": "track_published", + "room": { + "sid": "RM_hycBMAjmt6Ub", + "name": "Demo Room" + }, + "participant": { + "sid": "PA_abcdefg", + "identity": "user123", + "state": 2 + }, + "track": { + "sid": "TR_hijklm", + "name": "camera", + "kind": "VIDEO", + "source": "CAMERA", + "width": 640, + "height": 480, + "muted": false + }, + "id": "EV_track_published_test", + "createdAt": "1692985700" +} +""" + +# New test event: room_ended +TEST_EVENT_ROOM_ENDED = """ +{ + "event": "room_ended", + "room": { + "sid": "RM_hycBMAjmt6Ub", + "name": "Demo Room", + "emptyTimeout": 300, + "creationTime": "1692627281", + "numParticipants": 0 + }, + "id": "EV_room_ended_test", + "createdAt": "1692986000" +} +""" + + +def generate_webhook_token(event_body: str) -> str: + """Helper to generate a valid webhook token for a given event body.""" + hash64 = base64.b64encode(hashlib.sha256(event_body.encode()).digest()).decode() + token = AccessToken(TEST_API_KEY, TEST_API_SECRET) + token.claims.sha256 = hash64 + return token.to_jwt() + + +def test_webhook_receiver_room_started_details(): + """Test successful reception of a room_started event with content verification.""" + token_verifier = TokenVerifier(TEST_API_KEY, TEST_API_SECRET) + receiver = WebhookReceiver(token_verifier) + + jwt = generate_webhook_token(TEST_EVENT) # Using original TEST_EVENT here + event = receiver.receive(TEST_EVENT, jwt) + + assert event.event == "room_started" + assert event.room.sid == "RM_hycBMAjmt6Ub" + assert event.room.name == "Demo Room" + assert event.room.empty_timeout == 300 + assert event.room.creation_time == 1692627281 # Proto message parses as int + assert len(event.room.enabled_codecs) > 0 + + +def test_webhook_receiver_participant_connected_details(): + """Test successful reception of a participant_connected event with content verification.""" + token_verifier = TokenVerifier(TEST_API_KEY, TEST_API_SECRET) + receiver = WebhookReceiver(token_verifier) + + jwt = generate_webhook_token(TEST_EVENT_PARTICIPANT_CONNECTED) + event = receiver.receive(TEST_EVENT_PARTICIPANT_CONNECTED, jwt) + + assert event.event == "participant_connected" + assert isinstance(event.participant, ParticipantInfo) + assert event.participant.identity == "user123" + assert event.participant.sid == "PA_abcdefg" + assert event.participant.name == "User 1" + assert event.room.sid == "RM_hycBMAjmt6Ub" + assert event.room.num_participants == 2 + + +def test_webhook_receiver_track_published_details(): + """Test successful reception of a track_published event with content verification.""" + token_verifier = TokenVerifier(TEST_API_KEY, TEST_API_SECRET) + receiver = WebhookReceiver(token_verifier) + + jwt = generate_webhook_token(TEST_EVENT_TRACK_PUBLISHED) + event = receiver.receive(TEST_EVENT_TRACK_PUBLISHED, jwt) + + assert event.event == "track_published" + assert isinstance(event.track, TrackInfo) + assert event.track.sid == "TR_hijklm" + assert event.track.name == "camera" + assert event.track.kind == TrackKind.KIND_VIDEO + assert event.track.source == TrackSource.CAMERA + assert event.track.width == 640 + assert event.track.height == 480 + assert not event.track.muted + assert event.participant.identity == "user123" + + +def test_webhook_receiver_room_ended_details(): + """Test successful reception of a room_ended event with content verification.""" + token_verifier = TokenVerifier(TEST_API_KEY, TEST_API_SECRET) + receiver = WebhookReceiver(token_verifier) + + jwt = generate_webhook_token(TEST_EVENT_ROOM_ENDED) + event = receiver.receive(TEST_EVENT_ROOM_ENDED, jwt) + + assert event.event == "room_ended" + assert event.room.sid == "RM_hycBMAjmt6Ub" + assert event.room.name == "Demo Room" + assert event.room.num_participants == 0 + + +def test_missing_sha256_claim_raises_error(): + """Test that missing SHA256 in the token claims raises an exception.""" + token_verifier = TokenVerifier(TEST_API_KEY, TEST_API_SECRET) + receiver = WebhookReceiver(token_verifier) + + # Create a token without explicitly setting claims.sha256 + token_without_sha256 = AccessToken(TEST_API_KEY, TEST_API_SECRET).to_jwt() + + with pytest.raises(Exception, match="sha256 was not found in the token"): + receiver.receive(TEST_EVENT, token_without_sha256) \ No newline at end of file From b190d15e5d0d633f9e0b11791f80774f29ea5802 Mon Sep 17 00:00:00 2001 From: Nathan N <87051052+nat3058@users.noreply.github.com> Date: Sat, 28 Jun 2025 16:57:43 +0000 Subject: [PATCH 03/10] more tests --- livekit-api/tests/test_webhook.py | 182 +++++------------------------- 1 file changed, 31 insertions(+), 151 deletions(-) diff --git a/livekit-api/tests/test_webhook.py b/livekit-api/tests/test_webhook.py index fcd64f58..6353ae3e 100644 --- a/livekit-api/tests/test_webhook.py +++ b/livekit-api/tests/test_webhook.py @@ -1,16 +1,11 @@ import base64 import hashlib +from datetime import datetime, timedelta import pytest # type: ignore +import jwt.exceptions # Import for JWT specific exceptions from livekit.api import AccessToken, TokenVerifier, WebhookReceiver -from livekit.protocol.webhook import WebhookEvent # Keep this line -from livekit.protocol.models import ( # Added this import - Room, - ParticipantInfo, - TrackInfo, - TrackKind, - TrackSource, -) +from livekit.api.errors import LiveKitError # Import for LiveKit API specific errors TEST_API_KEY = "myapikey" TEST_API_SECRET = "thiskeyistotallyunsafe" @@ -72,7 +67,7 @@ def test_bad_hash(): hash64 = base64.b64encode(hashlib.sha256("wrong_hash".encode()).digest()).decode() token.claims.sha256 = hash64 jwt = token.to_jwt() - with pytest.raises(Exception): + with pytest.raises(Exception): # Using a broad Exception for existing test receiver.receive(TEST_EVENT, jwt) @@ -85,161 +80,46 @@ def test_invalid_body(): hash64 = base64.b64encode(hashlib.sha256(body.encode()).digest()).decode() token.claims.sha256 = hash64 jwt = token.to_jwt() - with pytest.raises(Exception): + with pytest.raises(Exception): # Using a broad Exception for existing test receiver.receive(body, jwt) -# --- ADDITIONAL TESTS START HERE --- +def test_mismatched_api_key_secret(): + """ + Test that receiving a webhook with a token signed by a different API key/secret + raises an error. + """ + TEST_API_KEY_BAD = "badkey" + TEST_API_SECRET_BAD = "badsecret" -# New test event: participant_connected -TEST_EVENT_PARTICIPANT_CONNECTED = """ -{ - "event": "participant_connected", - "room": { - "sid": "RM_hycBMAjmt6Ub", - "name": "Demo Room", - "emptyTimeout": 300, - "creationTime": "1692627281", - "numParticipants": 2 - }, - "participant": { - "sid": "PA_abcdefg", - "identity": "user123", - "state": 1, - "joinedAt": "1692985600", - "name": "User 1" - }, - "id": "EV_participant_connected_test", - "createdAt": "1692985600" -} -""" - -# New test event: track_published -TEST_EVENT_TRACK_PUBLISHED = """ -{ - "event": "track_published", - "room": { - "sid": "RM_hycBMAjmt6Ub", - "name": "Demo Room" - }, - "participant": { - "sid": "PA_abcdefg", - "identity": "user123", - "state": 2 - }, - "track": { - "sid": "TR_hijklm", - "name": "camera", - "kind": "VIDEO", - "source": "CAMERA", - "width": 640, - "height": 480, - "muted": false - }, - "id": "EV_track_published_test", - "createdAt": "1692985700" -} -""" - -# New test event: room_ended -TEST_EVENT_ROOM_ENDED = """ -{ - "event": "room_ended", - "room": { - "sid": "RM_hycBMAjmt6Ub", - "name": "Demo Room", - "emptyTimeout": 300, - "creationTime": "1692627281", - "numParticipants": 0 - }, - "id": "EV_room_ended_test", - "createdAt": "1692986000" -} -""" - - -def generate_webhook_token(event_body: str) -> str: - """Helper to generate a valid webhook token for a given event body.""" - hash64 = base64.b64encode(hashlib.sha256(event_body.encode()).digest()).decode() - token = AccessToken(TEST_API_KEY, TEST_API_SECRET) - token.claims.sha256 = hash64 - return token.to_jwt() - - -def test_webhook_receiver_room_started_details(): - """Test successful reception of a room_started event with content verification.""" token_verifier = TokenVerifier(TEST_API_KEY, TEST_API_SECRET) receiver = WebhookReceiver(token_verifier) - jwt = generate_webhook_token(TEST_EVENT) # Using original TEST_EVENT here - event = receiver.receive(TEST_EVENT, jwt) - - assert event.event == "room_started" - assert event.room.sid == "RM_hycBMAjmt6Ub" - assert event.room.name == "Demo Room" - assert event.room.empty_timeout == 300 - assert event.room.creation_time == 1692627281 # Proto message parses as int - assert len(event.room.enabled_codecs) > 0 - - -def test_webhook_receiver_participant_connected_details(): - """Test successful reception of a participant_connected event with content verification.""" - token_verifier = TokenVerifier(TEST_API_KEY, TEST_API_SECRET) - receiver = WebhookReceiver(token_verifier) - - jwt = generate_webhook_token(TEST_EVENT_PARTICIPANT_CONNECTED) - event = receiver.receive(TEST_EVENT_PARTICIPANT_CONNECTED, jwt) - - assert event.event == "participant_connected" - assert isinstance(event.participant, ParticipantInfo) - assert event.participant.identity == "user123" - assert event.participant.sid == "PA_abcdefg" - assert event.participant.name == "User 1" - assert event.room.sid == "RM_hycBMAjmt6Ub" - assert event.room.num_participants == 2 - - -def test_webhook_receiver_track_published_details(): - """Test successful reception of a track_published event with content verification.""" - token_verifier = TokenVerifier(TEST_API_KEY, TEST_API_SECRET) - receiver = WebhookReceiver(token_verifier) - - jwt = generate_webhook_token(TEST_EVENT_TRACK_PUBLISHED) - event = receiver.receive(TEST_EVENT_TRACK_PUBLISHED, jwt) + # Token signed with incorrect credentials + token = AccessToken(TEST_API_KEY_BAD, TEST_API_SECRET_BAD) + hash64 = base64.b64encode(hashlib.sha256(TEST_EVENT.encode()).digest()).decode() + token.claims.sha256 = hash64 + jwt = token.to_jwt() - assert event.event == "track_published" - assert isinstance(event.track, TrackInfo) - assert event.track.sid == "TR_hijklm" - assert event.track.name == "camera" - assert event.track.kind == TrackKind.KIND_VIDEO - assert event.track.source == TrackSource.CAMERA - assert event.track.width == 640 - assert event.track.height == 480 - assert not event.track.muted - assert event.participant.identity == "user123" + with pytest.raises(LiveKitError, match="could not verify token signature"): + receiver.receive(TEST_EVENT, jwt) -def test_webhook_receiver_room_ended_details(): - """Test successful reception of a room_ended event with content verification.""" +def test_expired_token(): + """ + Test that receiving a webhook with an expired token raises an ExpiredSignatureError. + """ token_verifier = TokenVerifier(TEST_API_KEY, TEST_API_SECRET) receiver = WebhookReceiver(token_verifier) - jwt = generate_webhook_token(TEST_EVENT_ROOM_ENDED) - event = receiver.receive(TEST_EVENT_ROOM_ENDED, jwt) - - assert event.event == "room_ended" - assert event.room.sid == "RM_hycBMAjmt6Ub" - assert event.room.name == "Demo Room" - assert event.room.num_participants == 0 + token = AccessToken(TEST_API_KEY, TEST_API_SECRET) + hash64 = base64.b64encode(hashlib.sha256(TEST_EVENT.encode()).digest()).decode() + token.claims.sha256 = hash64 + # Set the token's expiration to a time in the past + token.claims.exp = datetime.utcnow() - timedelta(seconds=60) # 1 minute ago -def test_missing_sha256_claim_raises_error(): - """Test that missing SHA256 in the token claims raises an exception.""" - token_verifier = TokenVerifier(TEST_API_KEY, TEST_API_SECRET) - receiver = WebhookReceiver(token_verifier) - - # Create a token without explicitly setting claims.sha256 - token_without_sha256 = AccessToken(TEST_API_KEY, TEST_API_SECRET).to_jwt() + jwt = token.to_jwt() - with pytest.raises(Exception, match="sha256 was not found in the token"): - receiver.receive(TEST_EVENT, token_without_sha256) \ No newline at end of file + with pytest.raises(jwt.exceptions.ExpiredSignatureError): + receiver.receive(TEST_EVENT, jwt) \ No newline at end of file From 1bf2d32d7b49961ea7c42b4b8e1dd187aaffed48 Mon Sep 17 00:00:00 2001 From: Nathan N <87051052+nat3058@users.noreply.github.com> Date: Sat, 28 Jun 2025 16:59:38 +0000 Subject: [PATCH 04/10] fixed ruff --- livekit-api/tests/test_webhook.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/livekit-api/tests/test_webhook.py b/livekit-api/tests/test_webhook.py index 6353ae3e..8d625ad9 100644 --- a/livekit-api/tests/test_webhook.py +++ b/livekit-api/tests/test_webhook.py @@ -3,9 +3,7 @@ from datetime import datetime, timedelta import pytest # type: ignore -import jwt.exceptions # Import for JWT specific exceptions from livekit.api import AccessToken, TokenVerifier, WebhookReceiver -from livekit.api.errors import LiveKitError # Import for LiveKit API specific errors TEST_API_KEY = "myapikey" TEST_API_SECRET = "thiskeyistotallyunsafe" From ad6c075be9713402b60ae10e94ba6c213132b543 Mon Sep 17 00:00:00 2001 From: Nathan N <87051052+nat3058@users.noreply.github.com> Date: Sat, 28 Jun 2025 17:06:18 +0000 Subject: [PATCH 05/10] cleaned up --- livekit-api/tests/test_webhook.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/livekit-api/tests/test_webhook.py b/livekit-api/tests/test_webhook.py index 8d625ad9..36d71953 100644 --- a/livekit-api/tests/test_webhook.py +++ b/livekit-api/tests/test_webhook.py @@ -99,7 +99,8 @@ def test_mismatched_api_key_secret(): token.claims.sha256 = hash64 jwt = token.to_jwt() - with pytest.raises(LiveKitError, match="could not verify token signature"): + # Now using broad Exception, as requested + with pytest.raises(Exception, match="could not verify token signature"): receiver.receive(TEST_EVENT, jwt) @@ -119,5 +120,6 @@ def test_expired_token(): jwt = token.to_jwt() - with pytest.raises(jwt.exceptions.ExpiredSignatureError): + # Now using broad Exception, as requested + with pytest.raises(Exception): receiver.receive(TEST_EVENT, jwt) \ No newline at end of file From 08f7e84fe5a086f3be85282f0ad49ed31d80beaa Mon Sep 17 00:00:00 2001 From: Nathan N <87051052+nat3058@users.noreply.github.com> Date: Sat, 28 Jun 2025 17:12:09 +0000 Subject: [PATCH 06/10] better tests --- livekit-api/tests/test_webhook.py | 18 ++++++++++++------ 1 file changed, 12 insertions(+), 6 deletions(-) diff --git a/livekit-api/tests/test_webhook.py b/livekit-api/tests/test_webhook.py index 36d71953..379316f0 100644 --- a/livekit-api/tests/test_webhook.py +++ b/livekit-api/tests/test_webhook.py @@ -1,6 +1,6 @@ import base64 import hashlib -from datetime import datetime, timedelta +from datetime import datetime, timedelta, UTC # Import UTC for timezone-aware datetime import pytest # type: ignore from livekit.api import AccessToken, TokenVerifier, WebhookReceiver @@ -99,8 +99,11 @@ def test_mismatched_api_key_secret(): token.claims.sha256 = hash64 jwt = token.to_jwt() - # Now using broad Exception, as requested - with pytest.raises(Exception, match="could not verify token signature"): + # The LiveKit API internally catches jwt.exceptions.InvalidSignatureError + # and re-raises it as a LiveKitError with a message starting "could not verify token: " + # followed by the original JWT error message. + expected_match = r"could not verify token: Signature verification failed" + with pytest.raises(Exception, match=expected_match): receiver.receive(TEST_EVENT, jwt) @@ -116,10 +119,13 @@ def test_expired_token(): token.claims.sha256 = hash64 # Set the token's expiration to a time in the past - token.claims.exp = datetime.utcnow() - timedelta(seconds=60) # 1 minute ago + # Using datetime.now(UTC) to address the DeprecationWarning + token.claims.exp = datetime.now(UTC) - timedelta(seconds=60) # 1 minute ago jwt = token.to_jwt() - # Now using broad Exception, as requested - with pytest.raises(Exception): + # Similar to mismatched key, LiveKit API wraps the ExpiredSignatureError + # The message will start with "could not verify token: " + expected_match = r"could not verify token: Signature has expired" + with pytest.raises(Exception, match=expected_match): receiver.receive(TEST_EVENT, jwt) \ No newline at end of file From e0967836767164955f60f500087e8424e8de9b1f Mon Sep 17 00:00:00 2001 From: Nathan N <87051052+nat3058@users.noreply.github.com> Date: Sat, 28 Jun 2025 17:20:07 +0000 Subject: [PATCH 07/10] simple test --- livekit-api/tests/test_webhook.py | 47 ++++--------------------------- 1 file changed, 5 insertions(+), 42 deletions(-) diff --git a/livekit-api/tests/test_webhook.py b/livekit-api/tests/test_webhook.py index 379316f0..62a630fa 100644 --- a/livekit-api/tests/test_webhook.py +++ b/livekit-api/tests/test_webhook.py @@ -81,51 +81,14 @@ def test_invalid_body(): with pytest.raises(Exception): # Using a broad Exception for existing test receiver.receive(body, jwt) - -def test_mismatched_api_key_secret(): - """ - Test that receiving a webhook with a token signed by a different API key/secret - raises an error. - """ - TEST_API_KEY_BAD = "badkey" - TEST_API_SECRET_BAD = "badsecret" - - token_verifier = TokenVerifier(TEST_API_KEY, TEST_API_SECRET) - receiver = WebhookReceiver(token_verifier) - - # Token signed with incorrect credentials - token = AccessToken(TEST_API_KEY_BAD, TEST_API_SECRET_BAD) - hash64 = base64.b64encode(hashlib.sha256(TEST_EVENT.encode()).digest()).decode() - token.claims.sha256 = hash64 - jwt = token.to_jwt() - - # The LiveKit API internally catches jwt.exceptions.InvalidSignatureError - # and re-raises it as a LiveKitError with a message starting "could not verify token: " - # followed by the original JWT error message. - expected_match = r"could not verify token: Signature verification failed" - with pytest.raises(Exception, match=expected_match): - receiver.receive(TEST_EVENT, jwt) - - -def test_expired_token(): +def test_malformed_jwt(): """ - Test that receiving a webhook with an expired token raises an ExpiredSignatureError. + Test that receiving a webhook with a malformed JWT string raises an error. """ token_verifier = TokenVerifier(TEST_API_KEY, TEST_API_SECRET) receiver = WebhookReceiver(token_verifier) - token = AccessToken(TEST_API_KEY, TEST_API_SECRET) - hash64 = base64.b64encode(hashlib.sha256(TEST_EVENT.encode()).digest()).decode() - token.claims.sha256 = hash64 - - # Set the token's expiration to a time in the past - # Using datetime.now(UTC) to address the DeprecationWarning - token.claims.exp = datetime.now(UTC) - timedelta(seconds=60) # 1 minute ago - - jwt = token.to_jwt() + malformed_jwt = "this.is.not.a.valid.jwt" # A clearly malformed string - # Similar to mismatched key, LiveKit API wraps the ExpiredSignatureError - # The message will start with "could not verify token: " - expected_match = r"could not verify token: Signature has expired" - with pytest.raises(Exception, match=expected_match): - receiver.receive(TEST_EVENT, jwt) \ No newline at end of file + with pytest.raises(Exception): + receiver.receive(TEST_EVENT, malformed_jwt) \ No newline at end of file From b1fbcdc831b9745c164ba6126f5d37892d5e2a82 Mon Sep 17 00:00:00 2001 From: Nathan N <87051052+nat3058@users.noreply.github.com> Date: Sat, 28 Jun 2025 17:48:36 +0000 Subject: [PATCH 08/10] simple test - cleaned --- livekit-api/tests/test_webhook.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/livekit-api/tests/test_webhook.py b/livekit-api/tests/test_webhook.py index 62a630fa..2c240a3b 100644 --- a/livekit-api/tests/test_webhook.py +++ b/livekit-api/tests/test_webhook.py @@ -1,7 +1,5 @@ import base64 import hashlib -from datetime import datetime, timedelta, UTC # Import UTC for timezone-aware datetime - import pytest # type: ignore from livekit.api import AccessToken, TokenVerifier, WebhookReceiver From 231bb46aef267409b71d4698faf42c6e402756dc Mon Sep 17 00:00:00 2001 From: Nathan N <87051052+nat3058@users.noreply.github.com> Date: Sat, 28 Jun 2025 17:52:55 +0000 Subject: [PATCH 09/10] ruff --- livekit-api/tests/test_webhook.py | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/livekit-api/tests/test_webhook.py b/livekit-api/tests/test_webhook.py index 2c240a3b..74af278b 100644 --- a/livekit-api/tests/test_webhook.py +++ b/livekit-api/tests/test_webhook.py @@ -1,10 +1,11 @@ import base64 import hashlib + import pytest # type: ignore from livekit.api import AccessToken, TokenVerifier, WebhookReceiver -TEST_API_KEY = "myapikey" -TEST_API_SECRET = "thiskeyistotallyunsafe" +TEST_API_KEY = 'myapikey' +TEST_API_SECRET = 'thiskeyistotallyunsafe' TEST_EVENT = """ { "event": "room_started", @@ -60,10 +61,10 @@ def test_bad_hash(): receiver = WebhookReceiver(token_verifier) token = AccessToken(TEST_API_KEY, TEST_API_SECRET) - hash64 = base64.b64encode(hashlib.sha256("wrong_hash".encode()).digest()).decode() + hash64 = base64.b64encode(hashlib.sha256('wrong_hash'.encode()).digest()).decode() token.claims.sha256 = hash64 jwt = token.to_jwt() - with pytest.raises(Exception): # Using a broad Exception for existing test + with pytest.raises(Exception): # Using a broad Exception for existing test receiver.receive(TEST_EVENT, jwt) @@ -72,13 +73,14 @@ def test_invalid_body(): receiver = WebhookReceiver(token_verifier) token = AccessToken(TEST_API_KEY, TEST_API_SECRET) - body = "invalid body" + body = 'invalid body' hash64 = base64.b64encode(hashlib.sha256(body.encode()).digest()).decode() token.claims.sha256 = hash64 jwt = token.to_jwt() - with pytest.raises(Exception): # Using a broad Exception for existing test + with pytest.raises(Exception): # Using a broad Exception for existing test receiver.receive(body, jwt) + def test_malformed_jwt(): """ Test that receiving a webhook with a malformed JWT string raises an error. @@ -86,7 +88,7 @@ def test_malformed_jwt(): token_verifier = TokenVerifier(TEST_API_KEY, TEST_API_SECRET) receiver = WebhookReceiver(token_verifier) - malformed_jwt = "this.is.not.a.valid.jwt" # A clearly malformed string + malformed_jwt = 'this.is.not.a.valid.jwt' # A clearly malformed string with pytest.raises(Exception): receiver.receive(TEST_EVENT, malformed_jwt) \ No newline at end of file From 4e87b75ae8992b6d29fc8a9e11a56feea9cdcfe6 Mon Sep 17 00:00:00 2001 From: Nathan N <87051052+nat3058@users.noreply.github.com> Date: Sat, 28 Jun 2025 17:54:15 +0000 Subject: [PATCH 10/10] ruff --- livekit-api/tests/test_webhook.py | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/livekit-api/tests/test_webhook.py b/livekit-api/tests/test_webhook.py index 74af278b..98f424cb 100644 --- a/livekit-api/tests/test_webhook.py +++ b/livekit-api/tests/test_webhook.py @@ -1,11 +1,10 @@ import base64 import hashlib - import pytest # type: ignore from livekit.api import AccessToken, TokenVerifier, WebhookReceiver -TEST_API_KEY = 'myapikey' -TEST_API_SECRET = 'thiskeyistotallyunsafe' +TEST_API_KEY = "myapikey" +TEST_API_SECRET = "thiskeyistotallyunsafe" TEST_EVENT = """ { "event": "room_started", @@ -61,7 +60,7 @@ def test_bad_hash(): receiver = WebhookReceiver(token_verifier) token = AccessToken(TEST_API_KEY, TEST_API_SECRET) - hash64 = base64.b64encode(hashlib.sha256('wrong_hash'.encode()).digest()).decode() + hash64 = base64.b64encode(hashlib.sha256("wrong_hash".encode()).digest()).decode() token.claims.sha256 = hash64 jwt = token.to_jwt() with pytest.raises(Exception): # Using a broad Exception for existing test @@ -73,7 +72,7 @@ def test_invalid_body(): receiver = WebhookReceiver(token_verifier) token = AccessToken(TEST_API_KEY, TEST_API_SECRET) - body = 'invalid body' + body = "invalid body" hash64 = base64.b64encode(hashlib.sha256(body.encode()).digest()).decode() token.claims.sha256 = hash64 jwt = token.to_jwt() @@ -88,7 +87,7 @@ def test_malformed_jwt(): token_verifier = TokenVerifier(TEST_API_KEY, TEST_API_SECRET) receiver = WebhookReceiver(token_verifier) - malformed_jwt = 'this.is.not.a.valid.jwt' # A clearly malformed string + malformed_jwt = "this.is.not.a.valid.jwt" # A clearly malformed string with pytest.raises(Exception): - receiver.receive(TEST_EVENT, malformed_jwt) \ No newline at end of file + receiver.receive(TEST_EVENT, malformed_jwt)