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.96

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

from typing import Type

import pytest

from heimdallm.bifrosts.sql import exc
from heimdallm.bifrosts.sql.common import FqColumn
from heimdallm.bifrosts.sql.sqlite.select.bifrost import Bifrost

from ..utils import dialects
from .utils import PermissiveConstraints


@dialects()
def test_where_alias(dialect: str, Bifrost: Type[Bifrost]):
    class MyConstraints(PermissiveConstraints):
        def condition_column_allowed(self, column: FqColumn) -> bool:
            return column.name == "t1.col"

    bifrost = Bifrost.validation_only(MyConstraints())

    query = """
    select t1.col as thing from t1
    where thing=42
    """
    bifrost.traverse(query)

    # same query, but the alias points to a disallowed table.column
    query = """
    select t2.col as thing from t2
    where thing=42
    """
    with pytest.raises(exc.IllegalConditionColumn) as e:
        bifrost.traverse(query)
    assert e.value.column.name == "t2.col"


@dialects()
def test_order_alias(dialect: str, Bifrost: Type[Bifrost]):
    class MyConstraints(PermissiveConstraints):
        def condition_column_allowed(self, column: FqColumn) -> bool:
            return column.name == "t1.col"

    bifrost = Bifrost.validation_only(MyConstraints())

    query = """
    select t1.col as thing from t1
    order by thing
    """
    bifrost.traverse(query)

    # same query, but the alias points to a disallowed table.column
    query = """
    select t2.col as thing from t2
    order by thing
    """
    with pytest.raises(exc.IllegalConditionColumn) as e:
        bifrost.traverse(query)
    assert e.value.column.name == "t2.col"


@dialects()
def test_select_function_alias(dialect: str, Bifrost: Type[Bifrost]):
    bifrost = Bifrost.validation_only(PermissiveConstraints())

    query = """
    select whatever(col) from t1
    """

    with pytest.raises(exc.UnqualifiedColumn) as e:
        bifrost.traverse(query, autofix=False)
    assert e.value.column == "col"


@dialects()
def test_group_by_alias(dialect: str, Bifrost: Type[Bifrost]):
    bifrost = Bifrost.validation_only(PermissiveConstraints())

    query = """
    SELECT COUNT(*) num_rented_movies,
        strftime('%Y', rental.rental_date) AS rental_year
    FROM rental
    JOIN customer ON rental.customer_id = customer.customer_id
    WHERE customer.customer_id = :customer_id
    GROUP BY rental_year
    LIMIT 20;
    """
    bifrost.traverse(query)


@dialects()
def test_count_star_alias(dialect: str, Bifrost: Type[Bifrost]):
    bifrost = Bifrost.validation_only(PermissiveConstraints())

    query = """
SELECT f.title AS movie_title, COUNT(*) AS rental_days
FROM rental r
JOIN inventory i ON r.inventory_id = i.inventory_id
JOIN film f ON i.film_id = f.film_id
WHERE r.customer_id = :customer_id
GROUP BY f.title
ORDER BY rental_days DESC
LIMIT 5;
    """

    bifrost.traverse(query)


@dialects()
def test_count_disallowed_column_alias(dialect: str, Bifrost: Type[Bifrost]):
    class GeneralConstraints(PermissiveConstraints):
        def select_column_allowed(self, fq_column: FqColumn) -> bool:
            return fq_column.name != "film_actor.film_id"

    bifrost = Bifrost.validation_only(GeneralConstraints())

    query = """
SELECT
    actor.actor_id,
    actor.first_name,
    actor.last_name,
    COUNT(film_actor.film_id) as film_count
FROM actor
JOIN film_actor ON actor.actor_id = film_actor.actor_id
GROUP BY actor.actor_id
ORDER BY film_count DESC
    """
    bifrost.traverse(query, autofix=False)

    # prove that the column isn't stripped by the reconstructor
    fixed = bifrost.traverse(query, autofix=True)
    assert "film_actor.film_id" in fixed


@dialects()
def test_subquery_alias_bleed(dialect: str, Bifrost: Type[Bifrost]):
    """An inner subquery alias should not mask an outer alias."""

    class GeneralConstraints(PermissiveConstraints):
        def condition_column_allowed(self, fq_column: FqColumn) -> bool:
            return fq_column.name not in {"t1.secret"}

    bifrost = Bifrost.validation_only(GeneralConstraints())

    query = """
    select
        t1.col as outer_alias,
        t1.secret as inner_alias
    from t1
    where outer_alias=(
        select t2.col as inner_alias from t2
        where inner_alias=42
    ) and inner_alias=42
    """
    with pytest.raises(exc.AliasConflict) as e:
        bifrost.traverse(query)
    assert e.value.alias == "inner_alias"


