6143038651

Committed 11 Sep 2023 07:02AM UTC coverage: 94.055% (+0.5%) from 93.539%

Build # 6143038651

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #14 from amoffat/dev

Release 1.0.0

Run Details

433 of 495 branches covered (0.0%)

Branch coverage included in aggregate %.

694 of 694 new or added lines in 41 files covered. (100.0%)

2035 of 2129 relevant lines covered (95.58%)

0.96 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.11

/heimdallm/bifrosts/sql/tests/sql/select/test_subquery.py

from typing import Sequence, Type

import pytest

from heimdallm.bifrosts.sql import exc
from heimdallm.bifrosts.sql.bifrost import Bifrost
from heimdallm.bifrosts.sql.common import FqColumn, ParameterizedConstraint

from ..utils import dialects
from .utils import PermissiveConstraints


@dialects()
def test_where_in(dialect: str, Bifrost: Type[Bifrost]):
    """A subquery's selected columns are subject to the constraint validator"""

    class Conservative(PermissiveConstraints):
        def select_column_allowed(self, column: FqColumn) -> bool:
            return column.name in {"t1.col"}

    query = """
select t1.col from t1
where t1.id in (
    select t1.id from t1
    where t1.col='foo'
)
    """
    bifrost = Bifrost.validation_only(Conservative())
    with pytest.raises(exc.IllegalSelectedColumn) as e:
        bifrost.traverse(query)
    assert e.value.column == "t1.id"

    class Liberal(PermissiveConstraints):
        def select_column_allowed(self, column: FqColumn) -> bool:
            return column.name in {"t1.col", "t1.id"}

    bifrost = Bifrost.validation_only(Liberal())
    bifrost.traverse(query)


@dialects()
def test_subquery_bad_limit(dialect: str, Bifrost: Type[Bifrost]):
    """A subquery does not require a limit"""

    class MyConstraints(PermissiveConstraints):
        def max_limit(self):
            return 10

    query = """
select t1.col from t1
where t1.id in (
    select t1.id from t1
    where t1.col='foo'
)
limit 10
    """
    bifrost = Bifrost.validation_only(MyConstraints())
    bifrost.traverse(query, autofix=False)


@dialects()
def test_subquery_dont_fix_limit(dialect: str, Bifrost: Type[Bifrost]):
    """The reconstructor should NOT add a limit to a subquery"""

    class MyConstraints(PermissiveConstraints):
        def max_limit(self):
            return 10

    query = """
select t1.col from t1
where t1.id in (
    select t1.id from t1
    where t1.col='foo'
)
limit 5
    """
    bifrost = Bifrost.validation_only(MyConstraints())
    fixed = bifrost.traverse(query)
    # outer limit preserved
    assert "limit 5" in fixed.lower()
    # inner limit added
    assert "limit 10" not in fixed.lower()


@dialects()
def test_subquery_alias(dialect: str, Bifrost: Type[Bifrost]):
    """Tests that a subquery can be aliased and referenced by that alias"""

    query = """
select t1.t2col from (
    select t2.col as t2col from t2
) t1
    """

    class MyConstraints(PermissiveConstraints):
        def select_column_allowed(self, column: FqColumn) -> bool:
            return column.name in {"t2.col"}

    bifrost = Bifrost.validation_only(MyConstraints())
    bifrost.traverse(query)

    class NoSelectConstraints(PermissiveConstraints):
        def select_column_allowed(self, column: FqColumn) -> bool:
            return False

    bifrost = Bifrost.validation_only(NoSelectConstraints())
    with pytest.raises(exc.IllegalSelectedColumn) as e:
        bifrost.traverse(query)
        assert e.value.column == "t2.col"


@dialects()
def test_subquery_alias_conflict(dialect: str, Bifrost: Type[Bifrost]):
    """If a subquery is aliased, and that alias conflicts with a table name,
    then the query is ambiguous and should be rejected"""

    query = """
select t1.col from t1
join (
    select t2.col from t2
) as t1 on t1.id=t1.col
    """

    bifrost = Bifrost.validation_only(PermissiveConstraints())
    with pytest.raises(exc.AliasConflict) as e:
        bifrost.traverse(query)
        assert e.value.alias == "t1"


@dialects()
def test_subquery_alias_conflict2(dialect: str, Bifrost: Type[Bifrost]):
    """Alias conflicts anywhere in the query should be rejected"""

    query = """
select t1.col, t1.col2 as alias_col from (
    select t2.col as alias_col from t2
) as sq
    """

    bifrost = Bifrost.validation_only(PermissiveConstraints())
    with pytest.raises(exc.AliasConflict) as e:
        bifrost.traverse(query)
    assert e.value.alias == "alias_col"


