1
1
import importlib .metadata
2
+ import os
2
3
import time
3
4
import uuid
4
- from typing import Annotated
5
+ from pathlib import Path
6
+ from typing import Any
5
7
6
8
import jwt
7
9
import pytest
8
- from pydantic import Field , HttpUrl , SecretStr , computed_field , field_validator
9
- from pydantic_settings import BaseSettings , NoDecode , SettingsConfigDict
10
10
11
11
# sora_sdk から SoraLoggingSeverity をインポート
12
12
from sora_sdk import SoraLoggingSeverity
13
13
14
14
15
- class Settings (BaseSettings ):
16
- model_config = SettingsConfigDict (env_file = ".env" , env_prefix = ".env" , env_file_encoding = "utf-8" )
17
-
18
- # TODO: list[WebsocketUrl] 型にする
19
- signaling_urls : Annotated [list [str ], NoDecode ] = Field (default = [], alias = "TEST_SIGNALING_URLS" )
20
- channel_id_prefix : str = Field (default = "" , alias = "TEST_CHANNEL_ID_PREFIX" )
21
- secret : SecretStr | None = Field (default = None , alias = "TEST_SECRET_KEY" )
22
- api_url : HttpUrl | None = Field (default = None , alias = "TEST_API_URL" )
23
- # TODO: openh264_path は FilePath 型にする
24
- openh264_path : str | None = Field (default = None , alias = "OPENH264_PATH" )
25
- libwebrtc_log : SoraLoggingSeverity | None = Field (default = None , alias = "TEST_LIBWEBRTC_LOG" )
26
-
27
- channel_id_suffix : str = Field (default_factory = lambda : str (uuid .uuid4 ()))
28
-
29
- @field_validator ("signaling_urls" , mode = "before" )
30
- @classmethod
31
- def validate_signaling_urls (cls , v : str ) -> list [str ]:
32
- """
33
- TEST_SIGNALING_URLS が , で区切られている場合、それぞれの URL をリストに変換する
34
- """
35
- return [x .strip () for x in v .split ("," )]
36
-
37
- @field_validator ("libwebrtc_log" , mode = "before" )
38
- @classmethod
39
- def validate_libwebrtc_log (cls , v : str | None ) -> SoraLoggingSeverity | None :
40
- if v is None :
15
+ class Settings :
16
+ def __init__ (self ):
17
+ # .env ファイルから環境変数を読み込む
18
+ self ._load_env_file (".env" )
19
+
20
+ # 環境変数から設定を読み込む
21
+ # TEST_SIGNALING_URL (単数形) と TEST_SIGNALING_URLS (複数形) の両方をサポート
22
+ signaling_urls_env = os .getenv ("TEST_SIGNALING_URLS" , os .getenv ("TEST_SIGNALING_URL" , "" ))
23
+ self .signaling_urls = self ._parse_signaling_urls (signaling_urls_env )
24
+ self .channel_id_prefix = os .getenv ("TEST_CHANNEL_ID_PREFIX" , "" )
25
+ self .secret = os .getenv ("TEST_SECRET_KEY" )
26
+ self .api_url = self ._parse_api_url (os .getenv ("TEST_API_URL" ))
27
+ self .openh264_path = os .getenv ("OPENH264_PATH" )
28
+ self .libwebrtc_log = self ._parse_libwebrtc_log (os .getenv ("TEST_LIBWEBRTC_LOG" ))
29
+ self .channel_id_suffix = str (uuid .uuid4 ())
30
+
31
+ def _load_env_file (self , env_file : str ) -> None :
32
+ """環境変数ファイルを読み込む"""
33
+ env_path = Path (env_file )
34
+ if not env_path .exists ():
35
+ return
36
+
37
+ with open (env_path , "r" , encoding = "utf-8" ) as f :
38
+ for line in f :
39
+ line = line .strip ()
40
+ if not line or line .startswith ("#" ):
41
+ continue
42
+
43
+ if "=" in line :
44
+ key , value = line .split ("=" , 1 )
45
+ key = key .strip ()
46
+ value = value .strip ().strip ('"' ).strip ("'" )
47
+ # 環境変数が既に設定されていない場合のみ設定
48
+ if key not in os .environ :
49
+ os .environ [key ] = value
50
+
51
+ def _parse_signaling_urls (self , value : str ) -> list [str ]:
52
+ """TEST_SIGNALING_URLS が , で区切られている場合、それぞれの URL をリストに変換する"""
53
+ if not value :
54
+ return []
55
+ return [x .strip () for x in value .split ("," ) if x .strip ()]
56
+
57
+ def _parse_api_url (self , value : str | None ) -> str | None :
58
+ """API URLのバリデーション"""
59
+ if not value :
41
60
return None
42
-
43
- match v .lower ():
61
+ # 簡易的なHTTP/HTTPS URLチェック
62
+ if not (value .startswith ("http://" ) or value .startswith ("https://" )):
63
+ raise ValueError (f"Invalid API URL: { value } " )
64
+ return value
65
+
66
+ def _parse_libwebrtc_log (self , value : str | None ) -> SoraLoggingSeverity | None :
67
+ """libwebrtc_logの値をSoraLoggingSeverityに変換"""
68
+ if not value :
69
+ return None
70
+
71
+ match value .lower ():
44
72
case "verbose" :
45
73
return SoraLoggingSeverity .VERBOSE
46
74
case "info" :
@@ -52,18 +80,15 @@ def validate_libwebrtc_log(cls, v: str | None) -> SoraLoggingSeverity | None:
52
80
case "none" :
53
81
return SoraLoggingSeverity .NONE
54
82
case _:
55
- # TODO: 道の値が設定されてたらエラーにした方がいい気がする
83
+ # TODO: 未知の値が設定されてたらエラーにした方がいい気がする
56
84
return None
57
-
58
- @computed_field
85
+
59
86
@property
60
87
def channel_id (self ) -> str :
61
- """
62
- TEST_CHANNEL_ID_PREFIX と TEST_CHANNEL_ID_SUFFIX を組み合わせて channel_id を生成する
63
- """
88
+ """TEST_CHANNEL_ID_PREFIX と TEST_CHANNEL_ID_SUFFIX を組み合わせて channel_id を生成する"""
64
89
return f"{ self .channel_id_prefix } _{ self .channel_id_suffix } "
65
90
66
- def access_token (self , ** kwargs ) -> str | None :
91
+ def access_token (self , ** kwargs : Any ) -> str | None :
67
92
if self .secret is None :
68
93
return None
69
94
@@ -76,7 +101,7 @@ def access_token(self, **kwargs) -> str | None:
76
101
77
102
access_token = jwt .encode (
78
103
payload ,
79
- self .secret . get_secret_value () ,
104
+ self .secret ,
80
105
algorithm = "HS256" ,
81
106
)
82
107
@@ -85,6 +110,7 @@ def access_token(self, **kwargs) -> str | None:
85
110
86
111
def pytest_report_header (config ):
87
112
"""pytest の実行時に特定のライブラリのバージョンを追加表示"""
113
+ # config パラメータは pytest から渡されるが、この関数では使用しない
88
114
try :
89
115
version = importlib .metadata .version ("sora_sdk" )
90
116
return f"sora_sdk: { version } "
0 commit comments