diff --git a/tests/test_az177_video_single_write.py b/tests/test_az177_video_single_write.py index ff2bfd8..559e420 100644 --- a/tests/test_az177_video_single_write.py +++ b/tests/test_az177_video_single_write.py @@ -14,19 +14,15 @@ pytest.importorskip("inference") import inference as inference_mod +_TEST_JWT_SECRET = "az-test-secret-for-unit-tests-only-32b" + def _access_jwt(sub: str = "u1") -> str: - secret = os.environ.get("JWT_SECRET", "") - if secret: - return pyjwt.encode( - {"exp": int(time.time()) + 3600, "sub": sub}, - secret, - algorithm="HS256", - ) - import base64, json - raw = json.dumps({"exp": int(time.time()) + 3600, "sub": sub}, separators=(",", ":")).encode() - payload = base64.urlsafe_b64encode(raw).decode().rstrip("=") - return f"h.{payload}.s" + return pyjwt.encode( + {"exp": int(time.time()) + 3600, "sub": sub}, + _TEST_JWT_SECRET, + algorithm="HS256", + ) def _jpeg_bytes() -> bytes: @@ -73,6 +69,7 @@ def test_auth_image_still_writes_once_before_detect(reset_main_inference): expected_path = os.path.abspath(os.path.join(idir, f"{content_hash}.jpg")) client = TestClient(main.app) with ( + patch.object(main, "JWT_SECRET", _TEST_JWT_SECRET), patch.object(builtins, "open", tracking_open), patch.object(main.http_requests, "post", mock_post), patch.object(main.http_requests, "put", mock_put), diff --git a/tests/test_az178_streaming_video.py b/tests/test_az178_streaming_video.py index b672bff..a526cae 100644 --- a/tests/test_az178_streaming_video.py +++ b/tests/test_az178_streaming_video.py @@ -1,6 +1,4 @@ import asyncio -import base64 -import json import os import tempfile import threading @@ -275,20 +273,16 @@ class TestMediaContentHashFromFile: os.unlink(path) +_TEST_JWT_SECRET = "az-test-secret-for-unit-tests-only-32b" + + def _access_jwt(sub: str = "u1") -> str: import jwt as pyjwt - secret = os.environ.get("JWT_SECRET", "") - if secret: - return pyjwt.encode( - {"exp": int(time.time()) + 3600, "sub": sub}, - secret, - algorithm="HS256", - ) - raw = json.dumps( - {"exp": int(time.time()) + 3600, "sub": sub}, separators=(",", ":") - ).encode() - payload = base64.urlsafe_b64encode(raw).decode().rstrip("=") - return f"h.{payload}.s" + return pyjwt.encode( + {"exp": int(time.time()) + 3600, "sub": sub}, + _TEST_JWT_SECRET, + algorithm="HS256", + ) class _FakeInfStream: @@ -337,6 +331,7 @@ class TestDetectVideoEndpoint: from fastapi.testclient import TestClient client = TestClient(main.app) with ( + patch.object(main, "JWT_SECRET", _TEST_JWT_SECRET), patch.object(main, "get_inference", return_value=_FakeInfStream()), patch.object(main.http_requests, "post", mock_post), patch.object(main.http_requests, "put", mock_put),