wehub-resource-sync
228 lines
7.4 KiB
Python
228 lines
7.4 KiB
Python
import asyncio
|
|
import os
|
|
import tempfile
|
|
|
|
import pytest
|
|
|
|
from ragas.utils import (
|
|
async_to_sync,
|
|
batched,
|
|
check_if_sum_is_close,
|
|
create_nano_id,
|
|
get_from_dict,
|
|
get_test_directory,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
["values", "close_to", "num_places"],
|
|
[
|
|
[[0.1, 0.2, 0.3], 0.6, 1],
|
|
[[0.8, 0.1, 0.1], 1.0, 1],
|
|
[[0.94, 0.03, 0.03], 1.0, 2],
|
|
[[0.3948, 0.3948, 0.2104], 1.0, 4],
|
|
[[10.19, 10.19, 10.19], 30.57, 2],
|
|
],
|
|
)
|
|
def test_check_if_sum_is_close(values, close_to, num_places):
|
|
assert check_if_sum_is_close(values, close_to, num_places)
|
|
|
|
|
|
data_dict = {
|
|
"something": {"nested": {"key": "value"}},
|
|
"other": {"key": "value"},
|
|
"key": "value",
|
|
"another_key": "value",
|
|
"nested_key": {"key": "value"},
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
["data_dict", "key", "expected"],
|
|
[
|
|
(data_dict, "something.nested.key", "value"),
|
|
(data_dict, "other.key", "value"),
|
|
(data_dict, "something.not_there_in_key", None),
|
|
(data_dict, "something.nested.not_here", None),
|
|
],
|
|
)
|
|
def test_get_from_dict(data_dict, key, expected):
|
|
assert get_from_dict(data_dict, key) == expected
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
["camel_case_string", "expected"],
|
|
[
|
|
("myVariableName", "my_variable_name"),
|
|
("CamelCaseString", "camel_case_string"),
|
|
("AnotherCamelCaseString", "another_camel_case_string"),
|
|
],
|
|
)
|
|
def test_camel_to_snake(camel_case_string, expected):
|
|
from ragas.utils import camel_to_snake
|
|
|
|
assert camel_to_snake(camel_case_string) == expected
|
|
|
|
|
|
class TestBatched:
|
|
# Test cases for the `batched` function
|
|
@pytest.mark.parametrize(
|
|
"iterable, n, expected",
|
|
[
|
|
("ABCDEFG", 3, [("A", "B", "C"), ("D", "E", "F"), ("G",)]),
|
|
([1, 2, 3, 4, 5, 6, 7], 2, [(1, 2), (3, 4), (5, 6), (7,)]),
|
|
(range(5), 5, [(0, 1, 2, 3, 4)]),
|
|
(["a", "b", "c", "d"], 1, [("a",), ("b",), ("c",), ("d",)]),
|
|
([], 3, []), # Edge case: empty iterable
|
|
],
|
|
)
|
|
def test_batched(self, iterable, n: int, expected):
|
|
result = list(batched(iterable, n))
|
|
assert result == expected, f"Expected {expected}, but got {result}"
|
|
|
|
def test_batched_invalid_n(self):
|
|
"""Test that `batched` raises ValueError if n < 1."""
|
|
with pytest.raises(ValueError, match="n must be at least one"):
|
|
list(batched("ABCDEFG", 0)) # n = 0 should raise ValueError
|
|
|
|
@pytest.mark.parametrize(
|
|
"iterable, n, expected_type",
|
|
[
|
|
("ABCDEFG", 3, str),
|
|
([1, 2, 3], 2, int),
|
|
(["x", "y", "z"], 1, str),
|
|
],
|
|
)
|
|
def test_batched_output_type(self, iterable, n, expected_type: type):
|
|
"""Test that items in each batch maintain the original data type."""
|
|
result = list(batched(iterable, n))
|
|
for batch in result:
|
|
assert all(isinstance(item, expected_type) for item in batch)
|
|
|
|
|
|
class TestCreateNanoId:
|
|
"""Test cases for the create_nano_id function."""
|
|
|
|
def test_create_nano_id_default_size(self):
|
|
"""Test that create_nano_id generates IDs of default size (12)."""
|
|
nano_id = create_nano_id()
|
|
assert len(nano_id) == 12
|
|
assert nano_id.isalnum()
|
|
|
|
def test_create_nano_id_custom_size(self):
|
|
"""Test that create_nano_id respects custom size parameter."""
|
|
for size in [5, 8, 16, 20]:
|
|
nano_id = create_nano_id(size=size)
|
|
assert len(nano_id) == size
|
|
assert nano_id.isalnum()
|
|
|
|
def test_create_nano_id_uniqueness(self):
|
|
"""Test that create_nano_id generates unique IDs."""
|
|
ids = set()
|
|
for _ in range(100):
|
|
nano_id = create_nano_id()
|
|
assert nano_id not in ids, "Generated duplicate ID"
|
|
ids.add(nano_id)
|
|
|
|
def test_create_nano_id_alphanumeric(self):
|
|
"""Test that create_nano_id only uses alphanumeric characters."""
|
|
nano_id = create_nano_id(size=50) # Larger size for better coverage
|
|
for char in nano_id:
|
|
assert char.isalnum(), f"Non-alphanumeric character found: {char}"
|
|
|
|
|
|
class TestAsyncToSync:
|
|
"""Test cases for the async_to_sync function."""
|
|
|
|
def test_async_to_sync_basic(self):
|
|
"""Test basic async to sync conversion."""
|
|
|
|
async def async_add(a, b):
|
|
await asyncio.sleep(0.001) # Small delay to make it truly async
|
|
return a + b
|
|
|
|
sync_add = async_to_sync(async_add)
|
|
result = sync_add(3, 4)
|
|
assert result == 7
|
|
|
|
def test_async_to_sync_with_kwargs(self):
|
|
"""Test async to sync conversion with keyword arguments."""
|
|
|
|
async def async_multiply(x, multiplier=2):
|
|
await asyncio.sleep(0.001)
|
|
return x * multiplier
|
|
|
|
sync_multiply = async_to_sync(async_multiply)
|
|
result = sync_multiply(5, multiplier=3)
|
|
assert result == 15
|
|
|
|
def test_async_to_sync_exception_handling(self):
|
|
"""Test that exceptions in async functions are properly propagated."""
|
|
|
|
async def async_error():
|
|
await asyncio.sleep(0.001)
|
|
raise ValueError("Test error")
|
|
|
|
sync_error = async_to_sync(async_error)
|
|
with pytest.raises(ValueError, match="Test error"):
|
|
sync_error()
|
|
|
|
def test_async_to_sync_return_types(self):
|
|
"""Test that return types are preserved."""
|
|
|
|
async def async_return_dict():
|
|
await asyncio.sleep(0.001)
|
|
return {"key": "value", "number": 42}
|
|
|
|
sync_return_dict = async_to_sync(async_return_dict)
|
|
result = sync_return_dict()
|
|
expected = {"key": "value", "number": 42}
|
|
assert isinstance(result, dict) and result == expected
|
|
|
|
|
|
class TestGetTestDirectory:
|
|
"""Test cases for the get_test_directory function."""
|
|
|
|
def test_get_test_directory_exists(self):
|
|
"""Test that get_test_directory creates a directory that exists."""
|
|
test_dir = get_test_directory()
|
|
assert os.path.exists(test_dir)
|
|
assert os.path.isdir(test_dir)
|
|
|
|
def test_get_test_directory_in_temp(self):
|
|
"""Test that test directory is created in system temp directory."""
|
|
test_dir = get_test_directory()
|
|
temp_root = tempfile.gettempdir()
|
|
assert test_dir.startswith(temp_root)
|
|
|
|
def test_get_test_directory_unique(self):
|
|
"""Test that get_test_directory creates unique directories."""
|
|
dirs = set()
|
|
for _ in range(5):
|
|
test_dir = get_test_directory()
|
|
assert test_dir not in dirs, "Generated duplicate directory path"
|
|
dirs.add(test_dir)
|
|
|
|
def test_get_test_directory_naming_pattern(self):
|
|
"""Test that test directory follows expected naming pattern."""
|
|
test_dir = get_test_directory()
|
|
dir_name = os.path.basename(test_dir)
|
|
assert dir_name.startswith("ragas_test_")
|
|
# The suffix should be the nano_id, which is alphanumeric
|
|
suffix = dir_name[len("ragas_test_") :]
|
|
assert suffix.isalnum()
|
|
|
|
def test_get_test_directory_writable(self):
|
|
"""Test that the created test directory is writable."""
|
|
test_dir = get_test_directory()
|
|
# Try to create a file in the directory
|
|
test_file = os.path.join(test_dir, "test_file.txt")
|
|
with open(test_file, "w") as f:
|
|
f.write("test content")
|
|
|
|
# Verify file was created and has correct content
|
|
assert os.path.exists(test_file)
|
|
with open(test_file, "r") as f:
|
|
content = f.read()
|
|
assert content == "test content"
|