@dialects()
def test_subquery_alias_conflict3(dialect: str, Bifrost: Type[Bifrost]):
    """Alias conflicts anywhere in the query should be rejected"""

    query = """
select t1.col from (
    select t1.col from t2 as t1
) t1
    """

    bifrost = Bifrost.validation_only(PermissiveConstraints())
    with pytest.raises(exc.AliasConflict) as e:
        bifrost.traverse(query)
    assert e.value.alias == "t1"


@dialects()
def test_subquery_fq(dialect: str, Bifrost: Type[Bifrost]):
    """A subquery should be able to have its single columns fully-qualified
    automatically"""

    query = """
select t1.col, t1.col2 from (
    select col, col2 from (
        select thing from t3
    ) t2
) t1
    """
    bifrost = Bifrost.validation_only(PermissiveConstraints())
    fixed = bifrost.traverse(query)
    assert "t2.col" in fixed.lower()
    assert "t2.col2" in fixed.lower()
    assert "t3.thing" in fixed.lower()


@dialects()
def test_subquery_parameterized_comparison(dialect: str, Bifrost: Type[Bifrost]):
    """Parameterized comparisons must exist outside of the subquery to count"""

    class MyConstraints(PermissiveConstraints):
        def parameterized_constraints(self) -> Sequence[ParameterizedConstraint]:
            return [ParameterizedConstraint(column="t1.email", placeholder="email")]

    bifrost = Bifrost.validation_only(MyConstraints())

    query = """
select t.email from (
    select t1.email from t1
    where t1.email=:email
) t
"""

    with pytest.raises(exc.MissingParameterizedConstraint) as e:
        bifrost.traverse(query)
    assert e.value.placeholder == "email"