@dialects()
def test_subquery_as_alias(dialect: str, Bifrost: Type[Bifrost]):
    """A subquery can be selected as an alias, but all of the columns it references are
    going to be bound to the outer alias."""
    bifrost = Bifrost.validation_only(PermissiveConstraints())

    query = """
    select (select t1.col from t1 where t1.secret='a') as thing
    from t1
    where thing=42
    """

    bifrost.traverse(query)

    class MyConstraints(PermissiveConstraints):
        def select_column_allowed(self, column: FqColumn) -> bool:
            return column.name == "t1.col"

    bifrost = Bifrost.validation_only(MyConstraints())
    with pytest.raises(exc.IllegalConditionColumn) as e:
        bifrost.traverse(query)
    assert e.value.column.name == "t1.secret"


@dialects()
def test_selected_table_alias(dialect: str, Bifrost: Type[Bifrost]):
    """If the selected table is aliased and the query is reconstructed to have
    authoritative column names, use the authoritative selected table name."""
    bifrost = Bifrost.validation_only(PermissiveConstraints())

    query = """
    select col
    from t1 as table_alias
    """

    fixed = bifrost.traverse(query)
    assert "t1.col" in fixed

1	from typing import Type	1✔
2
3	import pytest	1✔
4
5	from heimdallm.bifrosts.sql import exc	1✔
6	from heimdallm.bifrosts.sql.common import FqColumn	1✔
7	from heimdallm.bifrosts.sql.sqlite.select.bifrost import Bifrost	1✔
8
9	from ..utils import dialects	1✔
10	from .utils import PermissiveConstraints	1✔
11
12
13	@dialects()	1✔
14	def test_where_alias(dialect: str, Bifrost: Type[Bifrost]):	1✔
15	class MyConstraints(PermissiveConstraints):	1✔
16	def condition_column_allowed(self, column: FqColumn) -> bool:	1✔
17	return column.name == "t1.col"	1✔
18
19	bifrost = Bifrost.validation_only(MyConstraints())	1✔
20
21	query = """	1✔
22	select t1.col as thing from t1
23	where thing=42
24	"""
25	bifrost.traverse(query)	1✔
26
27	# same query, but the alias points to a disallowed table.column
28	query = """	1✔
29	select t2.col as thing from t2
30	where thing=42
31	"""
32	with pytest.raises(exc.IllegalConditionColumn) as e:	1✔
33	bifrost.traverse(query)	1✔
34	assert e.value.column.name == "t2.col"	1✔
35
36
37	@dialects()	1✔
38	def test_order_alias(dialect: str, Bifrost: Type[Bifrost]):	1✔
39	class MyConstraints(PermissiveConstraints):	1✔
40	def condition_column_allowed(self, column: FqColumn) -> bool:	1✔
41	return column.name == "t1.col"	1✔
42
43	bifrost = Bifrost.validation_only(MyConstraints())	1✔
44
45	query = """	1✔
46	select t1.col as thing from t1
47	order by thing
48	"""
49	bifrost.traverse(query)	1✔
50
51	# same query, but the alias points to a disallowed table.column
52	query = """	1✔
53	select t2.col as thing from t2
54	order by thing
55	"""
56	with pytest.raises(exc.IllegalConditionColumn) as e:	1✔
57	bifrost.traverse(query)	1✔
58	assert e.value.column.name == "t2.col"	1✔
59
60
61	@dialects()	1✔
62	def test_select_function_alias(dialect: str, Bifrost: Type[Bifrost]):	1✔
63	bifrost = Bifrost.validation_only(PermissiveConstraints())	1✔
64
65	query = """	1✔
66	select whatever(col) from t1
67	"""
68
69	with pytest.raises(exc.UnqualifiedColumn) as e:	1✔
70	bifrost.traverse(query, autofix=False)	1✔
71	assert e.value.column == "col"	1✔
72
73
74	@dialects()	1✔
75	def test_group_by_alias(dialect: str, Bifrost: Type[Bifrost]):	1✔
76	bifrost = Bifrost.validation_only(PermissiveConstraints())	1✔
77
78	query = """	1✔
79	SELECT COUNT(*) num_rented_movies,
80	strftime('%Y', rental.rental_date) AS rental_year
81	FROM rental
82	JOIN customer ON rental.customer_id = customer.customer_id
83	WHERE customer.customer_id = :customer_id
84	GROUP BY rental_year
85	LIMIT 20;
86	"""
87	bifrost.traverse(query)	1✔
88
89
90	@dialects()	1✔
91	def test_count_star_alias(dialect: str, Bifrost: Type[Bifrost]):	1✔
92	bifrost = Bifrost.validation_only(PermissiveConstraints())	1✔
93
94	query = """	1✔
95	SELECT f.title AS movie_title, COUNT(*) AS rental_days
96	FROM rental r
97	JOIN inventory i ON r.inventory_id = i.inventory_id
98	JOIN film f ON i.film_id = f.film_id
99	WHERE r.customer_id = :customer_id
100	GROUP BY f.title
101	ORDER BY rental_days DESC
102	LIMIT 5;
103	"""
104
105	bifrost.traverse(query)	1✔
106
107
108	@dialects()	1✔
109	def test_count_disallowed_column_alias(dialect: str, Bifrost: Type[Bifrost]):	1✔
110	class GeneralConstraints(PermissiveConstraints):	1✔
111	def select_column_allowed(self, fq_column: FqColumn) -> bool:	1✔
112	return fq_column.name != "film_actor.film_id"	1✔
113
114	bifrost = Bifrost.validation_only(GeneralConstraints())	1✔
115
116	query = """	1✔
117	SELECT
118	actor.actor_id,
119	actor.first_name,
120	actor.last_name,
121	COUNT(film_actor.film_id) as film_count
122	FROM actor
123	JOIN film_actor ON actor.actor_id = film_actor.actor_id
124	GROUP BY actor.actor_id
125	ORDER BY film_count DESC
126	"""
127	bifrost.traverse(query, autofix=False)	1✔
128
129	# prove that the column isn't stripped by the reconstructor
130	fixed = bifrost.traverse(query, autofix=True)	1✔
131	assert "film_actor.film_id" in fixed	1✔
132
133
134	@dialects()	1✔
135	def test_subquery_alias_bleed(dialect: str, Bifrost: Type[Bifrost]):	1✔
136	"""An inner subquery alias should not mask an outer alias."""
137
138	class GeneralConstraints(PermissiveConstraints):	1✔
139	def condition_column_allowed(self, fq_column: FqColumn) -> bool:	1✔
140	return fq_column.name not in {"t1.secret"}	×
141
142	bifrost = Bifrost.validation_only(GeneralConstraints())	1✔
143
144	query = """	1✔
145	select
146	t1.col as outer_alias,
147	t1.secret as inner_alias
148	from t1
149	where outer_alias=(
150	select t2.col as inner_alias from t2
151	where inner_alias=42
152	) and inner_alias=42
153	"""
154	with pytest.raises(exc.AliasConflict) as e:	1✔
155	bifrost.traverse(query)	1✔
156	assert e.value.alias == "inner_alias"	1✔
157
158
159	@dialects()	1✔
160	def test_subquery_as_alias(dialect: str, Bifrost: Type[Bifrost]):	1✔
161	"""A subquery can be selected as an alias, but all of the columns it references are
162	going to be bound to the outer alias."""
163	bifrost = Bifrost.validation_only(PermissiveConstraints())	1✔
164
165	query = """	1✔
166	select (select t1.col from t1 where t1.secret='a') as thing
167	from t1
168	where thing=42
169	"""
170
171	bifrost.traverse(query)	1✔
172
173	class MyConstraints(PermissiveConstraints):	1✔
174	def select_column_allowed(self, column: FqColumn) -> bool:	1✔
175	return column.name == "t1.col"	1✔
176
177	bifrost = Bifrost.validation_only(MyConstraints())	1✔
178	with pytest.raises(exc.IllegalConditionColumn) as e:	1✔
179	bifrost.traverse(query)	1✔
180	assert e.value.column.name == "t1.secret"	1✔
181
182
183	@dialects()	1✔
184	def test_selected_table_alias(dialect: str, Bifrost: Type[Bifrost]):	1✔
185	"""If the selected table is aliased and the query is reconstructed to have
186	authoritative column names, use the authoritative selected table name."""
187	bifrost = Bifrost.validation_only(PermissiveConstraints())	1✔
188
189	query = """	1✔
190	select col
191	from t1 as table_alias
192	"""
193
194	fixed = bifrost.traverse(query)	1✔
195	assert "t1.col" in fixed	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