test(io): verify storage config and utility results

This commit is contained in:
huanghuoguoguo
2026-10-17 15:07:52 +08:00
parent bbf517edbc
commit a2066c95e7
5 changed files with 46 additions and 38 deletions
@@ -13,6 +13,7 @@ from __future__ import annotations
import pytest
import json
import yaml
from langbot.pkg.config.impls.yaml import YAMLConfigFile
from langbot.pkg.config.impls.json import JSONConfigFile
@@ -126,8 +127,7 @@ name: custom_name
assert config_file.exists()
content = config_file.read_text()
assert 'saved_app' in content
assert 'new_key' in content
assert yaml.safe_load(content) == {'name': 'saved_app', 'new_key': 'new_value'}
def test_yaml_save_sync(self, tmp_path):
"""YAML config can be saved synchronously."""
@@ -142,7 +142,7 @@ name: custom_name
assert config_file.exists()
content = config_file.read_text()
assert 'sync_saved' in content
assert yaml.safe_load(content) == {'name': 'sync_saved'}
class TestJSONConfigFile:
@@ -260,7 +260,7 @@ class TestConfigManager:
await manager.dump_config()
content = config_file.read_text()
assert 'dumped' in content
assert yaml.safe_load(content) == {'name': 'dumped', 'new_field': 'value'}
def test_config_manager_dump_sync(self, tmp_path):
"""ConfigManager can dump config synchronously."""
@@ -277,6 +277,7 @@ class TestConfigManager:
manager.dump_config_sync()
assert config_file.exists()
assert yaml.safe_load(config_file.read_text()) == {'name': 'sync_dumped'}
class TestConfigExists:
+16 -7
View File
@@ -12,8 +12,9 @@ Uses moto library to mock AWS S3 service.
from __future__ import annotations
import pytest
from unittest.mock import Mock
from unittest.mock import Mock, patch
from importlib import import_module
from botocore.exceptions import ClientError
def get_s3storage_module():
@@ -102,6 +103,7 @@ class TestS3StorageProviderWithMoto:
assert provider.s3_client is not None
assert provider.bucket_name == 'test-langbot-storage'
mock_app_with_s3_config.logger.info.assert_called()
assert [bucket['Name'] for bucket in s3_mock.list_buckets()['Buckets']] == ['test-langbot-storage']
@pytest.mark.asyncio
async def test_initialize_uses_existing_bucket(self, mock_app_with_s3_config, s3_mock):
@@ -112,11 +114,16 @@ class TestS3StorageProviderWithMoto:
s3_mock.create_bucket(Bucket='test-langbot-storage')
provider = s3storage.S3StorageProvider(mock_app_with_s3_config)
await provider.initialize()
with (
patch.object(s3storage.boto3.session.Session, 'client', return_value=s3_mock),
patch.object(s3_mock, 'head_bucket', wraps=s3_mock.head_bucket) as head_bucket,
patch.object(s3_mock, 'create_bucket', wraps=s3_mock.create_bucket) as create_bucket,
):
await provider.initialize()
assert provider.s3_client is not None
# Bucket creation log should not be called since bucket exists
# Note: moto may still call head_bucket successfully
assert provider.s3_client is s3_mock
head_bucket.assert_called_once_with(Bucket='test-langbot-storage')
create_bucket.assert_not_called()
@pytest.mark.asyncio
async def test_save_and_load_bytes(self, mock_app_with_s3_config, s3_mock):
@@ -325,8 +332,9 @@ class TestS3StorageProviderErrorHandling:
provider = s3storage.S3StorageProvider(mock_app)
await provider.initialize()
with pytest.raises(Exception):
with pytest.raises(ClientError) as exc_info:
await provider.load('nonexistent.txt')
assert exc_info.value.response['Error']['Code'] == 'NoSuchKey'
@pytest.mark.asyncio
async def test_size_nonexistent_raises_error(self, s3_mock):
@@ -350,5 +358,6 @@ class TestS3StorageProviderErrorHandling:
provider = s3storage.S3StorageProvider(mock_app)
await provider.initialize()
with pytest.raises(Exception):
with pytest.raises(ClientError) as exc_info:
await provider.size('nonexistent.txt')
assert exc_info.value.response['Error']['Code'] == '404'
+2 -2
View File
@@ -70,8 +70,8 @@ class TestGetQQImageDownloadableUrl:
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 result_url == 'http://example.com/image.jpg'
assert query == {}
def test_https_url(self):
"""Parse HTTPS URL and preserve its scheme."""
+16 -16
View File
@@ -15,13 +15,14 @@ Uses mocking for import operations to avoid actual module imports.
import pytest
import importlib
from unittest.mock import patch, MagicMock
import os
from unittest.mock import patch, MagicMock, call
class TestImportDir:
"""Test import_dir function."""
def test_calls_importlib_for_each_python_file(self, tmp_path):
def test_calls_importlib_for_each_python_file(self, tmp_path, monkeypatch):
"""Should call importlib.import_module for each .py file."""
module_dir = tmp_path / 'test_modules'
module_dir.mkdir()
@@ -33,12 +34,17 @@ class TestImportDir:
from langbot.pkg.utils import importutil
monkeypatch.setattr(importutil, '__file__', str(tmp_path / 'pkg' / 'utils' / 'importutil.py'))
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
mock_import.assert_has_calls(
[call('test_prefix.test_modules.module_a'), call('test_prefix.test_modules.module_b')],
any_order=True,
)
def test_skips_init_py(self, tmp_path):
def test_skips_init_py(self, tmp_path, monkeypatch):
"""Should skip __init__.py when importing."""
module_dir = tmp_path / 'test_modules'
module_dir.mkdir()
@@ -48,15 +54,12 @@ class TestImportDir:
from langbot.pkg.utils import importutil
monkeypatch.setattr(importutil, '__file__', str(tmp_path / 'pkg' / 'utils' / 'importutil.py'))
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
mock_import.assert_called_once_with('test_prefix.test_modules.regular')
def test_ignores_non_py_files(self, tmp_path):
def test_ignores_non_py_files(self, tmp_path, monkeypatch):
"""Should ignore non-.py files."""
module_dir = tmp_path / 'test_modules'
module_dir.mkdir()
@@ -67,10 +70,10 @@ class TestImportDir:
from langbot.pkg.utils import importutil
monkeypatch.setattr(importutil, '__file__', str(tmp_path / 'pkg' / 'utils' / 'importutil.py'))
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
mock_import.assert_called_once_with('test_prefix.test_modules.module')
class TestImportModulesInPkg:
@@ -107,7 +110,7 @@ class TestImportModulesInPkgs:
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
assert mock_import.call_args_list == [call(mock_pkg1), call(mock_pkg2)]
class TestImportDotStyleDir:
@@ -124,10 +127,7 @@ class TestImportDotStyleDir:
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
mock_import_dir.assert_called_once_with(os.path.join('my', 'pkg', 'test'))
class TestReadResourceFile:
+7 -9
View File
@@ -9,7 +9,6 @@ Tests path utility functions:
Uses tmp_path for file system isolation where applicable.
"""
import os
import pytest
@@ -45,17 +44,16 @@ class TestCheckIfSourceInstall:
class TestGetFrontendPath:
"""Test get_frontend_path function."""
def test_returns_web_dist_by_default(self):
"""Should return a path containing web/dist as default."""
def test_returns_web_dist_by_default(self, monkeypatch):
"""Should return web/dist when no frontend build exists."""
from langbot.pkg.utils import paths
paths._is_source_install = None
monkeypatch.setattr(paths, '_is_source_install', False)
monkeypatch.setattr(paths.os.path, 'exists', lambda _: False)
monkeypatch.setattr(paths.Path, 'exists', lambda _: False)
result = paths.get_frontend_path()
# The result should contain web/dist or be an absolute path to it
assert 'web/dist' in result or result.endswith('dist')
paths._is_source_install = None
assert result == 'web/dist'
def test_finds_dist_directory_in_source_mode(self, tmp_path, monkeypatch):
"""Should find web/dist when running from source mode."""
@@ -130,7 +128,7 @@ class TestGetResourcePath:
paths._is_source_install = None
result = paths.get_resource_path('templates/config.yaml')
assert os.path.exists(result)
assert result == 'templates/config.yaml'
paths._is_source_install = None