1	from typing import Sequence, Type	1✔
2
3	import pytest	1✔
4
5	from heimdallm.bifrosts.sql import exc	1✔
6	from heimdallm.bifrosts.sql.bifrost import Bifrost	1✔
7	from heimdallm.bifrosts.sql.common import FqColumn, ParameterizedConstraint	1✔
8
9	from ..utils import dialects	1✔
10	from .utils import PermissiveConstraints	1✔
11
12
13	@dialects()	1✔
14	def test_where_in(dialect: str, Bifrost: Type[Bifrost]):	1✔
15	"""A subquery's selected columns are subject to the constraint validator"""
16
17	class Conservative(PermissiveConstraints):	1✔
18	def select_column_allowed(self, column: FqColumn) -> bool:	1✔
19	return column.name in {"t1.col"}	1✔
20
21	query = """	1✔
22	select t1.col from t1
23	where t1.id in (
24	select t1.id from t1
25	where t1.col='foo'
26	)
27	"""
28	bifrost = Bifrost.validation_only(Conservative())	1✔
29	with pytest.raises(exc.IllegalSelectedColumn) as e:	1✔
30	bifrost.traverse(query)	1✔
31	assert e.value.column == "t1.id"	1✔
32
33	class Liberal(PermissiveConstraints):	1✔
34	def select_column_allowed(self, column: FqColumn) -> bool:	1✔
35	return column.name in {"t1.col", "t1.id"}	1✔
36
37	bifrost = Bifrost.validation_only(Liberal())	1✔
38	bifrost.traverse(query)	1✔
39
40
41	@dialects()	1✔
42	def test_subquery_bad_limit(dialect: str, Bifrost: Type[Bifrost]):	1✔
43	"""A subquery does not require a limit"""
44
45	class MyConstraints(PermissiveConstraints):	1✔
46	def max_limit(self):	1✔
47	return 10	1✔
48
49	query = """	1✔
50	select t1.col from t1
51	where t1.id in (
52	select t1.id from t1
53	where t1.col='foo'
54	)
55	limit 10
56	"""
57	bifrost = Bifrost.validation_only(MyConstraints())	1✔
58	bifrost.traverse(query, autofix=False)	1✔
59
60
61	@dialects()	1✔
62	def test_subquery_dont_fix_limit(dialect: str, Bifrost: Type[Bifrost]):	1✔
63	"""The reconstructor should NOT add a limit to a subquery"""
64
65	class MyConstraints(PermissiveConstraints):	1✔
66	def max_limit(self):	1✔
67	return 10	1✔
68
69	query = """	1✔
70	select t1.col from t1
71	where t1.id in (
72	select t1.id from t1
73	where t1.col='foo'
74	)
75	limit 5
76	"""
77	bifrost = Bifrost.validation_only(MyConstraints())	1✔
78	fixed = bifrost.traverse(query)	1✔
79	# outer limit preserved
80	assert "limit 5" in fixed.lower()	1✔
81	# inner limit added
82	assert "limit 10" not in fixed.lower()	1✔
83
84
85	@dialects()	1✔
86	def test_subquery_alias(dialect: str, Bifrost: Type[Bifrost]):	1✔
87	"""Tests that a subquery can be aliased and referenced by that alias"""
88
89	query = """	1✔
90	select t1.t2col from (
91	select t2.col as t2col from t2
92	) t1
93	"""
94
95	class MyConstraints(PermissiveConstraints):	1✔
96	def select_column_allowed(self, column: FqColumn) -> bool:	1✔
97	return column.name in {"t2.col"}	1✔
98
99	bifrost = Bifrost.validation_only(MyConstraints())	1✔
100	bifrost.traverse(query)	1✔
101
102	class NoSelectConstraints(PermissiveConstraints):	1✔
103	def select_column_allowed(self, column: FqColumn) -> bool:	1✔
104	return False	1✔
105
106	bifrost = Bifrost.validation_only(NoSelectConstraints())	1✔
107	with pytest.raises(exc.IllegalSelectedColumn) as e:	1✔
108	bifrost.traverse(query)	1✔
109	assert e.value.column == "t2.col"	×
110
111
112	@dialects()	1✔
113	def test_subquery_alias_conflict(dialect: str, Bifrost: Type[Bifrost]):	1✔
114	"""If a subquery is aliased, and that alias conflicts with a table name,
115	then the query is ambiguous and should be rejected"""
116
117	query = """	1✔
118	select t1.col from t1
119	join (
120	select t2.col from t2
121	) as t1 on t1.id=t1.col
122	"""
123
124	bifrost = Bifrost.validation_only(PermissiveConstraints())	1✔
125	with pytest.raises(exc.AliasConflict) as e:	1✔
126	bifrost.traverse(query)	1✔
127	assert e.value.alias == "t1"	×
128
129
130	@dialects()	1✔
131	def test_subquery_alias_conflict2(dialect: str, Bifrost: Type[Bifrost]):	1✔
132	"""Alias conflicts anywhere in the query should be rejected"""
133
134	query = """	1✔
135	select t1.col, t1.col2 as alias_col from (
136	select t2.col as alias_col from t2
137	) as sq
138	"""
139
140	bifrost = Bifrost.validation_only(PermissiveConstraints())	1✔
141	with pytest.raises(exc.AliasConflict) as e:	1✔
142	bifrost.traverse(query)	1✔
143	assert e.value.alias == "alias_col"	1✔
144
145
146	@dialects()	1✔
147	def test_subquery_alias_conflict3(dialect: str, Bifrost: Type[Bifrost]):	1✔
148	"""Alias conflicts anywhere in the query should be rejected"""
149
150	query = """	1✔
151	select t1.col from (
152	select t1.col from t2 as t1
153	) t1
154	"""
155
156	bifrost = Bifrost.validation_only(PermissiveConstraints())	1✔
157	with pytest.raises(exc.AliasConflict) as e:	1✔
158	bifrost.traverse(query)	1✔
159	assert e.value.alias == "t1"	1✔
160
161
162	@dialects()	1✔
163	def test_subquery_fq(dialect: str, Bifrost: Type[Bifrost]):	1✔
164	"""A subquery should be able to have its single columns fully-qualified
165	automatically"""
166
167	query = """	1✔
168	select t1.col, t1.col2 from (
169	select col, col2 from (
170	select thing from t3
171	) t2
172	) t1
173	"""
174	bifrost = Bifrost.validation_only(PermissiveConstraints())	1✔
175	fixed = bifrost.traverse(query)	1✔
176	assert "t2.col" in fixed.lower()	1✔
177	assert "t2.col2" in fixed.lower()	1✔
178	assert "t3.thing" in fixed.lower()	1✔
179
180
181	@dialects()	1✔
182	def test_subquery_parameterized_comparison(dialect: str, Bifrost: Type[Bifrost]):	1✔
183	"""Parameterized comparisons must exist outside of the subquery to count"""
184
185	class MyConstraints(PermissiveConstraints):	1✔
186	def parameterized_constraints(self) -> Sequence[ParameterizedConstraint]:	1✔
187	return [ParameterizedConstraint(column="t1.email", placeholder="email")]	1✔
188
189	bifrost = Bifrost.validation_only(MyConstraints())	1✔
190
191	query = """	1✔
192	select t.email from (
193	select t1.email from t1
194	where t1.email=:email
195	) t
196	"""
197
198	with pytest.raises(exc.MissingParameterizedConstraint) as e:	1✔
199	bifrost.traverse(query)	1✔
200	assert e.value.placeholder == "email"	1✔

amoffat / HeimdaLLM / 6143038651

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous