xinyin025/LangBot
1
0
Fork 0
mirror of https://github.com/langbot-app/LangBot.git synced 2026-06-18 19:44:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

test: format test suite (#2252)

Browse Source
This commit is contained in:
huanghuoguoguo
2026-06-16 03:22:29 +00:00
committed by GitHub
parent 1ae5aacc00
commit f390980d0a
92 changed files with 1658 additions and 1713 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit_tests/utils/test_funcschema.py
+1
View File
@@ -9,6 +9,7 @@ Tests cover:
Note: Do NOT use 'from __future__ import annotations' because
funcschema.py expects actual type objects, not string annotations.
"""
import pytest
from importlib import import_module
tests/unit_tests/utils/test_image.py
+30 -32
View File
@@ -20,55 +20,53 @@ class TestGetQQImageDownloadableUrl:
def test_basic_url(self):
"""Parse basic image URL."""
url = "http://example.com/image.jpg"
url = 'http://example.com/image.jpg'
result_url, query = get_qq_image_downloadable_url(url)
assert result_url == "http://example.com/image.jpg"
assert result_url == 'http://example.com/image.jpg'
assert query == {}
def test_url_with_query_params(self):
"""Parse URL with query parameters."""
url = "http://example.com/image.jpg?param1=value1&param2=value2"
url = 'http://example.com/image.jpg?param1=value1&param2=value2'
result_url, query = get_qq_image_downloadable_url(url)
assert result_url == "http://example.com/image.jpg"
assert query == {"param1": ["value1"], "param2": ["value2"]}
assert result_url == 'http://example.com/image.jpg'
assert query == {'param1': ['value1'], 'param2': ['value2']}
def test_url_with_port(self):
"""Parse URL with port number."""
url = "http://example.com:8080/image.jpg"
url = 'http://example.com:8080/image.jpg'
result_url, query = get_qq_image_downloadable_url(url)
assert result_url == "http://example.com:8080/image.jpg"
assert result_url == 'http://example.com:8080/image.jpg'
def test_url_with_path(self):
"""Parse URL with complex path."""
url = "http://example.com/path/to/image.jpg"
url = 'http://example.com/path/to/image.jpg'
result_url, query = get_qq_image_downloadable_url(url)
assert result_url == "http://example.com/path/to/image.jpg"
assert result_url == 'http://example.com/path/to/image.jpg'
def test_url_with_fragment(self):
"""Parse URL with fragment (fragment is not part of query)."""
url = "http://example.com/image.jpg#fragment"
url = 'http://example.com/image.jpg#fragment'
result_url, query = get_qq_image_downloadable_url(url)
# Fragment is not included in query string parsing
assert "http://example.com/image.jpg" in result_url
assert 'http://example.com/image.jpg' in result_url
def test_https_url(self):
"""Parse HTTPS URL and preserve its scheme."""
url = "https://example.com/image.jpg"
url = 'https://example.com/image.jpg'
result_url, query = get_qq_image_downloadable_url(url)
assert result_url == "https://example.com/image.jpg"
assert result_url == 'https://example.com/image.jpg'
assert query == {}
def test_preserves_qq_https_scheme_and_query(self):
"""QQ image URLs keep HTTPS and query parameters."""
result_url, query = get_qq_image_downloadable_url(
'https://gchat.qpic.cn/gchatpic_new/abc/0?term=2&is_origin=1'
)
result_url, query = get_qq_image_downloadable_url('https://gchat.qpic.cn/gchatpic_new/abc/0?term=2&is_origin=1')
assert result_url == 'https://gchat.qpic.cn/gchatpic_new/abc/0'
assert query == {'term': ['2'], 'is_origin': ['1']}
@@ -88,50 +86,50 @@ class TestExtractB64AndFormat:
async def test_jpeg_data_uri(self):
"""Extract base64 and format from JPEG data URI."""
# Create a simple base64 string
original_data = b"test image data"
original_data = b'test image data'
b64_data = base64.b64encode(original_data).decode()
data_uri = f"data:image/jpeg;base64,{b64_data}"
data_uri = f'data:image/jpeg;base64,{b64_data}'
result_b64, result_format = await extract_b64_and_format(data_uri)
assert result_b64 == b64_data
assert result_format == "jpeg"
assert result_format == 'jpeg'
@pytest.mark.asyncio
async def test_png_data_uri(self):
"""Extract base64 and format from PNG data URI."""
original_data = b"test png data"
original_data = b'test png data'
b64_data = base64.b64encode(original_data).decode()
data_uri = f"data:image/png;base64,{b64_data}"
data_uri = f'data:image/png;base64,{b64_data}'
result_b64, result_format = await extract_b64_and_format(data_uri)
assert result_b64 == b64_data
assert result_format == "png"
assert result_format == 'png'
@pytest.mark.asyncio
async def test_gif_data_uri(self):
"""Extract base64 and format from GIF data URI."""
original_data = b"test gif data"
original_data = b'test gif data'
b64_data = base64.b64encode(original_data).decode()
data_uri = f"data:image/gif;base64,{b64_data}"
data_uri = f'data:image/gif;base64,{b64_data}'
result_b64, result_format = await extract_b64_and_format(data_uri)
assert result_b64 == b64_data
assert result_format == "gif"
assert result_format == 'gif'
@pytest.mark.asyncio
async def test_webp_data_uri(self):
"""Extract base64 and format from WebP data URI."""
original_data = b"test webp data"
original_data = b'test webp data'
b64_data = base64.b64encode(original_data).decode()
data_uri = f"data:image/webp;base64,{b64_data}"
data_uri = f'data:image/webp;base64,{b64_data}'
result_b64, result_format = await extract_b64_and_format(data_uri)
assert result_b64 == b64_data
assert result_format == "webp"
assert result_format == 'webp'
@pytest.mark.asyncio
async def test_complex_base64(self):
@@ -139,7 +137,7 @@ class TestExtractB64AndFormat:
# Base64 can include + and / characters
original_data = bytes(range(256)) # All byte values
b64_data = base64.b64encode(original_data).decode()
data_uri = f"data:image/png;base64,{b64_data}"
data_uri = f'data:image/png;base64,{b64_data}'
result_b64, result_format = await extract_b64_and_format(data_uri)
@@ -150,9 +148,9 @@ class TestExtractB64AndFormat:
@pytest.mark.asyncio
async def test_empty_base64(self):
"""Handle empty base64 string."""
data_uri = "data:image/png;base64,"
data_uri = 'data:image/png;base64,'
result_b64, result_format = await extract_b64_and_format(data_uri)
assert result_b64 == ""
assert result_format == "png"
assert result_b64 == ''
assert result_format == 'png'
tests/unit_tests/utils/test_importutil.py
+46 -46
View File
@@ -23,52 +23,52 @@ class TestImportDir:
def test_calls_importlib_for_each_python_file(self, tmp_path):
"""Should call importlib.import_module for each .py file."""
module_dir = tmp_path / "test_modules"
module_dir = tmp_path / 'test_modules'
module_dir.mkdir()
(module_dir / "__init__.py").write_text("")
(module_dir / "module_a.py").write_text("VALUE_A = 'a'\n")
(module_dir / "module_b.py").write_text("VALUE_B = 'b'\n")
(module_dir / "readme.txt").write_text("not a module")
(module_dir / '__init__.py').write_text('')
(module_dir / 'module_a.py').write_text("VALUE_A = 'a'\n")
(module_dir / 'module_b.py').write_text("VALUE_B = 'b'\n")
(module_dir / 'readme.txt').write_text('not a module')
from langbot.pkg.utils import importutil
with patch.object(importlib, "import_module") as mock_import:
importutil.import_dir(str(module_dir), path_prefix="test_prefix.")
with patch.object(importlib, 'import_module') as mock_import:
importutil.import_dir(str(module_dir), path_prefix='test_prefix.')
# Should call import_module for each .py file (excluding __init__.py)
assert mock_import.call_count == 2
def test_skips_init_py(self, tmp_path):
"""Should skip __init__.py when importing."""
module_dir = tmp_path / "test_modules"
module_dir = tmp_path / 'test_modules'
module_dir.mkdir()
(module_dir / "__init__.py").write_text("")
(module_dir / "regular.py").write_text("VALUE = 1\n")
(module_dir / '__init__.py').write_text('')
(module_dir / 'regular.py').write_text('VALUE = 1\n')
from langbot.pkg.utils import importutil
with patch.object(importlib, "import_module") as mock_import:
importutil.import_dir(str(module_dir), path_prefix="test_prefix.")
with patch.object(importlib, 'import_module') as mock_import:
importutil.import_dir(str(module_dir), path_prefix='test_prefix.')
# __init__.py should be skipped
mock_import.assert_called_once()
# The call should not include __init__
call_args = mock_import.call_args[0][0]
assert "__init__" not in call_args
assert '__init__' not in call_args
def test_ignores_non_py_files(self, tmp_path):
"""Should ignore non-.py files."""
module_dir = tmp_path / "test_modules"
module_dir = tmp_path / 'test_modules'
module_dir.mkdir()
(module_dir / "module.py").write_text("VALUE = 1\n")
(module_dir / "readme.txt").write_text("text")
(module_dir / "data.json").write_text("{}")
(module_dir / 'module.py').write_text('VALUE = 1\n')
(module_dir / 'readme.txt').write_text('text')
(module_dir / 'data.json').write_text('{}')
from langbot.pkg.utils import importutil
with patch.object(importlib, "import_module") as mock_import:
importutil.import_dir(str(module_dir), path_prefix="test_prefix.")
with patch.object(importlib, 'import_module') as mock_import:
importutil.import_dir(str(module_dir), path_prefix='test_prefix.')
# Only .py files should be imported
assert mock_import.call_count == 1
@@ -79,14 +79,14 @@ class TestImportModulesInPkg:
def test_imports_modules_from_package(self, tmp_path):
"""Should import all modules from a package object."""
mock_pkg = MagicMock()
mock_pkg.__file__ = str(tmp_path / "__init__.py")
mock_pkg.__file__ = str(tmp_path / '__init__.py')
(tmp_path / "__init__.py").write_text("")
(tmp_path / "mod1.py").write_text("MOD1 = 1\n")
(tmp_path / '__init__.py').write_text('')
(tmp_path / 'mod1.py').write_text('MOD1 = 1\n')
from langbot.pkg.utils import importutil
with patch.object(importutil, "import_dir") as mock_import_dir:
with patch.object(importutil, 'import_dir') as mock_import_dir:
importutil.import_modules_in_pkg(mock_pkg)
mock_import_dir.assert_called_once()
call_path = mock_import_dir.call_args[0][0]
@@ -101,11 +101,11 @@ class TestImportModulesInPkgs:
from langbot.pkg.utils import importutil
mock_pkg1 = MagicMock()
mock_pkg1.__file__ = "/path/to/pkg1/__init__.py"
mock_pkg1.__file__ = '/path/to/pkg1/__init__.py'
mock_pkg2 = MagicMock()
mock_pkg2.__file__ = "/path/to/pkg2/__init__.py"
mock_pkg2.__file__ = '/path/to/pkg2/__init__.py'
with patch.object(importutil, "import_modules_in_pkg") as mock_import:
with patch.object(importutil, 'import_modules_in_pkg') as mock_import:
importutil.import_modules_in_pkgs([mock_pkg1, mock_pkg2])
assert mock_import.call_count == 2
@@ -116,18 +116,18 @@ class TestImportDotStyleDir:
def test_converts_dot_notation_to_path(self, tmp_path):
"""Should convert dot notation to path and import."""
# Create structure matching the dot notation
(tmp_path / "my").mkdir()
(tmp_path / "my" / "pkg").mkdir()
(tmp_path / "my" / "pkg" / "test").mkdir()
(tmp_path / 'my').mkdir()
(tmp_path / 'my' / 'pkg').mkdir()
(tmp_path / 'my' / 'pkg' / 'test').mkdir()
from langbot.pkg.utils import importutil
with patch.object(importutil, "import_dir") as mock_import_dir:
importutil.import_dot_style_dir("my.pkg.test")
with patch.object(importutil, 'import_dir') as mock_import_dir:
importutil.import_dot_style_dir('my.pkg.test')
# The path should be converted using os.path.join
call_path = mock_import_dir.call_args[0][0]
# Should contain the path components joined
assert "my" in call_path
assert 'my' in call_path
class TestReadResourceFile:
@@ -137,16 +137,16 @@ class TestReadResourceFile:
"""Should read content from a resource file."""
from langbot.pkg.utils import importutil
content = importutil.read_resource_file("templates/config.yaml")
assert "admins:" in content
assert "edition: community" in content
content = importutil.read_resource_file('templates/config.yaml')
assert 'admins:' in content
assert 'edition: community' in content
def test_raises_for_nonexistent_file(self):
"""Should raise exception for non-existent resource file."""
from langbot.pkg.utils import importutil
with pytest.raises((FileNotFoundError, Exception)):
importutil.read_resource_file("nonexistent/path/file.txt")
importutil.read_resource_file('nonexistent/path/file.txt')
class TestReadResourceFileBytes:
@@ -156,16 +156,16 @@ class TestReadResourceFileBytes:
"""Should read content as bytes from a resource file."""
from langbot.pkg.utils import importutil
content = importutil.read_resource_file_bytes("templates/config.yaml")
assert b"admins:" in content
assert b"edition: community" in content
content = importutil.read_resource_file_bytes('templates/config.yaml')
assert b'admins:' in content
assert b'edition: community' in content
def test_raises_for_nonexistent_file_bytes(self):
"""Should raise exception for non-existent resource file."""
from langbot.pkg.utils import importutil
with pytest.raises((FileNotFoundError, Exception)):
importutil.read_resource_file_bytes("nonexistent/path/file.txt")
importutil.read_resource_file_bytes('nonexistent/path/file.txt')
class TestListResourceFiles:
@@ -175,9 +175,9 @@ class TestListResourceFiles:
"""Should list files in a resource directory."""
from langbot.pkg.utils import importutil
files = importutil.list_resource_files("templates")
assert "config.yaml" in files
assert "default-pipeline-config.json" in files
files = importutil.list_resource_files('templates')
assert 'config.yaml' in files
assert 'default-pipeline-config.json' in files
assert all(isinstance(file, str) for file in files)
def test_raises_for_nonexistent_directory(self):
@@ -185,8 +185,8 @@ class TestListResourceFiles:
from langbot.pkg.utils import importutil
with pytest.raises((FileNotFoundError, Exception)):
importutil.list_resource_files("nonexistent_directory_xyz")
importutil.list_resource_files('nonexistent_directory_xyz')
if __name__ == "__main__":
pytest.main([__file__, "-v"])
if __name__ == '__main__':
pytest.main([__file__, '-v'])
tests/unit_tests/utils/test_platform.py
+2 -1
View File
@@ -5,6 +5,7 @@ Tests cover:
- Docker environment detection
- WebSocket plugin runtime mode
"""
from __future__ import annotations
import os
@@ -86,4 +87,4 @@ class TestGetPlatform:
assert platform_module.use_websocket_to_connect_plugin_runtime() is True
# Restore
platform_module.standalone_runtime = original
platform_module.standalone_runtime = original
tests/unit_tests/utils/test_proxy.py
+17 -11
View File
@@ -60,10 +60,12 @@ class TestProxyManager:
async def test_initialize_config_overrides_env(self):
"""Config proxy overrides environment variables."""
mock_app = self._create_mock_app(proxy_config={
'http': 'http://config-proxy:8080',
'https': 'https://config-proxy:8443',
})
mock_app = self._create_mock_app(
proxy_config={
'http': 'http://config-proxy:8080',
'https': 'https://config-proxy:8443',
}
)
with patch.dict(os.environ, {'HTTP_PROXY': 'http://env-proxy:8080'}):
pm = ProxyManager(mock_app)
@@ -74,10 +76,12 @@ class TestProxyManager:
async def test_initialize_sets_env_variables(self):
"""initialize sets proxy to environment variables."""
mock_app = self._create_mock_app(proxy_config={
'http': 'http://test-proxy:8080',
'https': 'https://test-proxy:8443',
})
mock_app = self._create_mock_app(
proxy_config={
'http': 'http://test-proxy:8080',
'https': 'https://test-proxy:8443',
}
)
pm = ProxyManager(mock_app)
await pm.initialize()
@@ -143,9 +147,11 @@ class TestProxyManager:
async def test_initialize_http_only_config(self):
"""initialize handles http-only config."""
mock_app = self._create_mock_app(proxy_config={
'http': 'http://http-only:8080',
})
mock_app = self._create_mock_app(
proxy_config={
'http': 'http://http-only:8080',
}
)
# Clear any existing proxy env vars
env_backup = {}
tests/unit_tests/utils/test_runner.py
+69 -88
View File
@@ -29,63 +29,63 @@ class TestGetRunnerCategory:
def test_empty_url_returns_unknown(self):
"""Empty or None URL should return UNKNOWN."""
assert get_runner_category("test", "") == RunnerCategory.UNKNOWN
assert get_runner_category("test", None) == RunnerCategory.UNKNOWN
assert get_runner_category('test', '') == RunnerCategory.UNKNOWN
assert get_runner_category('test', None) == RunnerCategory.UNKNOWN
def test_localhost_returns_local(self):
"""localhost URL should be categorized as LOCAL."""
assert get_runner_category("test", "http://localhost:3000") == RunnerCategory.LOCAL
assert get_runner_category("test", "https://localhost") == RunnerCategory.LOCAL
assert get_runner_category('test', 'http://localhost:3000') == RunnerCategory.LOCAL
assert get_runner_category('test', 'https://localhost') == RunnerCategory.LOCAL
def test_127_0_0_1_returns_local(self):
"""127.0.0.1 URL should be categorized as LOCAL."""
assert get_runner_category("test", "http://127.0.0.1:8080") == RunnerCategory.LOCAL
assert get_runner_category("test", "https://127.0.0.1") == RunnerCategory.LOCAL
assert get_runner_category('test', 'http://127.0.0.1:8080') == RunnerCategory.LOCAL
assert get_runner_category('test', 'https://127.0.0.1') == RunnerCategory.LOCAL
def test_0_0_0_0_returns_local(self):
"""0.0.0.0 URL should be categorized as LOCAL."""
assert get_runner_category("test", "http://0.0.0.0:8080") == RunnerCategory.LOCAL
assert get_runner_category('test', 'http://0.0.0.0:8080') == RunnerCategory.LOCAL
def test_private_ip_192_168_returns_local(self):
"""192.168.x.x private IP should be categorized as LOCAL."""
assert get_runner_category("test", "http://192.168.1.1:3000") == RunnerCategory.LOCAL
assert get_runner_category("test", "http://192.168.0.100") == RunnerCategory.LOCAL
assert get_runner_category('test', 'http://192.168.1.1:3000') == RunnerCategory.LOCAL
assert get_runner_category('test', 'http://192.168.0.100') == RunnerCategory.LOCAL
def test_private_ip_10_returns_local(self):
"""10.x.x.x private IP should be categorized as LOCAL."""
assert get_runner_category("test", "http://10.0.0.1:8080") == RunnerCategory.LOCAL
assert get_runner_category("test", "http://10.255.255.255") == RunnerCategory.LOCAL
assert get_runner_category('test', 'http://10.0.0.1:8080') == RunnerCategory.LOCAL
assert get_runner_category('test', 'http://10.255.255.255') == RunnerCategory.LOCAL
def test_private_ip_172_16_31_returns_local(self):
"""172.16.x.x - 172.31.x.x private IP range should be categorized as LOCAL."""
assert get_runner_category("test", "http://172.16.0.1:8080") == RunnerCategory.LOCAL
assert get_runner_category("test", "http://172.20.0.1") == RunnerCategory.LOCAL
assert get_runner_category("test", "http://172.31.255.255") == RunnerCategory.LOCAL
assert get_runner_category('test', 'http://172.16.0.1:8080') == RunnerCategory.LOCAL
assert get_runner_category('test', 'http://172.20.0.1') == RunnerCategory.LOCAL
assert get_runner_category('test', 'http://172.31.255.255') == RunnerCategory.LOCAL
def test_n8n_cloud_returns_cloud(self):
"""n8n.cloud domain should be categorized as CLOUD."""
assert get_runner_category("test", "https://myinstance.n8n.cloud") == RunnerCategory.CLOUD
assert get_runner_category("test", "https://test.n8n.io") == RunnerCategory.CLOUD
assert get_runner_category('test', 'https://myinstance.n8n.cloud') == RunnerCategory.CLOUD
assert get_runner_category('test', 'https://test.n8n.io') == RunnerCategory.CLOUD
def test_dify_cloud_returns_cloud(self):
"""Dify cloud domains should be categorized as CLOUD."""
assert get_runner_category("test", "https://api.dify.ai/v1") == RunnerCategory.CLOUD
assert get_runner_category("test", "https://cloud.dify.ai") == RunnerCategory.CLOUD
assert get_runner_category('test', 'https://api.dify.ai/v1') == RunnerCategory.CLOUD
assert get_runner_category('test', 'https://cloud.dify.ai') == RunnerCategory.CLOUD
def test_coze_cloud_returns_cloud(self):
"""Coze domains should be categorized as CLOUD."""
assert get_runner_category("test", "https://api.coze.com") == RunnerCategory.CLOUD
assert get_runner_category("test", "https://api.coze.cn") == RunnerCategory.CLOUD
assert get_runner_category('test', 'https://api.coze.com') == RunnerCategory.CLOUD
assert get_runner_category('test', 'https://api.coze.cn') == RunnerCategory.CLOUD
def test_langflow_cloud_returns_cloud(self):
"""Langflow domains should be categorized as CLOUD."""
assert get_runner_category("test", "https://cloud.langflow.ai") == RunnerCategory.CLOUD
assert get_runner_category("test", "https://test.langflow.org") == RunnerCategory.CLOUD
assert get_runner_category('test', 'https://cloud.langflow.ai') == RunnerCategory.CLOUD
assert get_runner_category('test', 'https://test.langflow.org') == RunnerCategory.CLOUD
def test_other_url_returns_cloud(self):
"""Other URLs should default to CLOUD category."""
assert get_runner_category("test", "https://example.com") == RunnerCategory.CLOUD
assert get_runner_category("test", "https://myserver.example.org") == RunnerCategory.CLOUD
assert get_runner_category('test', 'https://example.com') == RunnerCategory.CLOUD
assert get_runner_category('test', 'https://myserver.example.org') == RunnerCategory.CLOUD
@pytest.mark.parametrize(
'runner_url',
@@ -101,7 +101,7 @@ class TestGetRunnerCategory:
)
def test_invalid_urls_return_unknown(self, runner_url):
"""Invalid or incomplete URLs should return UNKNOWN."""
assert get_runner_category("test", runner_url) == RunnerCategory.UNKNOWN
assert get_runner_category('test', runner_url) == RunnerCategory.UNKNOWN
def test_urlparse_exception_returns_unknown(self):
"""Exception during URL parsing should return UNKNOWN."""
@@ -109,15 +109,15 @@ class TestGetRunnerCategory:
from langbot.pkg.utils import runner
def mock_urlparse(url):
raise Exception("URL parsing failed")
raise Exception('URL parsing failed')
with patch("langbot.pkg.utils.runner.urlparse", side_effect=mock_urlparse):
result = runner.get_runner_category("test", "http://example.com")
with patch('langbot.pkg.utils.runner.urlparse', side_effect=mock_urlparse):
result = runner.get_runner_category('test', 'http://example.com')
assert result == RunnerCategory.UNKNOWN
def test_url_without_scheme_returns_unknown(self):
"""URL without scheme should return UNKNOWN."""
assert get_runner_category("test", "example.com") == RunnerCategory.UNKNOWN
assert get_runner_category('test', 'example.com') == RunnerCategory.UNKNOWN
@pytest.mark.parametrize(
'runner_url',
@@ -146,20 +146,21 @@ class TestGetRunnerCategory:
"""Domain names that only look like private IP prefixes should not be LOCAL."""
assert get_runner_category('langflow-api', runner_url) == RunnerCategory.CLOUD
class TestIsCloudRunner:
"""Test is_cloud_runner helper function."""
def test_cloud_runner_returns_true(self):
"""Cloud URL should return True."""
assert is_cloud_runner("test", "https://api.dify.ai") is True
assert is_cloud_runner('test', 'https://api.dify.ai') is True
def test_local_runner_returns_false(self):
"""Local URL should return False."""
assert is_cloud_runner("test", "http://localhost:3000") is False
assert is_cloud_runner('test', 'http://localhost:3000') is False
def test_unknown_returns_false(self):
"""Unknown category should return False."""
assert is_cloud_runner("test", None) is False
assert is_cloud_runner('test', None) is False
class TestIsLocalRunner:
@@ -167,15 +168,15 @@ class TestIsLocalRunner:
def test_local_runner_returns_true(self):
"""Local URL should return True."""
assert is_local_runner("test", "http://localhost:3000") is True
assert is_local_runner('test', 'http://localhost:3000') is True
def test_cloud_runner_returns_false(self):
"""Cloud URL should return False."""
assert is_local_runner("test", "https://api.dify.ai") is False
assert is_local_runner('test', 'https://api.dify.ai') is False
def test_unknown_returns_false(self):
"""Unknown category should return False."""
assert is_local_runner("test", None) is False
assert is_local_runner('test', None) is False
class TestGetRunnerInfo:
@@ -183,17 +184,17 @@ class TestGetRunnerInfo:
def test_returns_dict_with_expected_keys(self):
"""Should return dict with name, url, and category keys."""
info = get_runner_info("my-runner", "http://localhost:3000")
assert "name" in info
assert "url" in info
assert "category" in info
info = get_runner_info('my-runner', 'http://localhost:3000')
assert 'name' in info
assert 'url' in info
assert 'category' in info
def test_includes_correct_values(self):
"""Should include correct values in dict."""
info = get_runner_info("my-runner", "http://localhost:3000")
assert info["name"] == "my-runner"
assert info["url"] == "http://localhost:3000"
assert info["category"] == RunnerCategory.LOCAL
info = get_runner_info('my-runner', 'http://localhost:3000')
assert info['name'] == 'my-runner'
assert info['url'] == 'http://localhost:3000'
assert info['category'] == RunnerCategory.LOCAL
class TestExtractRunnerUrl:
@@ -203,74 +204,58 @@ class TestExtractRunnerUrl:
"""Should extract base-url from dify-service-api config."""
runner = Mock()
runner.pipeline_config = {}
pipeline_config = {
"ai": {
"dify-service-api": {"base-url": "https://api.dify.ai"}
}
}
url = extract_runner_url("dify-service-api", runner, pipeline_config)
assert url == "https://api.dify.ai"
pipeline_config = {'ai': {'dify-service-api': {'base-url': 'https://api.dify.ai'}}}
url = extract_runner_url('dify-service-api', runner, pipeline_config)
assert url == 'https://api.dify.ai'
def test_n8n_service_api_extracts_url(self):
"""Should extract webhook-url from n8n-service-api config."""
runner = Mock()
runner.pipeline_config = {}
pipeline_config = {
"ai": {
"n8n-service-api": {"webhook-url": "https://my.n8n.cloud/webhook"}
}
}
url = extract_runner_url("n8n-service-api", runner, pipeline_config)
assert url == "https://my.n8n.cloud/webhook"
pipeline_config = {'ai': {'n8n-service-api': {'webhook-url': 'https://my.n8n.cloud/webhook'}}}
url = extract_runner_url('n8n-service-api', runner, pipeline_config)
assert url == 'https://my.n8n.cloud/webhook'
def test_coze_api_extracts_url(self):
"""Should extract api-base from coze-api config."""
runner = Mock()
runner.pipeline_config = {}
pipeline_config = {
"ai": {
"coze-api": {"api-base": "https://api.coze.com"}
}
}
url = extract_runner_url("coze-api", runner, pipeline_config)
assert url == "https://api.coze.com"
pipeline_config = {'ai': {'coze-api': {'api-base': 'https://api.coze.com'}}}
url = extract_runner_url('coze-api', runner, pipeline_config)
assert url == 'https://api.coze.com'
def test_langflow_api_extracts_url(self):
"""Should extract base-url from langflow-api config."""
runner = Mock()
runner.pipeline_config = {}
pipeline_config = {
"ai": {
"langflow-api": {"base-url": "https://cloud.langflow.ai"}
}
}
url = extract_runner_url("langflow-api", runner, pipeline_config)
assert url == "https://cloud.langflow.ai"
pipeline_config = {'ai': {'langflow-api': {'base-url': 'https://cloud.langflow.ai'}}}
url = extract_runner_url('langflow-api', runner, pipeline_config)
assert url == 'https://cloud.langflow.ai'
def test_unknown_runner_returns_none(self):
"""Unknown runner name should return None."""
runner = Mock()
runner.pipeline_config = {}
pipeline_config = {}
url = extract_runner_url("unknown-runner", runner, pipeline_config)
url = extract_runner_url('unknown-runner', runner, pipeline_config)
assert url is None
def test_none_runner_returns_none(self):
"""None runner should return None."""
url = extract_runner_url("test", None, {})
url = extract_runner_url('test', None, {})
assert url is None
def test_runner_without_pipeline_config_returns_none(self):
"""Runner without pipeline_config attribute should return None."""
runner = Mock(spec=[]) # Empty spec means no attributes
url = extract_runner_url("test", runner, {})
url = extract_runner_url('test', runner, {})
assert url is None
def test_none_pipeline_config_returns_none(self):
"""None pipeline_config should return None."""
runner = Mock()
runner.pipeline_config = {}
url = extract_runner_url("dify-service-api", runner, None)
url = extract_runner_url('dify-service-api', runner, None)
assert url is None
def test_missing_ai_config_returns_none(self):
@@ -278,7 +263,7 @@ class TestExtractRunnerUrl:
runner = Mock()
runner.pipeline_config = {}
pipeline_config = {}
url = extract_runner_url("dify-service-api", runner, pipeline_config)
url = extract_runner_url('dify-service-api', runner, pipeline_config)
assert url is None
@@ -289,19 +274,15 @@ class TestGetRunnerCategoryFromRunner:
"""Should extract URL and return correct category."""
runner = Mock()
runner.pipeline_config = {}
pipeline_config = {
"ai": {
"dify-service-api": {"base-url": "https://api.dify.ai"}
}
}
category = get_runner_category_from_runner("dify-service-api", runner, pipeline_config)
pipeline_config = {'ai': {'dify-service-api': {'base-url': 'https://api.dify.ai'}}}
category = get_runner_category_from_runner('dify-service-api', runner, pipeline_config)
assert category == RunnerCategory.CLOUD
def test_returns_unknown_for_missing_url(self):
"""Should return UNKNOWN when URL cannot be extracted."""
runner = Mock()
runner.pipeline_config = {}
category = get_runner_category_from_runner("unknown", runner, {})
category = get_runner_category_from_runner('unknown', runner, {})
assert category == RunnerCategory.UNKNOWN
@@ -310,9 +291,9 @@ class TestConstants:
def test_runner_category_constants(self):
"""RunnerCategory should have LOCAL, CLOUD, UNKNOWN."""
assert RunnerCategory.LOCAL == "local"
assert RunnerCategory.CLOUD == "cloud"
assert RunnerCategory.UNKNOWN == "unknown"
assert RunnerCategory.LOCAL == 'local'
assert RunnerCategory.CLOUD == 'cloud'
assert RunnerCategory.UNKNOWN == 'unknown'
def test_cloud_domains_not_empty(self):
"""CLOUD_DOMAINS should not be empty."""
@@ -323,5 +304,5 @@ class TestConstants:
assert len(LOCAL_PATTERNS) > 0
if __name__ == "__main__":
pytest.main([__file__, "-v"])
if __name__ == '__main__':
pytest.main([__file__, '-v'])