Skip to content

Commit c70a1e9

Browse files
committed
test: add targeted coverage for low-coverage patch areas
- cover backend_mixin validation branches (deadlock priority range/ whitelist, language whitelist, dateformat whitelist) and parameterized IDENT_CURRENT/CHECKIDENT queries via mocked cursors - cover introspector parameterized SQL builders (table/column/primary key/index/foreign key/view/trigger) offline - cover __main__.parse_args with custom argv - cover transaction manager set_deadlock_priority backend delegation
1 parent cbd6064 commit c70a1e9

4 files changed

Lines changed: 272 additions & 0 deletions

File tree

Lines changed: 103 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,103 @@
1+
# tests/rhosocial/activerecord_sqlserver_test/feature/backend/test_backend_mixin_validation.py
2+
"""Offline tests for SQLServerBackendMixin validation and parameterization.
3+
4+
Covers the whitelist/range validation branches of ``set_deadlock_priority``,
5+
``set_language`` and ``set_dateformat``, plus the parameterized identity
6+
queries, without requiring a live server.
7+
"""
8+
from unittest.mock import MagicMock
9+
10+
import pytest
11+
12+
from rhosocial.activerecord.backend.impl.sqlserver import SQLServerBackend
13+
from rhosocial.activerecord.backend.impl.sqlserver.config import SQLServerConnectionConfig
14+
15+
16+
def make_backend() -> SQLServerBackend:
17+
config = SQLServerConnectionConfig(
18+
host="localhost", port=1433, database="master",
19+
username="sa", password="",
20+
)
21+
backend = SQLServerBackend(connection_config=config)
22+
backend._connection = object()
23+
return backend
24+
25+
26+
def make_cursor(backend):
27+
cursor = MagicMock()
28+
backend._get_cursor = lambda: cursor
29+
return cursor
30+
31+
32+
class TestSetDeadlockPriority:
33+
def test_integer_in_range(self):
34+
backend = make_backend()
35+
cursor = make_cursor(backend)
36+
backend.set_deadlock_priority(-5)
37+
cursor.execute.assert_called_once_with("SET DEADLOCK_PRIORITY -5")
38+
39+
def test_integer_out_of_range_raises(self):
40+
backend = make_backend()
41+
make_cursor(backend)
42+
for value in (-11, 11):
43+
with pytest.raises(ValueError, match="DEADLOCK_PRIORITY"):
44+
backend.set_deadlock_priority(value)
45+
46+
def test_valid_name(self):
47+
backend = make_backend()
48+
cursor = make_cursor(backend)
49+
backend.set_deadlock_priority("high")
50+
cursor.execute.assert_called_once_with("SET DEADLOCK_PRIORITY HIGH")
51+
52+
def test_invalid_name_raises(self):
53+
backend = make_backend()
54+
make_cursor(backend)
55+
with pytest.raises(ValueError, match="DEADLOCK_PRIORITY"):
56+
backend.set_deadlock_priority("EVIL")
57+
58+
59+
class TestSetLanguage:
60+
def test_valid_language_lowercased(self):
61+
backend = make_backend()
62+
cursor = make_cursor(backend)
63+
backend.set_language("US_ENGLISH")
64+
cursor.execute.assert_called_once_with("SET LANGUAGE 'us_english'")
65+
66+
def test_invalid_language_raises(self):
67+
backend = make_backend()
68+
make_cursor(backend)
69+
with pytest.raises(ValueError, match="Unsupported SQL Server language"):
70+
backend.set_language("klingon; DROP TABLE--")
71+
72+
73+
class TestSetDateformat:
74+
def test_valid_format_lowercased(self):
75+
backend = make_backend()
76+
cursor = make_cursor(backend)
77+
backend.set_dateformat("MDY")
78+
cursor.execute.assert_called_once_with("SET DATEFORMAT mdy")
79+
80+
def test_invalid_format_raises(self):
81+
backend = make_backend()
82+
make_cursor(backend)
83+
with pytest.raises(ValueError, match="DATEFORMAT"):
84+
backend.set_dateformat("XYZ")
85+
86+
87+
class TestIdentityQueries:
88+
def test_get_identity_current_parameterized(self):
89+
backend = make_backend()
90+
cursor = make_cursor(backend)
91+
cursor.fetchone.return_value = (42,)
92+
assert backend.get_identity_current("users") == 42
93+
cursor.execute.assert_called_once_with(
94+
"SELECT IDENT_CURRENT(?)", ("users",)
95+
)
96+
97+
def test_reset_identity_parameterized(self):
98+
backend = make_backend()
99+
cursor = make_cursor(backend)
100+
backend.reset_identity("users", 5)
101+
cursor.execute.assert_called_once_with(
102+
"DBCC CHECKIDENT(?, RESEED, ?)", ("users", 5)
103+
)
Lines changed: 112 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,112 @@
1+
# tests/rhosocial/activerecord_sqlserver_test/feature/backend/test_introspection_sql_builders.py
2+
"""Offline tests for the parameterized SQL builder methods in the introspector.
3+
4+
Covers the SQL injection hardening changes: every ``_build_*_sql`` method
5+
now returns parameterized SQL (``?`` placeholders) with separate params,
6+
instead of string-interpolated values.
7+
"""
8+
from rhosocial.activerecord.backend.impl.sqlserver import SQLServerBackend
9+
from rhosocial.activerecord.backend.impl.sqlserver.config import SQLServerConnectionConfig
10+
from rhosocial.activerecord.backend.impl.sqlserver.introspection import (
11+
SyncSQLServerIntrospector,
12+
)
13+
14+
15+
def make_introspector() -> SyncSQLServerIntrospector:
16+
config = SQLServerConnectionConfig(
17+
host="localhost", port=1433, database="master", username="sa", password="",
18+
)
19+
backend = SQLServerBackend(connection_config=config)
20+
backend._version = (16, 0, 0)
21+
introspector = SyncSQLServerIntrospector(backend, executor=object())
22+
return introspector
23+
24+
25+
class TestBuildTableListSql:
26+
def test_default_schema_parameterized(self):
27+
introspector = make_introspector()
28+
sql, params = introspector._build_table_list_sql(None, False)
29+
assert "TABLE_SCHEMA = ?" in sql
30+
assert params == ("dbo",)
31+
32+
def test_with_table_type_parameterized(self):
33+
introspector = make_introspector()
34+
sql, params = introspector._build_table_list_sql("sys", False, table_type="BASE TABLE")
35+
assert "TABLE_TYPE = ?" in sql
36+
assert params == ("sys", "BASE TABLE")
37+
38+
def test_without_views_no_table_type(self):
39+
introspector = make_introspector()
40+
sql, params = introspector._build_table_list_sql("dbo", False, include_views=False)
41+
assert "BASE TABLE" in sql
42+
assert params == ("dbo",)
43+
44+
45+
class TestBuildColumnInfoSql:
46+
def test_parameterized(self):
47+
introspector = make_introspector()
48+
sql, params = introspector._build_column_info_sql("users", "dbo")
49+
assert "c.TABLE_SCHEMA = ?" in sql
50+
assert params == ("dbo", "users")
51+
52+
53+
class TestBuildPrimaryKeySql:
54+
def test_parameterized(self):
55+
introspector = make_introspector()
56+
sql, params = introspector._build_primary_key_sql("users", "dbo")
57+
assert "tc.TABLE_SCHEMA = ?" in sql
58+
assert params == ("dbo", "users")
59+
60+
61+
class TestBuildIndexInfoSql:
62+
def test_parameterized(self):
63+
introspector = make_introspector()
64+
sql, params = introspector._build_index_info_sql("users", "dbo")
65+
assert "s.name = ?" in sql
66+
assert params == ("dbo", "users")
67+
68+
69+
class TestBuildForeignKeySql:
70+
def test_parameterized(self):
71+
introspector = make_introspector()
72+
sql, params = introspector._build_foreign_key_sql("users", "dbo")
73+
assert "s.name = ?" in sql
74+
assert params == ("dbo", "users")
75+
76+
77+
class TestBuildViewListSql:
78+
def test_parameterized(self):
79+
introspector = make_introspector()
80+
sql, params = introspector._build_view_list_sql("dbo", False)
81+
assert "TABLE_SCHEMA = ?" in sql
82+
assert params == ("dbo",)
83+
84+
85+
class TestBuildViewInfoSql:
86+
def test_parameterized(self):
87+
introspector = make_introspector()
88+
sql, params = introspector._build_view_info_sql("active_users", "dbo")
89+
assert "TABLE_SCHEMA = ?" in sql
90+
assert params == ("dbo", "active_users")
91+
92+
93+
class TestBuildTriggerListSql:
94+
def test_with_table_name_parameterized(self):
95+
introspector = make_introspector()
96+
sql, params = introspector._build_trigger_list_sql("orders", "dbo")
97+
assert "AND t.name = ?" in sql
98+
assert params == ("dbo", "orders")
99+
100+
def test_without_table_name(self):
101+
introspector = make_introspector()
102+
sql, params = introspector._build_trigger_list_sql(None, "dbo")
103+
assert "AND t.name = ?" not in sql
104+
assert params == ("dbo",)
105+
106+
107+
class TestBuildDatabaseInfoSql:
108+
def test_returns_sql(self):
109+
introspector = make_introspector()
110+
sql, params = introspector._build_database_info_sql()
111+
assert "DB_NAME()" in sql
112+
assert params == ()
Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
# tests/rhosocial/activerecord_sqlserver_test/feature/backend/test_main_parse_args.py
2+
"""Offline tests for ``__main__.parse_args`` with custom argv."""
3+
import pytest
4+
5+
from rhosocial.activerecord.backend.impl.sqlserver.__main__ import parse_args
6+
7+
8+
class TestParseArgs:
9+
def test_parse_args_with_help_exits(self):
10+
with pytest.raises(SystemExit) as excinfo:
11+
parse_args(["--help"])
12+
assert excinfo.value.code == 0
13+
14+
def test_parse_args_with_query(self):
15+
args = parse_args(["query", "SELECT 1"])
16+
assert args.command == "query"
17+
18+
def test_parse_args_with_info(self):
19+
args = parse_args(["info"])
20+
assert args.command == "info"
Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
# tests/rhosocial/activerecord_sqlserver_test/feature/backend/test_transaction_deadlock_priority.py
2+
"""Offline tests for SQLServerTransactionManager.set_deadlock_priority delegation."""
3+
from unittest.mock import MagicMock
4+
5+
import pytest
6+
7+
from rhosocial.activerecord.backend.transaction import TransactionState
8+
from rhosocial.activerecord.backend.impl.sqlserver.transaction import (
9+
SQLServerTransactionManager,
10+
)
11+
12+
13+
def make_manager() -> tuple[SQLServerTransactionManager, MagicMock]:
14+
backend = MagicMock()
15+
mgr = SQLServerTransactionManager(backend)
16+
return mgr, backend
17+
18+
19+
class TestSetDeadlockPriority:
20+
def test_delegates_to_backend(self):
21+
mgr, backend = make_manager()
22+
mgr._state = TransactionState.ACTIVE
23+
mgr.set_deadlock_priority("LOW")
24+
backend.set_deadlock_priority.assert_called_once_with("LOW")
25+
26+
def test_integer_delegates_to_backend(self):
27+
mgr, backend = make_manager()
28+
mgr._state = TransactionState.ACTIVE
29+
mgr.set_deadlock_priority(-5)
30+
backend.set_deadlock_priority.assert_called_once_with(-5)
31+
32+
def test_raises_without_active_transaction(self):
33+
mgr, backend = make_manager()
34+
mgr._state = TransactionState.INACTIVE
35+
with pytest.raises(Exception, match="No active transaction"):
36+
mgr.set_deadlock_priority("LOW")
37+
backend.set_deadlock_priority.assert_not_called()

0 commit comments

Comments
 (0)