{{ message }}
Repository navigation
Expand file tree
/
Copy pathtest_015_pyformat_parameters.py
More file actions
2304 lines (1937 loc) · 96.6 KB
/
Copy pathtest_015_pyformat_parameters.py
File metadata and controls
2304 lines (1937 loc) · 96.6 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
"""
Copyright (c) Microsoft Corporation.
Licensed under the MIT license.
Comprehensive tests for pyformat parameter style support.
Tests cover parse_pyformat_params(), convert_pyformat_to_qmark(),
and detect_and_convert_parameters() functions.
Goal: 100% code coverage of mssql_python/parameter_helper.py
"""
import pytest
from datetime import date, datetime
from decimal import Decimal
from mssql_python.parameter_helper import (
parse_pyformat_params,
convert_pyformat_to_qmark,
detect_and_convert_parameters,
_has_unquoted_question_marks,
_skip_quoted_context,
)
class TestParsePyformatParams:
"""Test parse_pyformat_params() function."""
def test_parse_single_parameter(self):
"""Test parsing SQL with single parameter."""
sql = "SELECT * FROM users WHERE id = %(id)s"
params = parse_pyformat_params(sql)
assert params == ["id"]
def test_parse_multiple_parameters(self):
"""Test parsing SQL with multiple different parameters."""
sql = "SELECT * FROM users WHERE name = %(name)s AND age = %(age)s AND city = %(city)s"
params = parse_pyformat_params(sql)
assert params == ["name", "age", "city"]
def test_parse_parameter_reuse(self):
"""Test parsing when same parameter appears multiple times."""
sql = "SELECT * FROM users WHERE first_name = %(name)s OR last_name = %(name)s"
params = parse_pyformat_params(sql)
assert params == ["name", "name"]
def test_parse_multiple_reuses(self):
"""Test parsing with multiple parameters reused."""
sql = "WHERE (user_id = %(id)s OR admin_id = %(id)s OR creator_id = %(id)s) AND date > %(date)s"
params = parse_pyformat_params(sql)
assert params == ["id", "id", "id", "date"]
def test_parse_no_parameters(self):
"""Test parsing SQL with no parameters."""
sql = "SELECT * FROM users"
params = parse_pyformat_params(sql)
assert params == []
def test_parse_empty_string(self):
"""Test parsing empty SQL string."""
params = parse_pyformat_params("")
assert params == []
def test_parse_parameter_with_underscores(self):
"""Test parsing parameter names with underscores."""
sql = "WHERE user_id = %(user_id)s AND first_name = %(first_name)s"
params = parse_pyformat_params(sql)
assert params == ["user_id", "first_name"]
def test_parse_parameter_with_numbers(self):
"""Test parsing parameter names with numbers."""
sql = "WHERE col1 = %(param1)s AND col2 = %(param2)s AND col3 = %(param3)s"
params = parse_pyformat_params(sql)
assert params == ["param1", "param2", "param3"]
def test_parse_parameter_in_string_literal(self):
"""Test that parameters in string literals are correctly skipped."""
sql = "SELECT '%(example)s' AS literal, id FROM users WHERE id = %(id)s"
params = parse_pyformat_params(sql)
# Context-aware scanner skips %(example)s inside string literal
assert params == ["id"]
def test_parse_parameter_in_comment(self):
"""Test that parameters in comments are correctly skipped."""
sql = """
SELECT * FROM users
-- This comment has %(commented)s parameter
WHERE id = %(id)s
"""
params = parse_pyformat_params(sql)
# Context-aware scanner skips %(commented)s inside comment
assert params == ["id"]
def test_parse_complex_query_with_cte(self):
"""Test parsing complex CTE query."""
sql = """
WITH recent_orders AS (
SELECT customer_id, SUM(total) as sum_total
FROM orders
WHERE order_date >= %(start_date)s
GROUP BY customer_id
)
SELECT u.name, ro.sum_total
FROM users u
JOIN recent_orders ro ON u.id = ro.customer_id
WHERE ro.sum_total > %(min_amount)s
"""
params = parse_pyformat_params(sql)
assert params == ["start_date", "min_amount"]
def test_parse_incomplete_pattern_no_closing_paren(self):
"""Test that incomplete %(name pattern without ) is ignored."""
sql = "SELECT * FROM users WHERE id = %(id"
params = parse_pyformat_params(sql)
assert params == []
def test_parse_incomplete_pattern_no_s(self):
"""Test that %(name) without 's' is ignored."""
sql = "SELECT * FROM users WHERE id = %(id)"
params = parse_pyformat_params(sql)
assert params == []
def test_parse_percent_without_paren(self):
"""Test that % without ( is ignored."""
sql = "SELECT * FROM users WHERE discount = %10 AND id = %(id)s"
params = parse_pyformat_params(sql)
assert params == ["id"]
def test_parse_special_characters_in_name(self):
"""Test parsing parameter names with special characters (though not recommended)."""
sql = "WHERE x = %(my-param)s"
params = parse_pyformat_params(sql)
assert params == ["my-param"]
def test_parse_empty_parameter_name(self):
"""Test parsing empty parameter name %()s."""
sql = "WHERE x = %()s AND y = %(name)s"
params = parse_pyformat_params(sql)
assert params == ["", "name"]
def test_parse_long_query_many_parameters(self):
"""Test parsing query with many parameters."""
conditions = [f"col{i} = %(param{i})s" for i in range(20)]
sql = "SELECT * FROM table WHERE " + " AND ".join(conditions)
params = parse_pyformat_params(sql)
expected = [f"param{i}" for i in range(20)]
assert params == expected
class TestConvertPyformatToQmark:
"""Test convert_pyformat_to_qmark() function."""
def test_convert_single_parameter(self):
"""Test converting single parameter."""
sql = "SELECT * FROM users WHERE id = %(id)s"
param_dict = {"id": 42}
result_sql, result_params = convert_pyformat_to_qmark(sql, param_dict)
assert result_sql == "SELECT * FROM users WHERE id = ?"
assert result_params == (42,)
def test_convert_multiple_parameters(self):
"""Test converting multiple parameters."""
sql = "INSERT INTO users (name, age, city) VALUES (%(name)s, %(age)s, %(city)s)"
param_dict = {"name": "Alice", "age": 30, "city": "NYC"}
result_sql, result_params = convert_pyformat_to_qmark(sql, param_dict)
assert result_sql == "INSERT INTO users (name, age, city) VALUES (?, ?, ?)"
assert result_params == ("Alice", 30, "NYC")
def test_convert_parameter_reuse(self):
"""Test converting when same parameter is reused."""
sql = "SELECT * FROM logs WHERE user = %(user)s OR admin = %(user)s"
param_dict = {"user": "alice"}
result_sql, result_params = convert_pyformat_to_qmark(sql, param_dict)
assert result_sql == "SELECT * FROM logs WHERE user = ? OR admin = ?"
assert result_params == ("alice", "alice")
def test_convert_parameter_reuse_multiple(self):
"""Test converting with parameter used 3+ times."""
sql = "WHERE a = %(x)s OR b = %(x)s OR c = %(x)s"
param_dict = {"x": 100}
result_sql, result_params = convert_pyformat_to_qmark(sql, param_dict)
assert result_sql == "WHERE a = ? OR b = ? OR c = ?"
assert result_params == (100, 100, 100)
def test_convert_missing_parameter_single(self):
"""Test that missing parameter raises KeyError with helpful message."""
sql = "SELECT * FROM users WHERE id = %(id)s"
param_dict = {"name": "test"}
with pytest.raises(KeyError) as exc_info:
convert_pyformat_to_qmark(sql, param_dict)
error_msg = str(exc_info.value)
assert "'id'" in error_msg
assert "Missing required parameter" in error_msg
def test_convert_missing_parameter_multiple(self):
"""Test that multiple missing parameters are reported."""
sql = "WHERE id = %(id)s AND name = %(name)s AND age = %(age)s"
param_dict = {"id": 42}
with pytest.raises(KeyError) as exc_info:
convert_pyformat_to_qmark(sql, param_dict)
error_msg = str(exc_info.value)
assert "'age'" in error_msg or "'name'" in error_msg
assert "Missing required parameter" in error_msg
def test_convert_extra_parameters_allowed(self):
"""Test that extra parameters in dict are ignored (not an error)."""
sql = "SELECT * FROM users WHERE id = %(id)s"
param_dict = {"id": 42, "name": "Alice", "age": 30}
result_sql, result_params = convert_pyformat_to_qmark(sql, param_dict)
assert result_sql == "SELECT * FROM users WHERE id = ?"
assert result_params == (42,)
def test_convert_empty_dict_no_parameters(self):
"""Test converting query with no parameters and empty dict."""
sql = "SELECT * FROM users"
param_dict = {}
result_sql, result_params = convert_pyformat_to_qmark(sql, param_dict)
assert result_sql == "SELECT * FROM users"
assert result_params == ()
def test_convert_none_value(self):
"""Test converting with NULL/None value."""
sql = "INSERT INTO users (name, phone) VALUES (%(name)s, %(phone)s)"
param_dict = {"name": "John", "phone": None}
result_sql, result_params = convert_pyformat_to_qmark(sql, param_dict)
assert result_sql == "INSERT INTO users (name, phone) VALUES (?, ?)"
assert result_params == ("John", None)
def test_convert_various_types(self):
"""Test converting with various Python data types."""
sql = """
INSERT INTO data (str_col, int_col, float_col, bool_col, date_col, bytes_col, decimal_col)
VALUES (%(s)s, %(i)s, %(f)s, %(b)s, %(d)s, %(by)s, %(dec)s)
"""
param_dict = {
"s": "text",
"i": 42,
"f": 3.14,
"b": True,
"d": date(2025, 1, 1),
"by": b"\x00\x01\x02",
"dec": Decimal("99.99"),
}
result_sql, result_params = convert_pyformat_to_qmark(sql, param_dict)
assert "?" in result_sql
assert "%(s)s" not in result_sql
assert len(result_params) == 7
assert result_params[0] == "text"
assert result_params[1] == 42
assert result_params[2] == 3.14
assert result_params[3] is True
assert result_params[4] == date(2025, 1, 1)
assert result_params[5] == b"\x00\x01\x02"
assert result_params[6] == Decimal("99.99")
def test_convert_unicode_values(self):
"""Test converting with Unicode characters in values."""
sql = "INSERT INTO users (name) VALUES (%(name)s)"
param_dict = {"name": "José María 日本語 🎉"}
result_sql, result_params = convert_pyformat_to_qmark(sql, param_dict)
assert result_sql == "INSERT INTO users (name) VALUES (?)"
assert result_params == ("José María 日本語 🎉",)
def test_convert_sql_injection_attempt(self):
"""Test that SQL injection attempts are safely handled as parameter values."""
sql = "SELECT * FROM users WHERE name = %(name)s"
param_dict = {"name": "'; DROP TABLE users; --"}
result_sql, result_params = convert_pyformat_to_qmark(sql, param_dict)
assert result_sql == "SELECT * FROM users WHERE name = ?"
assert result_params == ("'; DROP TABLE users; --",)
def test_convert_complex_cte_query(self):
"""Test converting complex CTE query."""
sql = """
WITH recent_orders AS (
SELECT customer_id, SUM(total) as sum_total
FROM orders
WHERE order_date >= %(start_date)s
GROUP BY customer_id
)
SELECT u.name, ro.sum_total
FROM users u
JOIN recent_orders ro ON u.id = ro.customer_id
WHERE ro.sum_total > %(min_amount)s
"""
param_dict = {"start_date": "2025-01-01", "min_amount": 1000.00}
result_sql, result_params = convert_pyformat_to_qmark(sql, param_dict)
assert "%(start_date)s" not in result_sql
assert "%(min_amount)s" not in result_sql
assert result_sql.count("?") == 2
assert result_params == ("2025-01-01", 1000.00)
def test_convert_with_escaped_percent(self):
"""Test that %% is converted to single %."""
sql = "SELECT * FROM users WHERE discount = '%%10' AND id = %(id)s"
param_dict = {"id": 42}
result_sql, result_params = convert_pyformat_to_qmark(sql, param_dict)
assert result_sql == "SELECT * FROM users WHERE discount = '%10' AND id = ?"
assert result_params == (42,)
def test_convert_with_multiple_escaped_percent(self):
"""Test multiple %% escapes with context-aware parsing.
The %(name)s inside the string literal '%%%(name)s%%' is correctly
skipped by context-aware scanning, so only %(id)s is treated as a
real parameter. After %% unescaping, the string literal becomes
'%%(name)s%' (literal percent + literal %(name)s + literal percent).
"""
sql = (
"SELECT '%%test%%' AS txt, id FROM users WHERE id = %(id)s AND name LIKE '%%%(name)s%%'"
)
param_dict = {"id": 1, "name": "alice"}
result_sql, result_params = convert_pyformat_to_qmark(sql, param_dict)
assert "'%test%'" in result_sql
assert "?" in result_sql
# %(name)s inside string literal is preserved as literal text after %% unescaping
assert "'%%(name)s%'" in result_sql
assert result_params == (1,)
def test_convert_only_escaped_percent_no_params(self):
"""Test SQL with only %% and no parameters."""
sql = "SELECT * FROM users WHERE discount = '%%10'"
param_dict = {}
result_sql, result_params = convert_pyformat_to_qmark(sql, param_dict)
assert result_sql == "SELECT * FROM users WHERE discount = '%10'"
assert result_params == ()
def test_convert_empty_parameter_name(self):
"""Test converting with empty parameter name (edge case)."""
sql = "WHERE x = %()s"
param_dict = {"": "value"}
result_sql, result_params = convert_pyformat_to_qmark(sql, param_dict)
assert result_sql == "WHERE x = ?"
assert result_params == ("value",)
def test_convert_many_parameters(self):
"""Test converting with many parameters (performance test)."""
sql = "SELECT * FROM table WHERE " + " AND ".join(
[f"col{i} = %(param{i})s" for i in range(50)]
)
param_dict = {f"param{i}": i for i in range(50)}
result_sql, result_params = convert_pyformat_to_qmark(sql, param_dict)
assert result_sql.count("?") == 50
assert len(result_params) == 50
assert result_params == tuple(range(50))
class TestSkipQuotedContext:
"""Test _skip_quoted_context() helper function."""
def test_no_quoted_context(self):
"""Test returns -1 when position is not a quoted context."""
sql = "SELECT * FROM t"
assert _skip_quoted_context(sql, 0, len(sql)) == -1
assert _skip_quoted_context(sql, 7, len(sql)) == -1
def test_skip_single_line_comment(self):
"""Test skipping -- comment to end of line."""
sql = "SELECT 1 -- comment\nFROM t"
result = _skip_quoted_context(sql, 9, len(sql))
assert result == 19 # position of \n
def test_skip_multi_line_comment(self):
"""Test skipping /* ... */ block comment."""
sql = "SELECT /* block */ 1"
result = _skip_quoted_context(sql, 7, len(sql))
assert result == 18 # position after */
def test_skip_single_quoted_string(self):
"""Test skipping '...' string literal."""
sql = "SELECT 'hello' FROM t"
result = _skip_quoted_context(sql, 7, len(sql))
assert result == 14 # position after closing '
def test_skip_single_quoted_with_escaped_quote(self):
"""Test skipping string with '' escaped quote."""
sql = "SELECT 'it''s ok' FROM t"
result = _skip_quoted_context(sql, 7, len(sql))
assert result == 17 # position after closing '
def test_skip_double_quoted_identifier(self):
"""Test skipping \"...\" identifier."""
sql = 'SELECT "col name" FROM t'
result = _skip_quoted_context(sql, 7, len(sql))
assert result == 17 # position after closing "
def test_skip_bracketed_identifier(self):
"""Test skipping [...] identifier."""
sql = "SELECT [col name] FROM t"
result = _skip_quoted_context(sql, 7, len(sql))
assert result == 17 # position after closing ]
def test_skip_bracketed_with_question_mark(self):
"""Test skipping [col?name] — the core bug scenario."""
sql = "SELECT [q?marks] FROM t"
result = _skip_quoted_context(sql, 7, len(sql))
assert result == 16 # position after closing ]
def test_skip_bracketed_with_escaped_bracket(self):
"""Test skipping [a]]?b] — ]] is an escaped ] inside bracket identifier."""
sql = "SELECT [a]]?b] FROM t"
result = _skip_quoted_context(sql, 7, len(sql))
assert result == 14 # position after the real closing ]
def test_skip_unterminated_block_comment(self):
"""Test unterminated block comment returns length, not beyond it."""
sql = "SELECT /* unclosed"
result = _skip_quoted_context(sql, 7, len(sql))
assert result == len(sql) # should not exceed length
def test_skip_nested_block_comment(self):
"""Test skipping /* outer /* inner */ still comment */ with nesting."""
sql = "SELECT /* outer /* inner */ still comment */ 1"
result = _skip_quoted_context(sql, 7, len(sql))
assert result == 44 # position after the outer */
def test_skip_deeply_nested_block_comment(self):
"""Test skipping three levels of nested block comments."""
sql = "SELECT /* a /* b /* c */ b */ a */ 1"
result = _skip_quoted_context(sql, 7, len(sql))
assert result == 34 # position after the outermost */
def test_skip_nested_block_comment_unterminated(self):
"""Test unterminated nested block comment consumes to end."""
sql = "SELECT /* outer /* inner */ still open"
result = _skip_quoted_context(sql, 7, len(sql))
assert result == len(sql) # depth never reaches 0
def test_skip_unterminated_single_quote_does_not_exceed_length(self):
"""Test unterminated single-quoted string returns length, not length+1."""
sql = "SELECT 'unterminated"
result = _skip_quoted_context(sql, 7, len(sql))
assert result == len(sql) # must not exceed length
def test_skip_unterminated_double_quote_does_not_exceed_length(self):
"""Test unterminated double-quoted identifier returns length, not length+1."""
sql = 'SELECT "unterminated'
result = _skip_quoted_context(sql, 7, len(sql))
assert result == len(sql) # must not exceed length
def test_skip_double_quoted_with_escaped_quote(self):
"""Test skipping double-quoted identifier with escaped \"\" inside."""
# "col""?" is the identifier col"? — the ? is inside the identifier
sql = 'SELECT "col""?" FROM t'
result = _skip_quoted_context(sql, 7, len(sql))
assert result == 15 # position after the real closing "
def test_skip_double_quoted_with_escaped_quote_no_qmark(self):
"""Test skipping double-quoted identifier with escaped \"\" but no ?."""
sql = 'SELECT "col""name" FROM t'
result = _skip_quoted_context(sql, 7, len(sql))
assert result == 18 # position after the real closing "
def test_skip_double_quoted_with_escaped_quote_containing_qmark(self):
"""Test skipping \"col\"\"?name\" — the ? is inside the identifier col\"?name."""
sql = 'SELECT "col""?name" FROM t'
result = _skip_quoted_context(sql, 7, len(sql))
# "col""?name" spans positions 7..18, return position after closing "
assert result == 19
def test_skip_double_quoted_just_escaped_quote(self):
"""Test skipping \"\"\"\" — identifier content is \"\" representing a single \" char."""
sql = 'SELECT """" FROM t'
result = _skip_quoted_context(sql, 7, len(sql))
# """" spans positions 7..10, return position after closing " at 10
assert result == 11
def test_skip_empty_single_quoted_string(self):
"""Test skipping '' — empty string literal."""
sql = "SELECT '' FROM t"
result = _skip_quoted_context(sql, 7, len(sql))
# '' spans positions 7..8, return position after closing ' at 8
assert result == 9
def test_skip_empty_double_quoted_identifier(self):
"""Test skipping \"\" — empty double-quoted identifier."""
sql = 'SELECT "" FROM t'
result = _skip_quoted_context(sql, 7, len(sql))
# "" spans positions 7..8, return position after closing " at 8
assert result == 9
def test_skip_empty_bracketed_identifier(self):
"""Test skipping [] — empty bracketed identifier."""
sql = "SELECT [] FROM t"
result = _skip_quoted_context(sql, 7, len(sql))
# [] spans positions 7..8, return position after closing ] at 8
assert result == 9
class TestHasUnquotedQuestionMarks:
"""Test _has_unquoted_question_marks() helper function."""
# --- Cases that SHOULD return True (real qmark placeholders) ---
def test_simple_qmark_placeholder(self):
"""Test bare ? is detected as a real placeholder."""
assert _has_unquoted_question_marks("SELECT * FROM t WHERE id = ?") is True
def test_multiple_qmark_placeholders(self):
"""Test multiple ? placeholders."""
assert _has_unquoted_question_marks("WHERE a = ? AND b = ?") is True
def test_qmark_at_start(self):
"""Test ? at very start of SQL."""
assert _has_unquoted_question_marks("?") is True
def test_qmark_at_end(self):
"""Test ? at very end of SQL."""
assert _has_unquoted_question_marks("SELECT ?") is True
def test_qmark_after_bracket_identifier(self):
"""Test real ? placeholder after a bracketed identifier containing ?."""
assert _has_unquoted_question_marks("SELECT [q?marks] FROM t WHERE id = ?") is True
def test_qmark_after_string_literal(self):
"""Test real ? placeholder after a string literal containing ?."""
assert _has_unquoted_question_marks("SELECT 'what?' AS q WHERE id = ?") is True
# --- Cases that SHOULD return False (? only inside quoted contexts) ---
def test_qmark_inside_brackets(self):
"""Test ? inside bracketed identifier is NOT detected."""
assert _has_unquoted_question_marks("SELECT [q?marks] FROM t") is False
def test_qmark_inside_single_quotes(self):
"""Test ? inside single-quoted string literal is NOT detected."""
assert _has_unquoted_question_marks("SELECT 'is this ok?' AS col") is False
def test_qmark_inside_double_quotes(self):
"""Test ? inside double-quoted identifier is NOT detected."""
assert _has_unquoted_question_marks('SELECT "col?name" FROM t') is False
def test_qmark_inside_single_line_comment(self):
"""Test ? inside single-line comment is NOT detected."""
assert _has_unquoted_question_marks("SELECT 1 -- is this ok?") is False
def test_qmark_inside_multi_line_comment(self):
"""Test ? inside multi-line comment is NOT detected."""
assert _has_unquoted_question_marks("SELECT /* what? */ 1") is False
def test_no_question_marks_at_all(self):
"""Test SQL with no ? at all."""
assert _has_unquoted_question_marks("SELECT * FROM users") is False
def test_empty_string(self):
"""Test empty SQL string."""
assert _has_unquoted_question_marks("") is False
def test_multiple_bracketed_identifiers_with_qmark(self):
"""Test multiple bracketed identifiers each containing ?."""
assert _has_unquoted_question_marks("SELECT [col?1], [col?2] FROM [tbl?x]") is False
def test_qmark_in_string_with_escaped_quote(self):
"""Test ? in string literal that contains escaped single quotes."""
assert _has_unquoted_question_marks("SELECT 'it''s ok?' FROM t") is False
def test_qmark_in_multiline_comment_spanning_lines(self):
"""Test ? inside multi-line comment spanning multiple lines."""
sql = """SELECT 1
/* this is a
multi-line comment with ?
in it */
FROM t"""
assert _has_unquoted_question_marks(sql) is False
def test_mixed_contexts_no_real_qmark(self):
"""Test ? appearing in multiple quoted contexts but no real placeholder."""
sql = """SELECT [q?col], 'val?ue' AS x -- comment?
FROM t /* block? */"""
assert _has_unquoted_question_marks(sql) is False
def test_bracket_then_real_qmark(self):
"""Test bracketed ? followed by a real ? placeholder."""
assert _has_unquoted_question_marks("SELECT [q?marks] FROM t WHERE x = ?") is True
def test_string_then_comment_then_real_qmark(self):
"""Test ? in string, ? in comment, then a real ? placeholder."""
sql = "SELECT 'ok?' AS q -- comment?\nWHERE id = ?"
assert _has_unquoted_question_marks(sql) is True
def test_unclosed_bracket_treats_rest_as_identifier(self):
"""Test behavior with unclosed bracket - ? after [ is inside bracket context."""
# Unclosed bracket: scanner will consume to end looking for ]
assert _has_unquoted_question_marks("SELECT [unclosed?col") is False
def test_unclosed_single_quote_treats_rest_as_string(self):
"""Test behavior with unclosed quote - ? after ' is inside string context."""
assert _has_unquoted_question_marks("SELECT 'unclosed string?") is False
def test_escaped_bracket_inside_bracketed_identifier(self):
"""Test ? after ]] escaped bracket inside [...] is NOT detected."""
# SQL Server escapes ] as ]] inside bracketed identifiers
# [a]]?b] is the identifier a]?b — the ? is inside the brackets
assert _has_unquoted_question_marks("SELECT [a]]?b] FROM t") is False
def test_unclosed_block_comment_treats_rest_as_comment(self):
"""Test behavior with unclosed block comment - ? after /* is inside comment context."""
assert _has_unquoted_question_marks("SELECT /* unclosed?") is False
def test_qmark_inside_nested_block_comment(self):
"""Test ? inside nested block comment is NOT detected (SQL Server supports nesting)."""
assert (
_has_unquoted_question_marks("SELECT /* outer /* inner */ still comment ? */ 1")
is False
)
def test_qmark_after_nested_block_comment(self):
"""Test ? after a properly closed nested block comment IS detected."""
assert (
_has_unquoted_question_marks("SELECT /* outer /* inner */ outer */ FROM t WHERE id = ?")
is True
)
def test_qmark_inside_deeply_nested_block_comment(self):
"""Test ? inside three levels of nesting is NOT detected."""
assert _has_unquoted_question_marks("SELECT /* a /* b /* c ? */ b */ a */ 1") is False
def test_qmark_inside_double_quoted_with_escaped_quotes(self):
"""Test ? inside double-quoted identifier with escaped \"\" is NOT detected."""
assert _has_unquoted_question_marks('SELECT "col""?" FROM t') is False
def test_real_qmark_after_double_quoted_escaped(self):
"""Test real ? after double-quoted identifier with escaped \"\" IS detected."""
assert _has_unquoted_question_marks('SELECT "col""name" FROM t WHERE id = ?') is True
def test_qmark_inside_unterminated_double_quote(self):
"""Test ? inside unterminated double-quoted identifier is NOT detected."""
assert _has_unquoted_question_marks('SELECT "unterminated?') is False
def test_qmark_inside_double_quoted_escaped_with_name(self):
"""Test ? inside \"col\"\"?name\" is NOT detected — ? is within the identifier."""
assert _has_unquoted_question_marks('SELECT "col""?name" FROM t') is False
def test_real_qmark_after_double_quoted_escaped_with_qmark_inside(self):
"""Test real ? after \"col\"\"?name\" IS detected."""
assert _has_unquoted_question_marks('SELECT "col""?name" FROM t WHERE id = ?') is True
def test_multiple_double_quote_escapes_no_qmark(self):
"""Test multiple \"\" escapes with no ? anywhere."""
assert _has_unquoted_question_marks('SELECT "a""b""c" FROM t') is False
def test_qmark_sandwiched_between_double_quote_escapes(self):
"""Test ? sandwiched between \"\" escapes inside identifier \"a\"\"?\"\"b\"."""
assert _has_unquoted_question_marks('SELECT "a""?""b" FROM t') is False
def test_qmark_immediately_after_closing_bracket(self):
"""Test ? immediately after closing ] is a real unquoted placeholder."""
assert _has_unquoted_question_marks("SELECT [col]? FROM t") is True
def test_qmark_immediately_after_closing_single_quote(self):
"""Test ? immediately after closing ' is a real unquoted placeholder."""
assert _has_unquoted_question_marks("SELECT 'str'? FROM t") is True
def test_only_whitespace_no_qmark(self):
"""Test SQL with only whitespace characters — no ? at all."""
assert _has_unquoted_question_marks(" \n\t ") is False
def test_qmark_between_two_block_comments(self):
"""Test ? between two properly closed block comments is detected."""
assert _has_unquoted_question_marks("SELECT /* comment */ ? /* comment */") is True
class TestParsePyformatParamsContextAware:
"""Test that parse_pyformat_params skips quoted contexts."""
def test_pyformat_in_string_literal_is_skipped(self):
"""Test %(name)s inside single-quoted string is not detected."""
sql = "SELECT '%(fake)s' FROM t WHERE id = %(id)s"
params = parse_pyformat_params(sql)
assert params == ["id"]
def test_pyformat_in_single_line_comment_is_skipped(self):
"""Test %(name)s inside -- comment is not detected."""
sql = "SELECT 1 -- %(fake)s\nWHERE id = %(id)s"
params = parse_pyformat_params(sql)
assert params == ["id"]
def test_pyformat_in_block_comment_is_skipped(self):
"""Test %(name)s inside /* ... */ comment is not detected."""
sql = "SELECT /* %(fake)s */ 1 WHERE id = %(id)s"
params = parse_pyformat_params(sql)
assert params == ["id"]
def test_pyformat_in_double_quoted_identifier_is_skipped(self):
"""Test %(name)s inside double-quoted identifier is not detected."""
sql = 'SELECT "%(fake)s" FROM t WHERE id = %(id)s'
params = parse_pyformat_params(sql)
assert params == ["id"]
def test_pyformat_in_bracketed_identifier_is_skipped(self):
"""Test %(name)s inside bracketed identifier is not detected."""
sql = "SELECT [%(fake)s] FROM t WHERE id = %(id)s"
params = parse_pyformat_params(sql)
assert params == ["id"]
def test_fake_pyformat_in_string_with_real_qmark(self):
"""Test that fake %(name)s in string doesn't prevent qmark mismatch detection.
Reproduces: SELECT 'fake %(name)s' FROM t WHERE id = ? with {"id": 1}
The parse_pyformat_params should NOT find %(name)s inside the string,
so the mismatch check correctly raises TypeError for the real ? placeholder.
"""
sql = "SELECT 'fake %(name)s' FROM t WHERE id = ?"
# parse_pyformat_params should return empty (%(name)s is in string literal)
assert parse_pyformat_params(sql) == []
# Therefore detect_and_convert_parameters should raise TypeError for dict + ?
with pytest.raises(TypeError) as exc_info:
detect_and_convert_parameters(sql, {"id": 1})
assert "positional placeholders" in str(exc_info.value)
class TestBracketedIdentifierIntegration:
"""Integration tests for ? inside bracketed identifiers with detect_and_convert_parameters.
Reproduces the GitHub issue: question mark in bracketed column name
causes false positive 'Parameter style mismatch' error.
"""
def test_bracketed_qmark_with_pyformat_params(self):
"""Test ? inside brackets with correct pyformat params (the fix for the reported bug)."""
sql = "SELECT * FROM t WHERE [q?marks]=%(somename)s"
params = {"somename": "thename"}
result_sql, result_params = detect_and_convert_parameters(sql, params)
assert "[q?marks]" in result_sql
assert result_params == ("thename",)
assert "%(somename)s" not in result_sql
def test_bracketed_qmark_with_empty_dict(self):
"""Test ? inside brackets with empty dict should NOT raise TypeError."""
sql = "SELECT [q?marks] FROM t WHERE id = 1"
result_sql, result_params = detect_and_convert_parameters(sql, {})
assert result_sql == sql
assert result_params == ()
def test_bracketed_qmark_with_no_params(self):
"""Test ? inside brackets with None params."""
sql = "SELECT [q?marks] FROM t"
result_sql, result_params = detect_and_convert_parameters(sql, None)
assert result_sql == sql
assert result_params is None
def test_qmark_in_string_literal_with_dict_params(self):
"""Test ? inside string literal with dict params should NOT raise TypeError."""
sql = "SELECT 'is this ok?' AS col WHERE id = %(id)s"
params = {"id": 42}
result_sql, result_params = detect_and_convert_parameters(sql, params)
assert result_params == (42,)
def test_qmark_in_comment_with_dict_params(self):
"""Test ? inside comment with dict params should NOT raise TypeError."""
sql = "SELECT * FROM t -- filter by what?\nWHERE id = %(id)s"
params = {"id": 1}
result_sql, result_params = detect_and_convert_parameters(sql, params)
assert result_params == (1,)
def test_real_qmark_with_dict_still_raises(self):
"""Test that a genuine ? placeholder with dict params still raises TypeError."""
sql = "SELECT * FROM t WHERE id = ?"
with pytest.raises(TypeError) as exc_info:
detect_and_convert_parameters(sql, {"id": 1})
assert "positional placeholders" in str(exc_info.value)
def test_real_qmark_alongside_bracketed_qmark_with_dict_raises(self):
"""Test real ? placeholder alongside bracketed ? with dict still raises."""
sql = "SELECT [q?marks] FROM t WHERE id = ?"
with pytest.raises(TypeError) as exc_info:
detect_and_convert_parameters(sql, {"id": 1})
assert "positional placeholders" in str(exc_info.value)
def test_bracketed_qmark_with_tuple_params(self):
"""Test ? inside brackets with tuple params passes through fine."""
sql = "SELECT * FROM t WHERE [q?marks] = ?"
params = ("thename",)
result_sql, result_params = detect_and_convert_parameters(sql, params)
assert result_sql == sql
assert result_params == params
def test_multiple_bracketed_columns_with_qmarks(self):
"""Test multiple bracketed columns each containing ? with pyformat params."""
sql = "SELECT [col?1], [col?2] FROM t WHERE name = %(name)s"
params = {"name": "test"}
result_sql, result_params = detect_and_convert_parameters(sql, params)
assert "[col?1]" in result_sql
assert "[col?2]" in result_sql
assert result_params == ("test",)
def test_create_table_with_bracketed_qmark_column(self):
"""Test CREATE TABLE with ? in column name (no params)."""
sql = "CREATE TABLE t (id INT, [q?marks] VARCHAR(50))"
result_sql, result_params = detect_and_convert_parameters(sql, None)
assert result_sql == sql
assert result_params is None
def test_double_quoted_escaped_qmark_with_pyformat_params(self):
"""Test ? inside \"col\"\"?\" with pyformat params should NOT raise TypeError."""
sql = 'SELECT "col""?" FROM t WHERE id = %(id)s'
params = {"id": 1}
result_sql, result_params = detect_and_convert_parameters(sql, params)
assert result_params == (1,)
assert "%(id)s" not in result_sql
def test_double_quoted_escaped_qmark_with_real_qmark_and_dict_raises(self):
"""Test ? inside \"col\"\"?\" with real ? and dict raises TypeError."""
sql = 'SELECT "col""?" FROM t WHERE id = ?'
with pytest.raises(TypeError) as exc_info:
detect_and_convert_parameters(sql, {"id": 1})
assert "positional placeholders" in str(exc_info.value)
def test_qmark_in_block_comment_with_pyformat_params(self):
"""Test ? inside block comment with pyformat params should NOT raise TypeError."""
sql = "SELECT /* hint? */ * FROM t WHERE id = %(id)s"
params = {"id": 1}
result_sql, result_params = detect_and_convert_parameters(sql, params)
assert result_params == (1,)
assert "%(id)s" not in result_sql
def test_qmark_in_string_literal_alone_with_pyformat_params(self):
"""Test '?' in string literal with pyformat params should NOT raise TypeError."""
sql = "SELECT '?' FROM t WHERE id = %(id)s"
params = {"id": 42}
result_sql, result_params = detect_and_convert_parameters(sql, params)
assert result_params == (42,)
assert "%(id)s" not in result_sql
def test_qmark_in_single_line_comment_with_pyformat_params(self):
"""Test ? inside single-line comment with pyformat params should NOT raise TypeError."""
sql = "SELECT -- comment?\nFROM t WHERE id = %(id)s"
params = {"id": 5}
result_sql, result_params = detect_and_convert_parameters(sql, params)
assert result_params == (5,)
assert "%(id)s" not in result_sql
class TestDetectAndConvertParameters:
"""Test detect_and_convert_parameters() function."""
def test_detect_none_parameters(self):
"""Test detection when parameters is None."""
sql = "SELECT * FROM users"
result_sql, result_params = detect_and_convert_parameters(sql, None)
assert result_sql == sql
assert result_params is None
def test_detect_qmark_tuple(self):
"""Test detection of qmark style with tuple."""
sql = "SELECT * FROM users WHERE id = ?"
params = (42,)
result_sql, result_params = detect_and_convert_parameters(sql, params)
assert result_sql == sql
assert result_params == params
def test_detect_qmark_list(self):
"""Test detection of qmark style with list."""
sql = "SELECT * FROM users WHERE id = ? AND name = ?"
params = [42, "Alice"]
result_sql, result_params = detect_and_convert_parameters(sql, params)
assert result_sql == sql
assert result_params == params
def test_detect_pyformat_dict(self):
"""Test detection of pyformat style with dict."""
sql = "SELECT * FROM users WHERE id = %(id)s"
params = {"id": 42}
result_sql, result_params = detect_and_convert_parameters(sql, params)
assert result_sql == "SELECT * FROM users WHERE id = ?"
assert result_params == (42,)
def test_detect_pyformat_multiple_params(self):
"""Test detection and conversion with multiple pyformat params."""
sql = "INSERT INTO users (name, age) VALUES (%(name)s, %(age)s)"
params = {"name": "Bob", "age": 25}
result_sql, result_params = detect_and_convert_parameters(sql, params)
assert result_sql == "INSERT INTO users (name, age) VALUES (?, ?)"
assert result_params == ("Bob", 25)
def test_detect_type_mismatch_dict_with_qmark(self):
"""Test TypeError when dict is used with ? placeholders."""
sql = "SELECT * FROM users WHERE id = ?"
params = {"id": 42}
with pytest.raises(TypeError) as exc_info:
detect_and_convert_parameters(sql, params)
error_msg = str(exc_info.value)
assert "Parameter style mismatch" in error_msg
assert "positional placeholders (?)" in error_msg
assert "dict was provided" in error_msg
def test_detect_type_mismatch_tuple_with_pyformat(self):
"""Test TypeError when tuple is used with %(name)s placeholders."""
sql = "SELECT * FROM users WHERE id = %(id)s"
params = (42,)
with pytest.raises(TypeError) as exc_info:
detect_and_convert_parameters(sql, params)
error_msg = str(exc_info.value)
assert "Parameter style mismatch" in error_msg
assert "named placeholders" in error_msg
assert "tuple was provided" in error_msg
def test_detect_type_mismatch_list_with_pyformat(self):
"""Test TypeError when list is used with %(name)s placeholders."""
sql = "SELECT * FROM users WHERE id = %(id)s AND name = %(name)s"
params = [42, "Alice"]
with pytest.raises(TypeError) as exc_info:
detect_and_convert_parameters(sql, params)
error_msg = str(exc_info.value)
assert "Parameter style mismatch" in error_msg
assert "list was provided" in error_msg
def test_detect_invalid_type_string(self):
"""Test TypeError for unsupported parameter type (string)."""
sql = "SELECT * FROM users WHERE id = ?"
params = "42"
with pytest.raises(TypeError) as exc_info:
detect_and_convert_parameters(sql, params)
error_msg = str(exc_info.value)
assert "Parameters must be tuple, list, dict, or None" in error_msg
assert "str" in error_msg
def test_detect_invalid_type_int(self):
"""Test TypeError for unsupported parameter type (int)."""
sql = "SELECT * FROM users WHERE id = ?"
params = 42
with pytest.raises(TypeError) as exc_info:
detect_and_convert_parameters(sql, params)
error_msg = str(exc_info.value)
assert "Parameters must be tuple, list, dict, or None" in error_msg
assert "int" in error_msg
def test_detect_invalid_type_set(self):
"""Test TypeError for unsupported parameter type (set)."""
sql = "SELECT * FROM users WHERE id = ?"
params = {42, 43}
with pytest.raises(TypeError) as exc_info:
detect_and_convert_parameters(sql, params)
error_msg = str(exc_info.value)
assert "Parameters must be tuple, list, dict, or None" in error_msg
assert "set" in error_msg
def test_detect_qmark_with_no_question_marks(self):
"""Test qmark detection when SQL has no ? but tuple provided."""
sql = "SELECT * FROM users"
params = (42, "Alice")
result_sql, result_params = detect_and_convert_parameters(sql, params)
# Passes through - SQL execution will handle parameter count mismatch
assert result_sql == sql
assert result_params == params
def test_detect_pyformat_with_parameter_reuse(self):
"""Test detection and conversion with parameter reuse."""
sql = "WHERE user = %(user)s OR admin = %(user)s"
params = {"user": "alice"}
result_sql, result_params = detect_and_convert_parameters(sql, params)
assert result_sql == "WHERE user = ? OR admin = ?"
assert result_params == ("alice", "alice")
def test_detect_empty_tuple(self):
"""Test detection with empty tuple (no parameters)."""
sql = "SELECT * FROM users"
params = ()
result_sql, result_params = detect_and_convert_parameters(sql, params)
assert result_sql == sql
assert result_params == ()
def test_detect_empty_list(self):
"""Test detection with empty list (no parameters)."""
sql = "SELECT * FROM users"
params = []
result_sql, result_params = detect_and_convert_parameters(sql, params)
assert result_sql == sql
assert result_params == []
def test_detect_empty_dict(self):
"""Test detection with empty dict (no parameters)."""
sql = "SELECT * FROM users"
params = {}
result_sql, result_params = detect_and_convert_parameters(sql, params)
assert result_sql == sql
assert result_params == ()
def test_detect_pyformat_missing_parameter(self):
"""Test that missing pyformat parameter raises KeyError."""
sql = "WHERE id = %(id)s AND name = %(name)s"
params = {"id": 42}
with pytest.raises(KeyError) as exc_info:
detect_and_convert_parameters(sql, params)
error_msg = str(exc_info.value)
assert "Missing required parameter" in error_msg
assert "'name'" in error_msg
def test_detect_complex_query_pyformat(self):
"""Test detection and conversion with complex query."""
sql = """
WITH recent AS (
SELECT id FROM orders WHERE date >= %(date)s
)
SELECT * FROM users u
JOIN recent r ON u.id = r.id
WHERE u.status = %(status)s
"""
params = {"date": "2025-01-01", "status": "active"}
result_sql, result_params = detect_and_convert_parameters(sql, params)
assert "%(date)s" not in result_sql
assert "%(status)s" not in result_sql
You can’t perform that action at this time.