mirror of
https://github.com/Ichinga-Samuel/aiomql.git
synced 2026-08-25 01:38:04 +00:00
Update tests and docs across core, lib, and contrib modules
This commit is contained in:
Ichinga Samuel
@@ -0,0 +1,395 @@
|
||||
"""Comprehensive tests for the DB ORM module.
|
||||
|
||||
Tests cover:
|
||||
- DB initialization with dataclass
|
||||
- Table creation and column definitions
|
||||
- CRUD operations (save, get, filter, update, delete)
|
||||
- Type mapping (Python to SQLite)
|
||||
- Primary key handling
|
||||
- Raw SQL execution with validation
|
||||
- Data sanitization
|
||||
"""
|
||||
|
||||
import os
|
||||
import pytest
|
||||
import tempfile
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
from aiomql.core.db import DB
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def temp_db_path():
|
||||
"""Creates a temporary database file path."""
|
||||
fd, path = tempfile.mkstemp(suffix=".db")
|
||||
os.close(fd)
|
||||
yield path
|
||||
if os.path.exists(path):
|
||||
os.remove(path)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def setup_db_config(temp_db_path, monkeypatch):
|
||||
"""Sets up DB config with temp database."""
|
||||
from aiomql.core.config import Config
|
||||
config = Config()
|
||||
config.db_name = temp_db_path
|
||||
monkeypatch.setenv("DB_NAME", temp_db_path)
|
||||
yield temp_db_path
|
||||
|
||||
|
||||
@dataclass
|
||||
class TestModel(DB):
|
||||
"""Test model for DB tests."""
|
||||
id: int = field(metadata={"PRIMARY KEY": True})
|
||||
name: str = ""
|
||||
value: float = 0.0
|
||||
|
||||
|
||||
@dataclass
|
||||
class SimpleModel(DB):
|
||||
"""Simple model without primary key."""
|
||||
name: str = ""
|
||||
count: int = 0
|
||||
|
||||
|
||||
class TestDBInitialization:
|
||||
"""Tests for DB initialization."""
|
||||
|
||||
def test_dataclass_model_creates_table(self, setup_db_config):
|
||||
"""Test dataclass model creates table on instantiation."""
|
||||
record = TestModel(id=1, name="test", value=1.0)
|
||||
assert record is not None
|
||||
|
||||
def test_init_sets_config(self, setup_db_config):
|
||||
"""Test __new__ sets config."""
|
||||
record = TestModel(id=1, name="test", value=1.0)
|
||||
assert hasattr(record, "config")
|
||||
|
||||
def test_table_name_defaults_to_class_name(self, setup_db_config):
|
||||
"""Test table name defaults to lowercase class name."""
|
||||
record = TestModel(id=1, name="test", value=1.0)
|
||||
assert TestModel._table == "testmodel"
|
||||
|
||||
|
||||
class TestDBTypeMapping:
|
||||
"""Tests for Python to SQLite type mapping."""
|
||||
|
||||
def test_str_maps_to_text(self):
|
||||
"""Test str maps to TEXT."""
|
||||
assert DB.types(str) == "TEXT"
|
||||
|
||||
def test_int_maps_to_integer(self):
|
||||
"""Test int maps to INTEGER."""
|
||||
assert DB.types(int) == "INTEGER"
|
||||
|
||||
def test_float_maps_to_real(self):
|
||||
"""Test float maps to REAL."""
|
||||
assert DB.types(float) == "REAL"
|
||||
|
||||
def test_bool_maps_to_boolean(self):
|
||||
"""Test bool maps to BOOLEAN."""
|
||||
assert DB.types(bool) == "BOOLEAN"
|
||||
|
||||
def test_bytes_maps_to_blob(self):
|
||||
"""Test bytes maps to BLOB."""
|
||||
assert DB.types(bytes) == "BLOB"
|
||||
|
||||
def test_unknown_type_maps_to_text(self):
|
||||
"""Test unknown type maps to TEXT."""
|
||||
assert DB.types(list) == "TEXT"
|
||||
|
||||
|
||||
class TestDBSanitize:
|
||||
"""Tests for SQL identifier sanitization."""
|
||||
|
||||
def test_valid_identifier(self):
|
||||
"""Test valid identifier is quoted."""
|
||||
result = DB.sanitize("valid_name")
|
||||
assert result == '"valid_name"'
|
||||
|
||||
def test_identifier_starting_with_underscore(self):
|
||||
"""Test identifier starting with underscore."""
|
||||
result = DB.sanitize("_valid")
|
||||
assert result == '"_valid"'
|
||||
|
||||
def test_invalid_identifier_raises(self):
|
||||
"""Test invalid identifier raises ValueError."""
|
||||
with pytest.raises(ValueError):
|
||||
DB.sanitize("invalid-name")
|
||||
|
||||
def test_identifier_with_numbers(self):
|
||||
"""Test identifier with numbers."""
|
||||
result = DB.sanitize("name123")
|
||||
assert result == '"name123"'
|
||||
|
||||
def test_identifier_starting_with_number_raises(self):
|
||||
"""Test identifier starting with number raises."""
|
||||
with pytest.raises(ValueError):
|
||||
DB.sanitize("123invalid")
|
||||
|
||||
|
||||
class TestDBCRUDOperations:
|
||||
"""Tests for DB CRUD operations."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def setup_model(self, setup_db_config):
|
||||
"""Reset model state before each test."""
|
||||
TestModel._initialized = False
|
||||
TestModel._table = ""
|
||||
yield
|
||||
|
||||
def test_save_inserts_record(self, setup_db_config):
|
||||
"""Test save inserts new record."""
|
||||
record = TestModel(id=1, name="test", value=1.0)
|
||||
record.save()
|
||||
|
||||
result = TestModel.get(id=1)
|
||||
assert result is not None
|
||||
assert result.name == "test"
|
||||
|
||||
def test_get_returns_record(self, setup_db_config):
|
||||
"""Test get returns matching record."""
|
||||
record = TestModel(id=1, name="test", value=1.0)
|
||||
record.save()
|
||||
|
||||
result = TestModel.get(id=1)
|
||||
assert result.id == 1
|
||||
assert result.name == "test"
|
||||
|
||||
def test_get_returns_none_for_no_match(self, setup_db_config):
|
||||
"""Test get returns None for no match."""
|
||||
TestModel(id=1, name="test", value=1.0) # Initialize table
|
||||
result = TestModel.get(id=999)
|
||||
assert result is None
|
||||
|
||||
def test_filter_returns_all_matching(self, setup_db_config):
|
||||
"""Test filter returns all matching records."""
|
||||
TestModel(id=1, name="test", value=1.0).save()
|
||||
TestModel(id=2, name="test", value=2.0).save()
|
||||
TestModel(id=3, name="other", value=3.0).save()
|
||||
|
||||
results = TestModel.filter(name="test")
|
||||
assert len(results) == 2
|
||||
|
||||
def test_filter_returns_all_when_no_criteria(self, setup_db_config):
|
||||
"""Test filter returns all records when no criteria."""
|
||||
TestModel(id=1, name="test", value=1.0).save()
|
||||
TestModel(id=2, name="other", value=2.0).save()
|
||||
|
||||
results = TestModel.filter()
|
||||
assert len(results) == 2
|
||||
|
||||
def test_all_returns_all_records(self, setup_db_config):
|
||||
"""Test all returns all records."""
|
||||
TestModel(id=1, name="test1", value=1.0).save()
|
||||
TestModel(id=2, name="test2", value=2.0).save()
|
||||
|
||||
results = TestModel.all()
|
||||
assert len(results) == 2
|
||||
|
||||
def test_all_with_limit(self, setup_db_config):
|
||||
"""Test all with limit returns limited records."""
|
||||
TestModel(id=1, name="test1", value=1.0).save()
|
||||
TestModel(id=2, name="test2", value=2.0).save()
|
||||
TestModel(id=3, name="test3", value=3.0).save()
|
||||
|
||||
results = TestModel.all(limit=2)
|
||||
assert len(results) == 2
|
||||
|
||||
def test_clear_removes_all_records(self, setup_db_config):
|
||||
"""Test clear removes all records."""
|
||||
TestModel(id=1, name="test1", value=1.0).save()
|
||||
TestModel(id=2, name="test2", value=2.0).save()
|
||||
|
||||
TestModel.clear()
|
||||
results = TestModel.all()
|
||||
assert len(results) == 0
|
||||
|
||||
def test_update_modifies_records(self, setup_db_config):
|
||||
"""Test update modifies matching records."""
|
||||
TestModel(id=1, name="old", value=1.0).save()
|
||||
|
||||
TestModel.update({"name": "new"}, id=1)
|
||||
result = TestModel.get(id=1)
|
||||
assert result.name == "new"
|
||||
|
||||
|
||||
class TestDBPrimaryKey:
|
||||
"""Tests for primary key handling."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def setup_model(self, setup_db_config):
|
||||
"""Reset model state before each test."""
|
||||
TestModel._initialized = False
|
||||
TestModel._table = ""
|
||||
yield
|
||||
|
||||
def test_pk_property_returns_pk_field(self, setup_db_config):
|
||||
"""Test pk property returns primary key field name and value."""
|
||||
record = TestModel(id=42, name="test", value=1.0)
|
||||
pk_name, pk_value = record.pk
|
||||
assert pk_name == "id"
|
||||
assert pk_value == 42
|
||||
|
||||
|
||||
class TestDBAsDict:
|
||||
"""Tests for asdict functionality."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def setup_model(self, setup_db_config):
|
||||
"""Reset model state before each test."""
|
||||
TestModel._initialized = False
|
||||
TestModel._table = ""
|
||||
yield
|
||||
|
||||
def test_asdict_returns_dict(self, setup_db_config):
|
||||
"""Test asdict returns dictionary."""
|
||||
record = TestModel(id=1, name="test", value=1.0)
|
||||
result = record.asdict()
|
||||
assert isinstance(result, dict)
|
||||
assert result["id"] == 1
|
||||
assert result["name"] == "test"
|
||||
assert result["value"] == 1.0
|
||||
|
||||
|
||||
class TestDBFields:
|
||||
"""Tests for fields class method."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def setup_model(self, setup_db_config):
|
||||
"""Reset model state before each test."""
|
||||
TestModel._initialized = False
|
||||
TestModel._table = ""
|
||||
yield
|
||||
|
||||
def test_fields_returns_field_names(self, setup_db_config):
|
||||
"""Test fields returns list of field names."""
|
||||
TestModel(id=1, name="test", value=1.0) # Initialize
|
||||
field_names = TestModel.fields()
|
||||
assert "id" in field_names
|
||||
assert "name" in field_names
|
||||
assert "value" in field_names
|
||||
|
||||
|
||||
class TestDBDropTable:
|
||||
"""Tests for drop_table functionality."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def setup_model(self, setup_db_config):
|
||||
"""Reset model state before each test."""
|
||||
TestModel._initialized = False
|
||||
TestModel._table = ""
|
||||
yield
|
||||
|
||||
def test_drop_table_removes_table(self, setup_db_config):
|
||||
"""Test drop_table removes the table."""
|
||||
record = TestModel(id=1, name="test", value=1.0)
|
||||
record.save()
|
||||
|
||||
TestModel.drop_table()
|
||||
|
||||
# Re-initializing should create fresh table
|
||||
TestModel._initialized = False
|
||||
TestModel._table = ""
|
||||
record2 = TestModel(id=1, name="new", value=2.0)
|
||||
record2.save()
|
||||
assert TestModel.all()[0].name == "new"
|
||||
|
||||
|
||||
class TestDBExecuteRaw:
|
||||
"""Tests for execute_raw SQL execution."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def setup_model(self, setup_db_config):
|
||||
"""Reset model state before each test."""
|
||||
TestModel._initialized = False
|
||||
TestModel._table = ""
|
||||
TestModel(id=1, name="test1", value=1.0).save()
|
||||
TestModel(id=2, name="test2", value=2.0).save()
|
||||
yield
|
||||
|
||||
def test_execute_raw_select(self, setup_db_config):
|
||||
"""Test execute_raw with SELECT query."""
|
||||
results = TestModel.execute_raw(
|
||||
"SELECT * FROM testmodel WHERE id = ?",
|
||||
(1,)
|
||||
)
|
||||
assert len(results) == 1
|
||||
assert results[0].name == "test1"
|
||||
|
||||
def test_execute_raw_with_named_params(self, setup_db_config):
|
||||
"""Test execute_raw with named parameters."""
|
||||
results = TestModel.execute_raw(
|
||||
"SELECT * FROM testmodel WHERE name = :name",
|
||||
{"name": "test1"}
|
||||
)
|
||||
assert len(results) == 1
|
||||
|
||||
def test_execute_raw_write_requires_flag(self, setup_db_config):
|
||||
"""Test execute_raw write operations require allow_write."""
|
||||
with pytest.raises(PermissionError):
|
||||
TestModel.execute_raw(
|
||||
"UPDATE testmodel SET name = ? WHERE id = ?",
|
||||
("updated", 1)
|
||||
)
|
||||
|
||||
def test_execute_raw_write_with_flag(self, setup_db_config):
|
||||
"""Test execute_raw write with allow_write=True."""
|
||||
affected = TestModel.execute_raw(
|
||||
"UPDATE testmodel SET name = ? WHERE id = ?",
|
||||
("updated", 1),
|
||||
allow_write=True
|
||||
)
|
||||
assert affected == 1
|
||||
|
||||
result = TestModel.get(id=1)
|
||||
assert result.name == "updated"
|
||||
|
||||
def test_execute_raw_rejects_dangerous_patterns(self, setup_db_config):
|
||||
"""Test execute_raw rejects dangerous SQL patterns."""
|
||||
with pytest.raises(ValueError):
|
||||
TestModel.execute_raw("SELECT * FROM testmodel; DROP TABLE testmodel")
|
||||
|
||||
def test_execute_raw_rejects_comments(self, setup_db_config):
|
||||
"""Test execute_raw rejects SQL comments."""
|
||||
with pytest.raises(ValueError):
|
||||
TestModel.execute_raw("SELECT * FROM testmodel -- comment")
|
||||
|
||||
def test_execute_raw_rejects_multiple_statements(self, setup_db_config):
|
||||
"""Test execute_raw rejects multiple statements."""
|
||||
with pytest.raises(ValueError):
|
||||
TestModel.execute_raw("SELECT 1; SELECT 2")
|
||||
|
||||
def test_execute_raw_empty_query_raises(self, setup_db_config):
|
||||
"""Test execute_raw with empty query raises."""
|
||||
with pytest.raises(ValueError):
|
||||
TestModel.execute_raw("")
|
||||
|
||||
def test_execute_raw_invalid_params_type_raises(self, setup_db_config):
|
||||
"""Test execute_raw with invalid params type raises."""
|
||||
with pytest.raises(ValueError):
|
||||
TestModel.execute_raw("SELECT * FROM testmodel", "invalid")
|
||||
|
||||
def test_execute_raw_unsupported_statement_raises(self, setup_db_config):
|
||||
"""Test execute_raw with unsupported statement raises."""
|
||||
with pytest.raises(ValueError):
|
||||
TestModel.execute_raw("CREATE TABLE newtable (id INTEGER)")
|
||||
|
||||
|
||||
class TestDBDictFactory:
|
||||
"""Tests for dict_factory row conversion."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def setup_model(self, setup_db_config):
|
||||
"""Reset model state before each test."""
|
||||
TestModel._initialized = False
|
||||
TestModel._table = ""
|
||||
yield
|
||||
|
||||
def test_dict_factory_returns_class_instances(self, setup_db_config):
|
||||
"""Test dict_factory converts rows to class instances."""
|
||||
TestModel(id=1, name="test", value=1.0).save()
|
||||
|
||||
results = TestModel.all()
|
||||
assert isinstance(results[0], TestModel)
|
||||
@@ -0,0 +1,415 @@
|
||||
"""Comprehensive tests for the State module.
|
||||
|
||||
Tests cover:
|
||||
- State initialization (singleton pattern)
|
||||
- MutableMapping interface
|
||||
- Key-value CRUD operations
|
||||
- Data persistence via commit/load
|
||||
- Singleton behavior
|
||||
- Autocommit functionality
|
||||
- Flush behavior
|
||||
"""
|
||||
|
||||
import os
|
||||
import pytest
|
||||
import tempfile
|
||||
from collections.abc import MutableMapping
|
||||
|
||||
from aiomql.core.state import State
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def reset_state_singleton():
|
||||
"""Reset State singleton before each test."""
|
||||
# Remove singleton instance to ensure clean state for each test
|
||||
if hasattr(State, "_instance"):
|
||||
delattr(State, "_instance")
|
||||
if hasattr(State, "_data"):
|
||||
delattr(State, "_data")
|
||||
State._initialized = False
|
||||
yield
|
||||
# Cleanup after test
|
||||
if hasattr(State, "_instance"):
|
||||
delattr(State, "_instance")
|
||||
if hasattr(State, "_data"):
|
||||
delattr(State, "_data")
|
||||
State._initialized = False
|
||||
|
||||
|
||||
class TestStateInitialization:
|
||||
"""Tests for State initialization."""
|
||||
|
||||
@pytest.fixture
|
||||
def temp_db(self):
|
||||
"""Creates a temporary database file."""
|
||||
fd, path = tempfile.mkstemp(suffix=".db")
|
||||
os.close(fd)
|
||||
yield path
|
||||
if os.path.exists(path):
|
||||
os.remove(path)
|
||||
|
||||
def test_init_creates_state(self, temp_db):
|
||||
"""Test State can be initialized."""
|
||||
state = State(db_name=temp_db)
|
||||
assert isinstance(state, State)
|
||||
|
||||
def test_init_with_initial_data(self, temp_db):
|
||||
"""Test State can be initialized with data."""
|
||||
initial_data = {"key1": "value1", "key2": "value2"}
|
||||
state = State(db_name=temp_db, data=initial_data, flush=True)
|
||||
assert state["key1"] == "value1"
|
||||
assert state["key2"] == "value2"
|
||||
|
||||
def test_init_with_flush(self, temp_db):
|
||||
"""Test State flush clears existing data."""
|
||||
# Create initial state with data
|
||||
state1 = State(db_name=temp_db, data={"existing": "data"}, flush=True)
|
||||
state1.commit()
|
||||
|
||||
# Reset singleton
|
||||
delattr(State, "_instance")
|
||||
delattr(State, "_data")
|
||||
State._initialized = False
|
||||
|
||||
# Create new state with flush
|
||||
state2 = State(db_name=temp_db, flush=True)
|
||||
assert "existing" not in state2
|
||||
|
||||
def test_init_default_autocommit(self, temp_db):
|
||||
"""Test State has autocommit False by default."""
|
||||
state = State(db_name=temp_db)
|
||||
assert state.autocommit is False
|
||||
|
||||
def test_init_autocommit_true(self, temp_db):
|
||||
"""Test State can be initialized with autocommit=True."""
|
||||
state = State(db_name=temp_db, autocommit=True)
|
||||
assert state.autocommit is True
|
||||
|
||||
|
||||
class TestStateSingleton:
|
||||
"""Tests for State singleton behavior."""
|
||||
|
||||
@pytest.fixture
|
||||
def temp_db(self):
|
||||
"""Creates a temporary database file."""
|
||||
fd, path = tempfile.mkstemp(suffix=".db")
|
||||
os.close(fd)
|
||||
yield path
|
||||
if os.path.exists(path):
|
||||
os.remove(path)
|
||||
|
||||
def test_singleton_returns_same_instance(self, temp_db):
|
||||
"""Test State returns same instance."""
|
||||
state1 = State(db_name=temp_db)
|
||||
state2 = State(db_name=temp_db)
|
||||
assert state1 is state2
|
||||
|
||||
def test_singleton_shares_data(self, temp_db):
|
||||
"""Test State instances share data."""
|
||||
state1 = State(db_name=temp_db)
|
||||
state1["key"] = "value"
|
||||
state2 = State(db_name=temp_db)
|
||||
assert state2["key"] == "value"
|
||||
|
||||
|
||||
class TestStateMutableMappingInterface:
|
||||
"""Tests for MutableMapping interface implementation."""
|
||||
|
||||
@pytest.fixture
|
||||
def temp_db(self):
|
||||
"""Creates a temporary database file."""
|
||||
fd, path = tempfile.mkstemp(suffix=".db")
|
||||
os.close(fd)
|
||||
yield path
|
||||
if os.path.exists(path):
|
||||
os.remove(path)
|
||||
|
||||
@pytest.fixture
|
||||
def state(self, temp_db):
|
||||
"""Creates a State instance."""
|
||||
return State(db_name=temp_db)
|
||||
|
||||
def test_is_mutable_mapping(self, state):
|
||||
"""Test State implements MutableMapping."""
|
||||
assert isinstance(state, MutableMapping)
|
||||
|
||||
def test_setitem_getitem(self, state):
|
||||
"""Test setting and getting items."""
|
||||
state["key"] = "value"
|
||||
assert state["key"] == "value"
|
||||
|
||||
def test_delitem(self, state):
|
||||
"""Test deleting items."""
|
||||
state["key"] = "value"
|
||||
del state["key"]
|
||||
assert "key" not in state
|
||||
|
||||
def test_delitem_raises_keyerror(self, state):
|
||||
"""Test deleting nonexistent key raises KeyError."""
|
||||
with pytest.raises(KeyError):
|
||||
del state["nonexistent"]
|
||||
|
||||
def test_len(self, state):
|
||||
"""Test len returns correct count."""
|
||||
assert len(state) == 0
|
||||
state["key1"] = "value1"
|
||||
state["key2"] = "value2"
|
||||
assert len(state) == 2
|
||||
|
||||
def test_contains(self, state):
|
||||
"""Test 'in' operator."""
|
||||
state["key"] = "value"
|
||||
assert "key" in state
|
||||
assert "nonexistent" not in state
|
||||
|
||||
def test_iter(self, state):
|
||||
"""Test iteration over keys."""
|
||||
state["key1"] = "value1"
|
||||
state["key2"] = "value2"
|
||||
keys = list(state)
|
||||
assert "key1" in keys
|
||||
assert "key2" in keys
|
||||
|
||||
def test_getitem_raises_keyerror(self, state):
|
||||
"""Test getting nonexistent key raises KeyError."""
|
||||
with pytest.raises(KeyError):
|
||||
_ = state["nonexistent"]
|
||||
|
||||
|
||||
class TestStateOperations:
|
||||
"""Tests for State CRUD operations."""
|
||||
|
||||
@pytest.fixture
|
||||
def temp_db(self):
|
||||
"""Creates a temporary database file."""
|
||||
fd, path = tempfile.mkstemp(suffix=".db")
|
||||
os.close(fd)
|
||||
yield path
|
||||
if os.path.exists(path):
|
||||
os.remove(path)
|
||||
|
||||
@pytest.fixture
|
||||
def state(self, temp_db):
|
||||
"""Creates a State instance."""
|
||||
return State(db_name=temp_db)
|
||||
|
||||
def test_get_existing_key(self, state):
|
||||
"""Test get returns value for existing key."""
|
||||
state["key"] = "value"
|
||||
assert state.get("key") == "value"
|
||||
|
||||
def test_get_nonexistent_key_default(self, state):
|
||||
"""Test get returns default for nonexistent key."""
|
||||
assert state.get("nonexistent") is None
|
||||
assert state.get("nonexistent", "default") == "default"
|
||||
|
||||
def test_setdefault_existing_key(self, state):
|
||||
"""Test setdefault returns existing value."""
|
||||
state["key"] = "existing"
|
||||
result = state.setdefault("key", "default")
|
||||
assert result == "existing"
|
||||
|
||||
def test_setdefault_nonexistent_key(self, state):
|
||||
"""Test setdefault sets and returns default."""
|
||||
result = state.setdefault("key", "default")
|
||||
assert result == "default"
|
||||
assert state["key"] == "default"
|
||||
|
||||
def test_update_with_dict(self, state):
|
||||
"""Test update with dictionary."""
|
||||
state.update({"key1": "value1", "key2": "value2"})
|
||||
assert state["key1"] == "value1"
|
||||
assert state["key2"] == "value2"
|
||||
|
||||
def test_update_with_kwargs(self, state):
|
||||
"""Test update with keyword arguments."""
|
||||
state.update({"key1": "value1"}, key2="value2")
|
||||
assert state["key1"] == "value1"
|
||||
assert state["key2"] == "value2"
|
||||
|
||||
def test_pop_existing_key(self, state):
|
||||
"""Test pop returns and removes value."""
|
||||
state["key"] = "value"
|
||||
result = state.pop("key")
|
||||
assert result == "value"
|
||||
assert "key" not in state
|
||||
|
||||
def test_pop_nonexistent_key_default(self, state):
|
||||
"""Test pop returns default for nonexistent key."""
|
||||
result = state.pop("nonexistent", "default")
|
||||
assert result == "default"
|
||||
|
||||
def test_pop_nonexistent_key_raises(self, state):
|
||||
"""Test pop raises KeyError without default."""
|
||||
with pytest.raises(KeyError):
|
||||
state.pop("nonexistent")
|
||||
|
||||
def test_keys(self, state):
|
||||
"""Test keys returns dict_keys."""
|
||||
state["key1"] = "value1"
|
||||
state["key2"] = "value2"
|
||||
keys = state.keys()
|
||||
assert "key1" in keys
|
||||
assert "key2" in keys
|
||||
|
||||
def test_values(self, state):
|
||||
"""Test values returns dict_values."""
|
||||
state["key1"] = "value1"
|
||||
state["key2"] = "value2"
|
||||
values = state.values()
|
||||
assert "value1" in values
|
||||
assert "value2" in values
|
||||
|
||||
def test_items(self, state):
|
||||
"""Test items returns dict_items."""
|
||||
state["key1"] = "value1"
|
||||
items = state.items()
|
||||
assert ("key1", "value1") in items
|
||||
|
||||
|
||||
class TestStateDataProperty:
|
||||
"""Tests for State data property."""
|
||||
|
||||
@pytest.fixture
|
||||
def temp_db(self):
|
||||
"""Creates a temporary database file."""
|
||||
fd, path = tempfile.mkstemp(suffix=".db")
|
||||
os.close(fd)
|
||||
yield path
|
||||
if os.path.exists(path):
|
||||
os.remove(path)
|
||||
|
||||
def test_data_returns_dict(self, temp_db):
|
||||
"""Test data property returns dictionary."""
|
||||
state = State(db_name=temp_db)
|
||||
state["key1"] = "value1"
|
||||
data = state.data
|
||||
assert isinstance(data, dict)
|
||||
|
||||
def test_data_setter(self, temp_db):
|
||||
"""Test data property can be set."""
|
||||
state = State(db_name=temp_db)
|
||||
state.data = {"key": "value"}
|
||||
assert state["key"] == "value"
|
||||
|
||||
def test_data_setter_requires_dict(self, temp_db):
|
||||
"""Test data setter raises on non-dict."""
|
||||
state = State(db_name=temp_db)
|
||||
with pytest.raises(AssertionError):
|
||||
state.data = "not a dict"
|
||||
|
||||
|
||||
class TestStatePersistence:
|
||||
"""Tests for State persistence functionality."""
|
||||
|
||||
@pytest.fixture
|
||||
def temp_db(self):
|
||||
"""Creates a temporary database file."""
|
||||
fd, path = tempfile.mkstemp(suffix=".db")
|
||||
os.close(fd)
|
||||
yield path
|
||||
if os.path.exists(path):
|
||||
os.remove(path)
|
||||
|
||||
def test_commit_persists_data(self, temp_db):
|
||||
"""Test commit persists data to database."""
|
||||
state1 = State(db_name=temp_db)
|
||||
state1["key"] = "value"
|
||||
state1.commit()
|
||||
|
||||
# Reset singleton
|
||||
delattr(State, "_instance")
|
||||
delattr(State, "_data")
|
||||
State._initialized = False
|
||||
|
||||
# Create new state and verify data loaded
|
||||
state2 = State(db_name=temp_db)
|
||||
assert state2["key"] == "value"
|
||||
|
||||
def test_flush_clears_and_sets_data(self, temp_db):
|
||||
"""Test flush clears and optionally sets new data."""
|
||||
state = State(db_name=temp_db)
|
||||
state["key1"] = "value1"
|
||||
state.flush({"key2": "value2"})
|
||||
assert "key1" not in state
|
||||
assert state["key2"] == "value2"
|
||||
|
||||
def test_flush_with_no_data(self, temp_db):
|
||||
"""Test flush with no data clears state."""
|
||||
state = State(db_name=temp_db)
|
||||
state["key"] = "value"
|
||||
state.flush()
|
||||
assert len(state) == 0
|
||||
|
||||
async def test_acommit(self, temp_db):
|
||||
"""Test async commit."""
|
||||
state1 = State(db_name=temp_db)
|
||||
state1["key"] = "value"
|
||||
await state1.acommit()
|
||||
|
||||
# Reset singleton
|
||||
delattr(State, "_instance")
|
||||
delattr(State, "_data")
|
||||
State._initialized = False
|
||||
|
||||
state2 = State(db_name=temp_db)
|
||||
assert state2["key"] == "value"
|
||||
|
||||
|
||||
class TestStateRepr:
|
||||
"""Tests for State repr."""
|
||||
|
||||
@pytest.fixture
|
||||
def temp_db(self):
|
||||
"""Creates a temporary database file."""
|
||||
fd, path = tempfile.mkstemp(suffix=".db")
|
||||
os.close(fd)
|
||||
yield path
|
||||
if os.path.exists(path):
|
||||
os.remove(path)
|
||||
|
||||
def test_repr(self, temp_db):
|
||||
"""Test repr returns data representation."""
|
||||
state = State(db_name=temp_db)
|
||||
state["key"] = "value"
|
||||
assert "key" in repr(state)
|
||||
assert "value" in repr(state)
|
||||
|
||||
|
||||
class TestStateAutocommit:
|
||||
"""Tests for State autocommit functionality."""
|
||||
|
||||
@pytest.fixture
|
||||
def temp_db(self):
|
||||
"""Creates a temporary database file."""
|
||||
fd, path = tempfile.mkstemp(suffix=".db")
|
||||
os.close(fd)
|
||||
yield path
|
||||
if os.path.exists(path):
|
||||
os.remove(path)
|
||||
|
||||
def test_autocommit_on_setitem(self, temp_db):
|
||||
"""Test autocommit commits on setitem."""
|
||||
state = State(db_name=temp_db, autocommit=True)
|
||||
state["key"] = "value"
|
||||
|
||||
# Reset singleton
|
||||
delattr(State, "_instance")
|
||||
delattr(State, "_data")
|
||||
State._initialized = False
|
||||
|
||||
state2 = State(db_name=temp_db)
|
||||
assert state2.get("key") == "value"
|
||||
|
||||
def test_autocommit_on_delitem(self, temp_db):
|
||||
"""Test autocommit commits on delitem."""
|
||||
state = State(db_name=temp_db, autocommit=True, data={"key": "value"}, flush=True)
|
||||
del state["key"]
|
||||
|
||||
# Reset singleton
|
||||
delattr(State, "_instance")
|
||||
delattr(State, "_data")
|
||||
State._initialized = False
|
||||
|
||||
state2 = State(db_name=temp_db)
|
||||
assert "key" not in state2
|
||||
@@ -0,0 +1,399 @@
|
||||
"""Comprehensive tests for the Store module.
|
||||
|
||||
Tests cover:
|
||||
- Store initialization and configuration
|
||||
- MutableMapping interface (dict-like operations)
|
||||
- Key-value CRUD operations
|
||||
- Iteration methods (keys, values, items)
|
||||
- Autocommit functionality
|
||||
- Data persistence
|
||||
- Flush behavior
|
||||
"""
|
||||
|
||||
import os
|
||||
import pytest
|
||||
import tempfile
|
||||
import gc
|
||||
from collections.abc import MutableMapping
|
||||
|
||||
from aiomql.core.store import Store
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def reset_store_connection():
|
||||
"""Reset Store class-level connection before each test."""
|
||||
# Reset the class-level cached connection
|
||||
Store._conn = None
|
||||
yield
|
||||
# Cleanup after test - force garbage collection to close any open connections
|
||||
gc.collect()
|
||||
if Store._conn is not None:
|
||||
try:
|
||||
Store._conn.close()
|
||||
except Exception:
|
||||
pass
|
||||
Store._conn = None
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def temp_db():
|
||||
"""Creates a temporary database file."""
|
||||
fd, path = tempfile.mkstemp(suffix=".db")
|
||||
os.close(fd)
|
||||
yield path
|
||||
# Force garbage collection to release any file handles
|
||||
gc.collect()
|
||||
# Try to remove the file multiple times with a small delay
|
||||
for _ in range(3):
|
||||
try:
|
||||
if os.path.exists(path):
|
||||
os.remove(path)
|
||||
break
|
||||
except PermissionError:
|
||||
gc.collect()
|
||||
import time
|
||||
time.sleep(0.1)
|
||||
|
||||
|
||||
class TestStoreInitialization:
|
||||
"""Tests for Store initialization."""
|
||||
|
||||
def test_init_creates_store(self, temp_db):
|
||||
"""Test Store can be initialized."""
|
||||
store = Store(db_name=temp_db)
|
||||
assert isinstance(store, Store)
|
||||
store.conn.close()
|
||||
|
||||
def test_init_creates_table(self, temp_db):
|
||||
"""Test Store creates table on initialization."""
|
||||
store = Store(db_name=temp_db, table_name="test_table")
|
||||
# Table should exist - verify by checking we can use the store
|
||||
store["key"] = "value"
|
||||
assert store["key"] == "value"
|
||||
store.conn.close()
|
||||
|
||||
def test_init_with_initial_data(self, temp_db):
|
||||
"""Test Store can be initialized with data."""
|
||||
initial_data = {"key1": "value1", "key2": "value2"}
|
||||
store = Store(db_name=temp_db, data=initial_data)
|
||||
assert store["key1"] == "value1"
|
||||
assert store["key2"] == "value2"
|
||||
store.conn.close()
|
||||
|
||||
def test_init_with_flush(self, temp_db):
|
||||
"""Test Store flush clears existing data."""
|
||||
# Create initial store with data (autocommit=True creates its own connection)
|
||||
store1 = Store(db_name=temp_db, data={"existing": "data"}, autocommit=True)
|
||||
store1.conn.close()
|
||||
|
||||
# Create new store with flush
|
||||
store2 = Store(db_name=temp_db, flush=True, autocommit=True)
|
||||
assert "existing" not in store2
|
||||
store2.conn.close()
|
||||
|
||||
def test_init_default_autocommit(self, temp_db):
|
||||
"""Test Store has autocommit True by default."""
|
||||
store = Store(db_name=temp_db)
|
||||
assert store.autocommit is True
|
||||
store.conn.close()
|
||||
|
||||
def test_init_autocommit_false(self, temp_db):
|
||||
"""Test Store can be initialized with autocommit=False."""
|
||||
store = Store(db_name=temp_db, autocommit=False)
|
||||
assert store.autocommit is False
|
||||
store.conn.close()
|
||||
|
||||
def test_init_default_table_name(self, temp_db):
|
||||
"""Test Store uses 'store' as default table name."""
|
||||
store = Store(db_name=temp_db)
|
||||
assert store.table_name == "store"
|
||||
store.conn.close()
|
||||
|
||||
def test_init_custom_table_name(self, temp_db):
|
||||
"""Test Store can use custom table name."""
|
||||
store = Store(db_name=temp_db, table_name="custom_table")
|
||||
assert store.table_name == "custom_table"
|
||||
store.conn.close()
|
||||
|
||||
|
||||
class TestStoreMutableMappingInterface:
|
||||
"""Tests for MutableMapping interface implementation."""
|
||||
|
||||
@pytest.fixture
|
||||
def store(self, temp_db):
|
||||
"""Creates a Store instance."""
|
||||
s = Store(db_name=temp_db)
|
||||
yield s
|
||||
s.conn.close()
|
||||
|
||||
def test_is_mutable_mapping(self, store):
|
||||
"""Test Store implements MutableMapping."""
|
||||
assert isinstance(store, MutableMapping)
|
||||
|
||||
def test_setitem_getitem(self, store):
|
||||
"""Test setting and getting items."""
|
||||
store["key"] = "value"
|
||||
assert store["key"] == "value"
|
||||
|
||||
def test_delitem(self, store):
|
||||
"""Test deleting items."""
|
||||
store["key"] = "value"
|
||||
del store["key"]
|
||||
assert "key" not in store
|
||||
|
||||
def test_delitem_raises_keyerror(self, store):
|
||||
"""Test deleting nonexistent key raises KeyError."""
|
||||
with pytest.raises(KeyError):
|
||||
del store["nonexistent"]
|
||||
|
||||
def test_len(self, store):
|
||||
"""Test len returns correct count."""
|
||||
assert len(store) == 0
|
||||
store["key1"] = "value1"
|
||||
store["key2"] = "value2"
|
||||
assert len(store) == 2
|
||||
|
||||
def test_contains(self, store):
|
||||
"""Test 'in' operator."""
|
||||
store["key"] = "value"
|
||||
assert "key" in store
|
||||
assert "nonexistent" not in store
|
||||
|
||||
def test_iter(self, store):
|
||||
"""Test iteration over keys."""
|
||||
store["key1"] = "value1"
|
||||
store["key2"] = "value2"
|
||||
keys = list(store)
|
||||
assert "key1" in keys
|
||||
assert "key2" in keys
|
||||
|
||||
def test_getitem_raises_keyerror(self, store):
|
||||
"""Test getting nonexistent key raises KeyError."""
|
||||
with pytest.raises(KeyError):
|
||||
_ = store["nonexistent"]
|
||||
|
||||
|
||||
class TestStoreOperations:
|
||||
"""Tests for Store CRUD operations."""
|
||||
|
||||
@pytest.fixture
|
||||
def store(self, temp_db):
|
||||
"""Creates a Store instance."""
|
||||
s = Store(db_name=temp_db)
|
||||
yield s
|
||||
s.conn.close()
|
||||
|
||||
def test_get_existing_key(self, store):
|
||||
"""Test get returns value for existing key."""
|
||||
store["key"] = "value"
|
||||
assert store.get("key") == "value"
|
||||
|
||||
def test_get_nonexistent_key_default(self, store):
|
||||
"""Test get returns default for nonexistent key."""
|
||||
assert store.get("nonexistent") is None
|
||||
assert store.get("nonexistent", "default") == "default"
|
||||
|
||||
def test_setdefault_existing_key(self, store):
|
||||
"""Test setdefault returns existing value."""
|
||||
store["key"] = "existing"
|
||||
result = store.setdefault("key", "default")
|
||||
assert result == "existing"
|
||||
|
||||
def test_setdefault_nonexistent_key(self, store):
|
||||
"""Test setdefault sets and returns default."""
|
||||
result = store.setdefault("key", "default")
|
||||
assert result == "default"
|
||||
assert store["key"] == "default"
|
||||
|
||||
def test_update_with_dict(self, store):
|
||||
"""Test update with dictionary."""
|
||||
store.update({"key1": "value1", "key2": "value2"})
|
||||
assert store["key1"] == "value1"
|
||||
assert store["key2"] == "value2"
|
||||
|
||||
def test_update_with_kwargs(self, store):
|
||||
"""Test update with keyword arguments."""
|
||||
store.update(key1="value1", key2="value2")
|
||||
assert store["key1"] == "value1"
|
||||
assert store["key2"] == "value2"
|
||||
|
||||
def test_pop_existing_key(self, store):
|
||||
"""Test pop returns and removes value."""
|
||||
store["key"] = "value"
|
||||
result = store.pop("key")
|
||||
assert result == "value"
|
||||
assert "key" not in store
|
||||
|
||||
def test_pop_nonexistent_key_default(self, store):
|
||||
"""Test pop returns default for nonexistent key."""
|
||||
result = store.pop("nonexistent", "default")
|
||||
assert result == "default"
|
||||
|
||||
def test_pop_nonexistent_key_raises(self, store):
|
||||
"""Test pop raises KeyError without default."""
|
||||
with pytest.raises(KeyError):
|
||||
store.pop("nonexistent")
|
||||
|
||||
def test_clear(self, store):
|
||||
"""Test clear removes all items."""
|
||||
store["key1"] = "value1"
|
||||
store["key2"] = "value2"
|
||||
store.clear()
|
||||
assert len(store) == 0
|
||||
|
||||
|
||||
class TestStoreIterationMethods:
|
||||
"""Tests for Store iteration methods."""
|
||||
|
||||
@pytest.fixture
|
||||
def store(self, temp_db):
|
||||
"""Creates a Store with test data."""
|
||||
s = Store(db_name=temp_db, data={"key1": "value1", "key2": "value2"})
|
||||
yield s
|
||||
s.conn.close()
|
||||
|
||||
def test_keys(self, store):
|
||||
"""Test keys returns list of keys."""
|
||||
keys = store.keys()
|
||||
assert isinstance(keys, list)
|
||||
assert "key1" in keys
|
||||
assert "key2" in keys
|
||||
|
||||
def test_values(self, store):
|
||||
"""Test values returns list of values."""
|
||||
values = store.values()
|
||||
assert isinstance(values, list)
|
||||
assert "value1" in values
|
||||
assert "value2" in values
|
||||
|
||||
def test_items(self, store):
|
||||
"""Test items returns list of tuples."""
|
||||
items = store.items()
|
||||
assert isinstance(items, list)
|
||||
assert ("key1", "value1") in items
|
||||
assert ("key2", "value2") in items
|
||||
|
||||
def test_iterkeys(self, store):
|
||||
"""Test iterkeys yields keys."""
|
||||
keys = list(store.iterkeys())
|
||||
assert "key1" in keys
|
||||
assert "key2" in keys
|
||||
|
||||
def test_itervalues(self, store):
|
||||
"""Test itervalues yields values."""
|
||||
values = list(store.itervalues())
|
||||
assert "value1" in values
|
||||
assert "value2" in values
|
||||
|
||||
def test_iteritems(self, store):
|
||||
"""Test iteritems yields key-value tuples."""
|
||||
items = list(store.iteritems())
|
||||
assert ("key1", "value1") in items
|
||||
assert ("key2", "value2") in items
|
||||
|
||||
|
||||
class TestStoreDataProperty:
|
||||
"""Tests for Store data property."""
|
||||
|
||||
def test_data_returns_dict(self, temp_db):
|
||||
"""Test data property returns dictionary."""
|
||||
store = Store(db_name=temp_db, data={"key1": "value1"})
|
||||
data = store.data
|
||||
assert isinstance(data, dict)
|
||||
store.conn.close()
|
||||
|
||||
def test_data_contains_all_items(self, temp_db):
|
||||
"""Test data contains all stored items."""
|
||||
store = Store(db_name=temp_db, data={"key1": "value1", "key2": "value2"})
|
||||
data = store.data
|
||||
assert data == {"key1": "value1", "key2": "value2"}
|
||||
store.conn.close()
|
||||
|
||||
|
||||
class TestStoreCommit:
|
||||
"""Tests for Store commit functionality."""
|
||||
|
||||
def test_commit_persists_data(self, temp_db):
|
||||
"""Test commit persists data to database."""
|
||||
store = Store(db_name=temp_db, autocommit=False)
|
||||
store["key"] = "value"
|
||||
store.commit()
|
||||
store.conn.close()
|
||||
|
||||
# Create new store instance and verify data persisted
|
||||
store2 = Store(db_name=temp_db, autocommit=True)
|
||||
assert store2["key"] == "value"
|
||||
store2.conn.close()
|
||||
|
||||
def test_autocommit_persists_immediately(self, temp_db):
|
||||
"""Test autocommit persists data immediately."""
|
||||
store = Store(db_name=temp_db, autocommit=True)
|
||||
store["key"] = "value"
|
||||
store.conn.close()
|
||||
|
||||
# Create new store instance and verify data persisted
|
||||
store2 = Store(db_name=temp_db, autocommit=True)
|
||||
assert store2["key"] == "value"
|
||||
store2.conn.close()
|
||||
|
||||
async def test_acommit(self, temp_db):
|
||||
"""Test async commit."""
|
||||
store = Store(db_name=temp_db, autocommit=False)
|
||||
store["key"] = "value"
|
||||
await store.acommit()
|
||||
store.conn.close()
|
||||
|
||||
store2 = Store(db_name=temp_db, autocommit=True)
|
||||
assert store2["key"] == "value"
|
||||
store2.conn.close()
|
||||
|
||||
def test_classmethod_commit(self, temp_db):
|
||||
"""Test commit as classmethod."""
|
||||
store = Store(db_name=temp_db, autocommit=False)
|
||||
store["key"] = "value"
|
||||
# Use the classmethod commit
|
||||
Store.commit()
|
||||
store.conn.close()
|
||||
|
||||
store2 = Store(db_name=temp_db, autocommit=True)
|
||||
assert store2["key"] == "value"
|
||||
store2.conn.close()
|
||||
|
||||
|
||||
class TestStoreRepr:
|
||||
"""Tests for Store repr."""
|
||||
|
||||
def test_repr(self, temp_db):
|
||||
"""Test repr returns class name."""
|
||||
store = Store(db_name=temp_db)
|
||||
assert repr(store) == "Store()"
|
||||
store.conn.close()
|
||||
|
||||
|
||||
class TestStoreConnectionHandling:
|
||||
"""Tests for Store connection handling."""
|
||||
|
||||
def test_autocommit_true_creates_own_connection(self, temp_db):
|
||||
"""Test autocommit=True creates its own connection."""
|
||||
store = Store(db_name=temp_db, autocommit=True)
|
||||
assert store.conn is not None
|
||||
# With autocommit=True, it should NOT use the class-level _conn
|
||||
store["key"] = "value"
|
||||
assert store["key"] == "value"
|
||||
store.conn.close()
|
||||
|
||||
def test_autocommit_false_uses_class_connection(self, temp_db):
|
||||
"""Test autocommit=False uses class-level connection."""
|
||||
store = Store(db_name=temp_db, autocommit=False)
|
||||
assert store.conn is Store._conn
|
||||
store["key"] = "value"
|
||||
store.conn.commit()
|
||||
assert store["key"] == "value"
|
||||
store.conn.close()
|
||||
|
||||
def test_connection_classmethod(self, temp_db):
|
||||
"""Test connection classmethod caches connection."""
|
||||
conn1 = Store.connection(temp_db)
|
||||
conn2 = Store.connection(temp_db)
|
||||
assert conn1 is conn2
|
||||
conn1.close()
|
||||
@@ -0,0 +1,225 @@
|
||||
"""Comprehensive tests for the utils module.
|
||||
|
||||
Tests cover:
|
||||
- sleep async function (live mode)
|
||||
- sleep_sync function (live mode)
|
||||
- auto_commit function
|
||||
|
||||
Note: Backtesting-related functions are excluded from these tests.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import time
|
||||
import pytest
|
||||
from unittest.mock import patch, MagicMock, AsyncMock
|
||||
|
||||
from aiomql.core.utils import sleep, sleep_sync, auto_commit
|
||||
|
||||
|
||||
class TestSleepAsync:
|
||||
"""Tests for async sleep function in live mode."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def set_live_mode(self):
|
||||
"""Ensure Config.mode is set to live (not backtest)."""
|
||||
with patch("aiomql.core.utils.Config") as mock_config:
|
||||
mock_config.mode = "live"
|
||||
yield mock_config
|
||||
|
||||
async def test_sleep_calls_asyncio_sleep_in_live_mode(self):
|
||||
"""Test sleep uses asyncio.sleep in live mode."""
|
||||
with patch("aiomql.core.utils.Config") as mock_config:
|
||||
mock_config.mode = "live"
|
||||
with patch("aiomql.core.utils.asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
|
||||
await sleep(1.5)
|
||||
mock_sleep.assert_called_once_with(1.5)
|
||||
|
||||
async def test_sleep_with_zero_seconds(self):
|
||||
"""Test sleep with zero seconds."""
|
||||
with patch("aiomql.core.utils.Config") as mock_config:
|
||||
mock_config.mode = "live"
|
||||
with patch("aiomql.core.utils.asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
|
||||
await sleep(0)
|
||||
mock_sleep.assert_called_once_with(0)
|
||||
|
||||
async def test_sleep_with_integer_seconds(self):
|
||||
"""Test sleep with integer seconds."""
|
||||
with patch("aiomql.core.utils.Config") as mock_config:
|
||||
mock_config.mode = "live"
|
||||
with patch("aiomql.core.utils.asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
|
||||
await sleep(5)
|
||||
mock_sleep.assert_called_once_with(5)
|
||||
|
||||
async def test_sleep_with_float_seconds(self):
|
||||
"""Test sleep with float seconds."""
|
||||
with patch("aiomql.core.utils.Config") as mock_config:
|
||||
mock_config.mode = "live"
|
||||
with patch("aiomql.core.utils.asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
|
||||
await sleep(0.5)
|
||||
mock_sleep.assert_called_once_with(0.5)
|
||||
|
||||
async def test_sleep_actually_delays_execution(self):
|
||||
"""Test sleep actually delays execution in live mode."""
|
||||
with patch("aiomql.core.utils.Config") as mock_config:
|
||||
mock_config.mode = "live"
|
||||
start_time = time.time()
|
||||
await sleep(0.1)
|
||||
elapsed = time.time() - start_time
|
||||
assert elapsed >= 0.1
|
||||
|
||||
|
||||
class TestSleepSync:
|
||||
"""Tests for sync sleep function in live mode."""
|
||||
|
||||
def test_sleep_sync_calls_time_sleep_in_live_mode(self):
|
||||
"""Test sleep_sync uses time.sleep in live mode."""
|
||||
with patch("aiomql.core.utils.Config") as mock_config:
|
||||
mock_config.mode = "live"
|
||||
with patch("aiomql.core.utils.time.sleep") as mock_sleep:
|
||||
sleep_sync(1.5)
|
||||
mock_sleep.assert_called_once_with(1.5)
|
||||
|
||||
def test_sleep_sync_with_zero_seconds(self):
|
||||
"""Test sleep_sync with zero seconds."""
|
||||
with patch("aiomql.core.utils.Config") as mock_config:
|
||||
mock_config.mode = "live"
|
||||
with patch("aiomql.core.utils.time.sleep") as mock_sleep:
|
||||
sleep_sync(0)
|
||||
mock_sleep.assert_called_once_with(0)
|
||||
|
||||
def test_sleep_sync_with_integer_seconds(self):
|
||||
"""Test sleep_sync with integer seconds."""
|
||||
with patch("aiomql.core.utils.Config") as mock_config:
|
||||
mock_config.mode = "live"
|
||||
with patch("aiomql.core.utils.time.sleep") as mock_sleep:
|
||||
sleep_sync(5)
|
||||
mock_sleep.assert_called_once_with(5)
|
||||
|
||||
def test_sleep_sync_with_float_seconds(self):
|
||||
"""Test sleep_sync with float seconds."""
|
||||
with patch("aiomql.core.utils.Config") as mock_config:
|
||||
mock_config.mode = "live"
|
||||
with patch("aiomql.core.utils.time.sleep") as mock_sleep:
|
||||
sleep_sync(0.5)
|
||||
mock_sleep.assert_called_once_with(0.5)
|
||||
|
||||
def test_sleep_sync_actually_delays_execution(self):
|
||||
"""Test sleep_sync actually delays execution in live mode."""
|
||||
with patch("aiomql.core.utils.Config") as mock_config:
|
||||
mock_config.mode = "live"
|
||||
start_time = time.time()
|
||||
sleep_sync(0.1)
|
||||
elapsed = time.time() - start_time
|
||||
assert elapsed >= 0.1
|
||||
|
||||
|
||||
class TestAutoCommit:
|
||||
"""Tests for auto_commit function."""
|
||||
|
||||
async def test_auto_commit_stops_on_shutdown(self):
|
||||
"""Test auto_commit stops when shutdown is True."""
|
||||
mock_config = MagicMock()
|
||||
mock_config.shutdown = True # Start with shutdown True
|
||||
mock_config.db_commit_interval = 0.1
|
||||
mock_state = MagicMock()
|
||||
mock_state.conn.__enter__ = MagicMock(return_value=MagicMock())
|
||||
mock_state.conn.__exit__ = MagicMock(return_value=False)
|
||||
mock_state.acommit = AsyncMock()
|
||||
mock_config.state = mock_state
|
||||
|
||||
with patch("aiomql.core.utils.Config", return_value=mock_config):
|
||||
with patch("aiomql.core.utils.sleep", new_callable=AsyncMock):
|
||||
await auto_commit()
|
||||
|
||||
# Should not have called acommit since shutdown was True immediately
|
||||
mock_state.acommit.assert_not_called()
|
||||
|
||||
async def test_auto_commit_uses_config_interval(self):
|
||||
"""Test auto_commit uses db_commit_interval from config."""
|
||||
call_count = 0
|
||||
|
||||
async def mock_sleep(secs):
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
if call_count >= 2:
|
||||
# Stop the loop after a couple iterations
|
||||
mock_config.shutdown = True
|
||||
|
||||
mock_config = MagicMock()
|
||||
mock_config.shutdown = False
|
||||
mock_config.db_commit_interval = 5.0
|
||||
mock_conn = MagicMock()
|
||||
mock_state = MagicMock()
|
||||
mock_state.conn.__enter__ = MagicMock(return_value=mock_conn)
|
||||
mock_state.conn.__exit__ = MagicMock(return_value=False)
|
||||
mock_state.acommit = AsyncMock()
|
||||
mock_config.state = mock_state
|
||||
|
||||
with patch("aiomql.core.utils.Config", return_value=mock_config):
|
||||
with patch("aiomql.core.utils.sleep", side_effect=mock_sleep) as patched_sleep:
|
||||
await auto_commit()
|
||||
|
||||
# Verify sleep was called with the config interval
|
||||
patched_sleep.assert_called_with(5.0)
|
||||
|
||||
async def test_auto_commit_calls_acommit(self):
|
||||
"""Test auto_commit calls state.acommit."""
|
||||
call_count = 0
|
||||
|
||||
async def mock_sleep(secs):
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
if call_count >= 1:
|
||||
mock_config.shutdown = True
|
||||
|
||||
mock_config = MagicMock()
|
||||
mock_config.shutdown = False
|
||||
mock_config.db_commit_interval = 0.1
|
||||
mock_conn = MagicMock()
|
||||
mock_state = MagicMock()
|
||||
mock_state.conn.__enter__ = MagicMock(return_value=mock_conn)
|
||||
mock_state.conn.__exit__ = MagicMock(return_value=False)
|
||||
mock_state.acommit = AsyncMock()
|
||||
mock_config.state = mock_state
|
||||
|
||||
with patch("aiomql.core.utils.Config", return_value=mock_config):
|
||||
with patch("aiomql.core.utils.sleep", side_effect=mock_sleep):
|
||||
await auto_commit()
|
||||
|
||||
# Verify acommit was called with connection and close=False
|
||||
mock_state.acommit.assert_called_with(conn=mock_conn, close=False)
|
||||
|
||||
async def test_auto_commit_handles_exception(self):
|
||||
"""Test auto_commit handles exceptions gracefully."""
|
||||
mock_config = MagicMock()
|
||||
mock_config.shutdown = False
|
||||
mock_state = MagicMock()
|
||||
mock_state.conn.__enter__ = MagicMock(side_effect=Exception("Test error"))
|
||||
mock_state.conn.__exit__ = MagicMock(return_value=False)
|
||||
mock_config.state = mock_state
|
||||
|
||||
with patch("aiomql.core.utils.Config", return_value=mock_config):
|
||||
with patch("aiomql.core.utils.logger") as mock_logger:
|
||||
# Should not raise, just log the error
|
||||
await auto_commit()
|
||||
mock_logger.error.assert_called_once()
|
||||
|
||||
|
||||
class TestModeDispatch:
|
||||
"""Tests for mode-based dispatch in sleep functions."""
|
||||
|
||||
async def test_sleep_dispatches_to_backtest_in_backtest_mode(self):
|
||||
"""Test sleep calls backtest_sleep in backtest mode."""
|
||||
with patch("aiomql.core.utils.Config") as mock_config:
|
||||
mock_config.mode = "backtest"
|
||||
with patch("aiomql.core.utils.backtest_sleep", new_callable=AsyncMock) as mock_bt_sleep:
|
||||
await sleep(1.0)
|
||||
mock_bt_sleep.assert_called_once_with(1.0)
|
||||
|
||||
def test_sleep_sync_dispatches_to_backtest_in_backtest_mode(self):
|
||||
"""Test sleep_sync calls backtest_sleep_sync in backtest mode."""
|
||||
with patch("aiomql.core.utils.Config") as mock_config:
|
||||
mock_config.mode = "backtest"
|
||||
with patch("aiomql.core.utils.backtest_sleep_sync") as mock_bt_sleep:
|
||||
sleep_sync(1.0)
|
||||
mock_bt_sleep.assert_called_once_with(1.0)
|
||||
Reference in New Issue
Block a user