221 lines
7.7 KiB
Python
221 lines
7.7 KiB
Python
"""Tests for database connectivity and query execution.
|
|
|
|
Uses mocked pyodbc connections — no live database needed.
|
|
"""
|
|
|
|
from unittest.mock import Mock, patch, MagicMock
|
|
import pytest
|
|
|
|
from cteward_ng import database
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_pool():
|
|
"""Ensure the pool is reset between tests."""
|
|
database._pool = None
|
|
yield
|
|
database._pool = None
|
|
|
|
|
|
class TestInit:
|
|
def test_init_raises_without_db(self):
|
|
with pytest.raises(Exception):
|
|
database.init({'server': 'nonexistent', 'password': 'bad'})
|
|
|
|
@patch('cteward_ng.database.PooledDB')
|
|
@patch('cteward_ng.database.logger')
|
|
def test_init_success(self, mock_logger, mock_pooled):
|
|
mock_conn = Mock()
|
|
mock_cursor = Mock()
|
|
mock_conn.cursor.return_value = mock_cursor
|
|
mock_pooled.return_value.connection.return_value = mock_conn
|
|
|
|
database.init({
|
|
'user': 'test',
|
|
'password': 'secret',
|
|
'server': 'localhost',
|
|
'port': 1433,
|
|
'database': 'testdb',
|
|
})
|
|
|
|
assert database._pool is not None
|
|
mock_pooled.assert_called_once()
|
|
|
|
def test_init_with_defaults(self):
|
|
with pytest.raises(Exception):
|
|
database.init(None)
|
|
|
|
|
|
class TestConnected:
|
|
def test_connected_returns_false_when_not_initialized(self):
|
|
assert database.connected() is False
|
|
|
|
@patch.object(database, '_pool', new_callable=lambda: Mock())
|
|
def test_connected_returns_true_on_success(self, mock_pool):
|
|
mock_conn = Mock()
|
|
mock_cursor = Mock()
|
|
mock_pool.connection.return_value = mock_conn
|
|
mock_conn.cursor.return_value = mock_cursor
|
|
|
|
assert database.connected() is True
|
|
|
|
@patch.object(database, '_pool', new_callable=lambda: Mock())
|
|
def test_connected_returns_false_on_error(self, mock_pool):
|
|
mock_pool.connection.side_effect = Exception("DB down")
|
|
assert database.connected() is False
|
|
|
|
|
|
class TestCheckBackendOkay:
|
|
@patch.object(database, '_pool', new_callable=lambda: Mock())
|
|
def test_check_ok_when_enough_members(self, mock_pool):
|
|
mock_conn = Mock()
|
|
mock_cursor = Mock()
|
|
mock_pool.connection.return_value = mock_conn
|
|
mock_conn.cursor.return_value = mock_cursor
|
|
|
|
# First query: member count (fetchone), second: duplicates (fetchall → empty = OK)
|
|
mock_row = Mock()
|
|
mock_row.MemberCount = 10
|
|
mock_cursor.fetchone.return_value = mock_row
|
|
mock_cursor.fetchall.return_value = []
|
|
|
|
database.check_backend_okay()
|
|
|
|
@patch.object(database, '_pool', new_callable=lambda: Mock())
|
|
def test_check_fails_when_too_few_members(self, mock_pool):
|
|
mock_conn = Mock()
|
|
mock_cursor = Mock()
|
|
mock_pool.connection.return_value = mock_conn
|
|
mock_conn.cursor.return_value = mock_cursor
|
|
|
|
mock_row = Mock()
|
|
mock_row.MemberCount = 3
|
|
mock_cursor.fetchone.return_value = mock_row
|
|
|
|
with pytest.raises(RuntimeError, match="Too few members"):
|
|
database.check_backend_okay()
|
|
|
|
@patch.object(database, '_pool', new_callable=lambda: Mock())
|
|
def test_check_fails_on_duplicate_crewnames(self, mock_pool):
|
|
mock_conn = Mock()
|
|
mock_cursor = Mock()
|
|
mock_pool.connection.return_value = mock_conn
|
|
mock_conn.cursor.return_value = mock_cursor
|
|
|
|
mock_row = Mock()
|
|
mock_row.MemberCount = 10
|
|
mock_cursor.fetchone.return_value = mock_row
|
|
mock_cursor.fetchall.return_value = [Mock()]
|
|
|
|
with pytest.raises(RuntimeError, match="Duplicate membernames"):
|
|
database.check_backend_okay()
|
|
|
|
|
|
class TestRunQuery:
|
|
@patch.object(database, '_pool', new_callable=lambda: Mock())
|
|
def test_run_query_no_params(self, mock_pool):
|
|
mock_conn = Mock()
|
|
mock_cursor = Mock()
|
|
mock_pool.connection.return_value = mock_conn
|
|
mock_conn.cursor.return_value = mock_cursor
|
|
mock_cursor.description = [('id',), ('name',)]
|
|
mock_cursor.fetchall.return_value = [(1, 'Alice'), (2, 'Bob')]
|
|
|
|
result = database.run_query(
|
|
{'statement': 'SELECT id, name FROM test', 'params': []},
|
|
{},
|
|
)
|
|
|
|
assert len(result) == 2
|
|
assert result[0] == {'id': 1, 'name': 'Alice'}
|
|
|
|
@patch.object(database, '_pool', new_callable=lambda: Mock())
|
|
def test_run_query_with_params(self, mock_pool):
|
|
mock_conn = Mock()
|
|
mock_cursor = Mock()
|
|
mock_pool.connection.return_value = mock_conn
|
|
mock_conn.cursor.return_value = mock_cursor
|
|
mock_cursor.description = [('Kurzname',)]
|
|
mock_cursor.fetchall.return_value = [('alice',)]
|
|
|
|
result = database.run_query(
|
|
{
|
|
'statement': 'SELECT Kurzname FROM Adresse WHERE Kurzname = ?',
|
|
'params': ['crewname'],
|
|
},
|
|
{'crewname': 'alice'},
|
|
)
|
|
|
|
assert result[0]['Kurzname'] == 'alice'
|
|
mock_cursor.execute.assert_called_once()
|
|
call_args = mock_cursor.execute.call_args[0]
|
|
assert call_args[1] == ['alice']
|
|
|
|
def test_run_query_missing_params(self):
|
|
with pytest.raises(RuntimeError, match="Missing query parameters"):
|
|
database.run_query(
|
|
{
|
|
'statement': 'SELECT * FROM x WHERE a = ?',
|
|
'params': ['a'],
|
|
},
|
|
{},
|
|
)
|
|
|
|
|
|
class TestMemberLookup:
|
|
@patch.object(database, '_exec_query')
|
|
def test_lookup_success(self, mock_exec):
|
|
mock_exec.return_value = [
|
|
{'Kurzname': 'alice', 'Kennung3': 'crew', 'Eintritt': '2020-01-01', 'Austritt': None}
|
|
]
|
|
result = database.member_lookup('alice')
|
|
assert result['Kurzname'] == 'alice'
|
|
mock_exec.assert_called_once()
|
|
|
|
@patch.object(database, '_exec_query')
|
|
def test_lookup_not_found(self, mock_exec):
|
|
mock_exec.return_value = []
|
|
with pytest.raises(RuntimeError, match="expected 1 row, got 0"):
|
|
database.member_lookup('nobody')
|
|
|
|
@patch.object(database, '_exec_query')
|
|
def test_lookup_multiple_rows(self, mock_exec):
|
|
mock_exec.return_value = [{'Kurzname': 'a'}, {'Kurzname': 'b'}]
|
|
with pytest.raises(RuntimeError, match="expected 1 row, got 2"):
|
|
database.member_lookup('dup')
|
|
|
|
|
|
class TestQueryDefinitions:
|
|
"""Verify all query constants have the right structure."""
|
|
|
|
def _check_query_def(self, name, qdef):
|
|
if 'special' in qdef:
|
|
assert isinstance(qdef['special'], str), f"{name} special should be str"
|
|
else:
|
|
assert 'statement' in qdef, f"{name} missing statement"
|
|
assert 'params' in qdef, f"{name} missing params"
|
|
assert isinstance(qdef['statement'], str), f"{name} statement should be str"
|
|
assert isinstance(qdef['params'], list), f"{name} params should be list"
|
|
|
|
def test_all_query_definitions(self):
|
|
queries = [
|
|
'QUERY_CONTRACTLIST_BY_CREWNAME',
|
|
'QUERY_CONTRACT_BY_CREWNAME_AND_CONTRACT',
|
|
'QUERY_DEBITLIST_BY_CREWNAME',
|
|
'QUERY_DEBIT_BY_CREWNAME_AND_GUID',
|
|
'QUERY_MEMBERLIST',
|
|
'QUERY_MEMBERLIST_RAW',
|
|
'QUERY_MEMBER_BY_CREWNAME',
|
|
'QUERY_MEMBER_MEMO_BY_CREWNAME',
|
|
'QUERY_WITHDRAWALLIST_BY_CREWNAME',
|
|
'QUERY_WITHDRAWAL_BY_CREWNAME_AND_GUID',
|
|
'QUERY_PAYMENTLIST_BY_CREWNAME',
|
|
'QUERY_STATS_MEMBERS',
|
|
'QUERY_STATS_CONTRACTS',
|
|
'QUERY_STATS_GENDERS',
|
|
'QUERY_STATS_AGES',
|
|
]
|
|
for name in queries:
|
|
qdef = getattr(database, name, None)
|
|
assert qdef is not None, f"{name} not found"
|
|
self._check_query_def(name, qdef)
|