import pytest
import tempfile
import shutil
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'apexbase', 'python'))
try:
from apexbase import ApexClient
except ImportError as e:
pytest.skip(f"ApexBase not available: {e}", allow_module_level=True)
def xfail_sql(client, sql):
return client.execute(sql)
@pytest.fixture
def tmp_client():
d = tempfile.mkdtemp()
c = ApexClient(dirpath=d)
yield c
c.close()
shutil.rmtree(d, ignore_errors=True)
class TestSetOperations:
def setup_method(self):
self.d = tempfile.mkdtemp()
self.c = ApexClient(dirpath=self.d)
self.c.execute("CREATE TABLE a (val INT)")
self.c.execute("INSERT INTO a VALUES (1),(2),(3),(4)")
self.c.execute("CREATE TABLE b (val INT)")
self.c.execute("INSERT INTO b VALUES (2),(3),(5),(6)")
def teardown_method(self):
self.c.close()
shutil.rmtree(self.d, ignore_errors=True)
def test_union_dedup(self):
rv = xfail_sql(self.c, "SELECT val FROM a UNION SELECT val FROM b ORDER BY val")
vals = [r['val'] for r in rv]
assert sorted(vals) == [1, 2, 3, 4, 5, 6]
assert len(vals) == len(set(vals))
def test_union_all_keeps_dupes(self):
rv = xfail_sql(self.c, "SELECT val FROM a UNION ALL SELECT val FROM b ORDER BY val")
assert len(rv) == 8
def test_intersect(self):
rv = xfail_sql(self.c, "SELECT val FROM a INTERSECT SELECT val FROM b ORDER BY val")
assert [r['val'] for r in rv] == [2, 3]
def test_except(self):
rv = xfail_sql(self.c, "SELECT val FROM a EXCEPT SELECT val FROM b ORDER BY val")
assert [r['val'] for r in rv] == [1, 4]
def test_union_with_order_by_limit(self):
rv = xfail_sql(self.c, "SELECT val FROM a UNION SELECT val FROM b ORDER BY val LIMIT 3")
assert len(rv) == 3
vals = [r['val'] for r in rv]
assert vals == sorted(vals)
def test_union_with_order_by_offset(self):
rv = xfail_sql(self.c, "SELECT val FROM a UNION SELECT val FROM b ORDER BY val LIMIT 3 OFFSET 2")
assert len(rv) == 3
vals = [r['val'] for r in rv]
assert vals == sorted(vals)
def test_intersect_empty_result(self):
self.c.execute("CREATE TABLE c (val INT)")
self.c.execute("INSERT INTO c VALUES (99),(100)")
rv = xfail_sql(self.c, "SELECT val FROM a INTERSECT SELECT val FROM c ORDER BY val")
assert len(rv) == 0
def test_except_no_overlap_returns_all_left(self):
self.c.execute("CREATE TABLE d (val INT)")
self.c.execute("INSERT INTO d VALUES (99),(100)")
rv = xfail_sql(self.c, "SELECT val FROM a EXCEPT SELECT val FROM d ORDER BY val")
assert [r['val'] for r in rv] == [1, 2, 3, 4]
def test_union_all_count(self):
rv = xfail_sql(self.c, "SELECT COUNT(*) FROM (SELECT val FROM a UNION ALL SELECT val FROM b) t")
assert rv.scalar() == 8
def test_self_union_dedup(self):
rv = xfail_sql(self.c, "SELECT val FROM a UNION SELECT val FROM a ORDER BY val")
assert [r['val'] for r in rv] == [1, 2, 3, 4]
def test_self_union_all_doubles(self):
rv = xfail_sql(self.c, "SELECT val FROM a UNION ALL SELECT val FROM a ORDER BY val")
assert len(rv) == 8
class TestSetOperationsMultiCol:
def setup_method(self):
self.d = tempfile.mkdtemp()
self.c = ApexClient(dirpath=self.d)
self.c.execute("CREATE TABLE x (id INT, name STRING)")
self.c.execute("INSERT INTO x (id, name) VALUES (1,'Alice'),(2,'Bob'),(3,'Carol')")
self.c.execute("CREATE TABLE y (id INT, name STRING)")
self.c.execute("INSERT INTO y (id, name) VALUES (2,'Bob'),(3,'Carol'),(4,'Dave')")
def teardown_method(self):
self.c.close()
shutil.rmtree(self.d, ignore_errors=True)
def test_union_multi_col_dedup(self):
rv = xfail_sql(self.c, "SELECT id, name FROM x UNION SELECT id, name FROM y ORDER BY id")
ids = [r['id'] for r in rv]
assert ids == [1, 2, 3, 4]
assert len(ids) == len(set(ids))
def test_union_all_multi_col_count(self):
rv = xfail_sql(self.c, "SELECT id, name FROM x UNION ALL SELECT id, name FROM y ORDER BY id")
assert len(rv) == 6
def test_intersect_multi_col(self):
rv = xfail_sql(self.c, "SELECT id, name FROM x INTERSECT SELECT id, name FROM y ORDER BY id")
ids = [r['id'] for r in rv]
assert ids == [2, 3]
def test_except_multi_col(self):
rv = xfail_sql(self.c, "SELECT id, name FROM x EXCEPT SELECT id, name FROM y ORDER BY id")
ids = [r['id'] for r in rv]
assert ids == [1]
def test_union_string_values_dedup(self):
rv = xfail_sql(self.c, "SELECT name FROM x UNION SELECT name FROM y ORDER BY name")
names = [r['name'] for r in rv]
assert names == ['Alice', 'Bob', 'Carol', 'Dave']
assert len(names) == len(set(names))
class TestSubqueries:
def setup_method(self):
self.d = tempfile.mkdtemp()
self.c = ApexClient(dirpath=self.d)
self.c.execute("CREATE TABLE emp (name STRING, dept STRING, salary INT)")
self.c.execute("""INSERT INTO emp VALUES
('Alice','Eng',90000),('Bob','Eng',80000),
('Carol','Sales',70000),('Dave','Sales',75000),('Eve','HR',60000)""")
def teardown_method(self):
self.c.close()
shutil.rmtree(self.d, ignore_errors=True)
def test_scalar_subquery_in_where(self):
rv = xfail_sql(self.c,
"SELECT name FROM emp WHERE salary > (SELECT AVG(salary) FROM emp) ORDER BY name")
assert 'Alice' in [r['name'] for r in rv]
def test_in_subquery(self):
rv = xfail_sql(self.c,
"SELECT name FROM emp WHERE dept IN "
"(SELECT DISTINCT dept FROM emp WHERE salary > 75000) ORDER BY name")
names = [r['name'] for r in rv]
assert 'Alice' in names and 'Bob' in names
def test_not_in_subquery(self):
rv = xfail_sql(self.c,
"SELECT name FROM emp WHERE dept NOT IN "
"(SELECT DISTINCT dept FROM emp WHERE salary > 80000) ORDER BY name")
names = [r['name'] for r in rv]
assert 'Carol' in names and 'Alice' not in names
def test_exists_correlated(self):
rv = xfail_sql(self.c, """
SELECT DISTINCT dept FROM emp e1
WHERE EXISTS (SELECT 1 FROM emp e2 WHERE e2.dept=e1.dept AND e2.name!=e1.name)
ORDER BY dept""")
depts = [r['dept'] for r in rv]
assert 'Eng' in depts and 'HR' not in depts
def test_scalar_subquery_in_select(self):
rv = xfail_sql(self.c,
"SELECT name, (SELECT MAX(salary) FROM emp) as mx FROM emp ORDER BY name")
rows = {r['name']: r for r in rv}
assert rows['Alice']['mx'] == 90000 and rows['Eve']['mx'] == 90000
class TestMultipleCTEs:
def setup_method(self):
self.d = tempfile.mkdtemp()
self.c = ApexClient(dirpath=self.d)
self.c.execute("CREATE TABLE orders (customer STRING, amount INT)")
self.c.execute("""INSERT INTO orders VALUES
('Alice',100),('Alice',200),('Bob',300),('Bob',50),
('Carol',400),('Carol',80)""")
def teardown_method(self):
self.c.close()
shutil.rmtree(self.d, ignore_errors=True)
def test_two_ctes_chained(self):
rv = xfail_sql(self.c, """
WITH totals AS (
SELECT customer, SUM(amount) as total FROM orders GROUP BY customer
),
big AS (
SELECT customer, total FROM totals WHERE total > 200
)
SELECT customer FROM big ORDER BY total DESC""")
names = [r['customer'] for r in rv]
assert 'Carol' in names and 'Bob' in names
def test_cte_with_window_function(self):
rv = xfail_sql(self.c, """
WITH base AS (
SELECT customer, amount,
ROW_NUMBER() OVER (PARTITION BY customer ORDER BY amount DESC) as rn
FROM orders
)
SELECT customer, amount FROM base WHERE rn = 1 ORDER BY customer""")
rows = {r['customer']: r['amount'] for r in rv}
assert rows.get('Alice') == 200
assert rows.get('Bob') == 300
assert rows.get('Carol') == 400
class TestAdvancedWindowFunctions:
def setup_method(self):
self.d = tempfile.mkdtemp()
self.c = ApexClient(dirpath=self.d)
self.c.execute("CREATE TABLE scores (player STRING, game INT, score INT)")
self.c.execute("""INSERT INTO scores VALUES
('Alice',1,90),('Alice',2,85),('Alice',3,95),
('Bob',1,70),('Bob',2,80),('Bob',3,75)""")
def teardown_method(self):
self.c.close()
shutil.rmtree(self.d, ignore_errors=True)
def test_running_sum_partition(self):
rv = xfail_sql(self.c,
"SELECT player, game, SUM(score) OVER "
"(PARTITION BY player ORDER BY game) as rs "
"FROM scores ORDER BY player, game")
alice = sorted([r for r in rv if r['player'] == 'Alice'], key=lambda x: x['game'])
assert alice[0]['rs'] == 90
assert alice[1]['rs'] == 175
assert alice[2]['rs'] == 270
def test_rank_vs_dense_rank_ties(self):
self.c.execute("CREATE TABLE t2 (val INT)")
self.c.execute("INSERT INTO t2 VALUES (10),(10),(20),(30)")
rv = xfail_sql(self.c,
"SELECT val, RANK() OVER (ORDER BY val) as rnk, "
"DENSE_RANK() OVER (ORDER BY val) as drnk FROM t2 ORDER BY val, rnk")
rows = rv.to_dict()
tens = [r for r in rows if r['val'] == 10]
assert all(r['rnk'] == 1 for r in tens)
assert all(r['drnk'] == 1 for r in tens)
twenty = next(r for r in rows if r['val'] == 20)
assert twenty['rnk'] == 3 assert twenty['drnk'] == 2
def test_lag_lead(self):
rv = xfail_sql(self.c,
"SELECT player, game, "
"LAG(score) OVER (PARTITION BY player ORDER BY game) as prev, "
"LEAD(score) OVER (PARTITION BY player ORDER BY game) as nxt "
"FROM scores ORDER BY player, game")
rows = {(r['player'], r['game']): r for r in rv}
assert rows[('Alice', 1)]['prev'] is None
assert rows[('Alice', 2)]['prev'] == 90
assert rows[('Alice', 3)]['nxt'] is None
def test_first_value(self):
rv = xfail_sql(self.c,
"SELECT player, game, "
"FIRST_VALUE(score) OVER (PARTITION BY player ORDER BY game) as first "
"FROM scores ORDER BY player, game")
rows = {(r['player'], r['game']): r for r in rv}
assert rows[('Alice', 2)]['first'] == 90
assert rows[('Bob', 3)]['first'] == 70
class TestNullHandling:
def setup_method(self):
self.d = tempfile.mkdtemp()
self.c = ApexClient(dirpath=self.d)
self.c.execute("CREATE TABLE t (a INT, b STRING)")
self.c.execute("INSERT INTO t VALUES (1,'x'),(NULL,'y'),(3,NULL),(NULL,NULL)")
def teardown_method(self):
self.c.close()
shutil.rmtree(self.d, ignore_errors=True)
def test_is_null_filter(self):
assert len(self.c.execute("SELECT a FROM t WHERE a IS NULL")) == 2
def test_is_not_null_filter(self):
rv = self.c.execute("SELECT a FROM t WHERE a IS NOT NULL ORDER BY a")
assert [r['a'] for r in rv] == [1, 3]
def test_null_equals_null_is_empty(self):
assert len(self.c.execute("SELECT a FROM t WHERE a = NULL")) == 0
def test_coalesce_int(self):
rv = self.c.execute("SELECT COALESCE(a, 0) as v FROM t ORDER BY v")
assert 0 in [r['v'] for r in rv]
def test_coalesce_string(self):
rv = self.c.execute("SELECT COALESCE(b, 'missing') as v FROM t ORDER BY v")
assert 'missing' in [r['v'] for r in rv]
def test_count_star_includes_nulls(self):
assert self.c.execute("SELECT COUNT(*) as c FROM t").scalar() == 4
def test_count_column_excludes_nulls(self):
assert self.c.execute("SELECT COUNT(a) as c FROM t").scalar() == 2
def test_sum_ignores_nulls(self):
assert self.c.execute("SELECT SUM(a) as s FROM t").scalar() == 4
def test_avg_ignores_nulls(self):
assert self.c.execute("SELECT AVG(a) as v FROM t").scalar() == 2.0
def test_group_by_null_forms_own_group(self):
rv = self.c.execute("SELECT a, COUNT(*) as cnt FROM t GROUP BY a ORDER BY a")
null_row = next((r for r in rv if r['a'] is None), None)
assert null_row is not None and null_row['cnt'] == 2
def test_ifnull(self):
rv = self.c.execute("SELECT IFNULL(a, -1) as v FROM t ORDER BY v")
assert -1 in [r['v'] for r in rv]
class TestCaseExpressions:
def setup_method(self):
self.d = tempfile.mkdtemp()
self.c = ApexClient(dirpath=self.d)
self.c.execute("CREATE TABLE t (name STRING, score INT)")
self.c.execute("""INSERT INTO t VALUES
('Alice',95),('Bob',75),('Carol',55),('Dave',45),('Eve',NULL)""")
def teardown_method(self):
self.c.close()
shutil.rmtree(self.d, ignore_errors=True)
def test_case_when_grade(self):
rv = self.c.execute("""
SELECT name,
CASE WHEN score >= 90 THEN 'A'
WHEN score >= 70 THEN 'B'
WHEN score >= 50 THEN 'C'
ELSE 'F'
END as grade
FROM t ORDER BY name""")
rows = {r['name']: r['grade'] for r in rv}
assert rows['Alice'] == 'A' and rows['Bob'] == 'B'
assert rows['Carol'] == 'C' and rows['Dave'] == 'F'
def test_case_null_detection(self):
rv = self.c.execute("""
SELECT name,
CASE WHEN score IS NULL THEN 'no score' ELSE 'has score' END as status
FROM t ORDER BY name""")
rows = {r['name']: r['status'] for r in rv}
assert rows['Eve'] == 'no score' and rows['Alice'] == 'has score'
def test_case_in_group_by(self):
rv = xfail_sql(self.c, """
SELECT CASE WHEN score >= 70 THEN 'pass' ELSE 'fail' END as result,
COUNT(*) as cnt
FROM t WHERE score IS NOT NULL
GROUP BY result ORDER BY result""")
rows = {r['result']: r['cnt'] for r in rv}
assert rows.get('pass') == 2 and rows.get('fail') == 2
class TestStringFunctions:
def setup_method(self):
self.d = tempfile.mkdtemp()
self.c = ApexClient(dirpath=self.d)
self.c.execute("CREATE TABLE t (s STRING)")
self.c.execute("""INSERT INTO t VALUES
('Hello World'),(' spaces '),('foo BAR'),('abc123')""")
def teardown_method(self):
self.c.close()
shutil.rmtree(self.d, ignore_errors=True)
def test_upper(self):
assert self.c.execute("SELECT UPPER(s) as u FROM t WHERE s='foo BAR'").first()['u'] == 'FOO BAR'
def test_lower(self):
assert self.c.execute("SELECT LOWER(s) as l FROM t WHERE s='Hello World'").first()['l'] == 'hello world'
def test_length(self):
assert self.c.execute("SELECT LENGTH(s) as n FROM t WHERE s='Hello World'").first()['n'] == 11
def test_trim(self):
assert self.c.execute("SELECT TRIM(s) as t FROM t WHERE s=' spaces '").first()['t'] == 'spaces'
def test_substr(self):
assert self.c.execute("SELECT SUBSTR(s,1,5) as sub FROM t WHERE s='Hello World'").first()['sub'] == 'Hello'
def test_replace(self):
rv = xfail_sql(self.c, "SELECT REPLACE(s,'World','ApexBase') as r FROM t WHERE s='Hello World'")
assert rv.first()['r'] == 'Hello ApexBase'
def test_like_contains(self):
assert len(self.c.execute("SELECT s FROM t WHERE s LIKE '%World%'")) == 1
def test_like_prefix(self):
assert len(self.c.execute("SELECT s FROM t WHERE s LIKE 'Hello%'")) == 1
def test_like_suffix(self):
assert len(self.c.execute("SELECT s FROM t WHERE s LIKE '%123'")) == 1
def test_case_sensitive_string_filter(self):
assert len(self.c.execute("SELECT s FROM t WHERE s = 'foo BAR'")) == 1
assert len(self.c.execute("SELECT s FROM t WHERE s = 'FOO BAR'")) == 0
class TestNumericEdgeCases:
def setup_method(self):
self.d = tempfile.mkdtemp()
self.c = ApexClient(dirpath=self.d)
self.c.execute("CREATE TABLE t (a INT, b FLOAT)")
self.c.execute("INSERT INTO t VALUES (0,0.0),(1,1.5),(100,3.14),(-5,-2.7),(2,0.001)")
def teardown_method(self):
self.c.close()
shutil.rmtree(self.d, ignore_errors=True)
def test_abs_negative(self):
assert self.c.execute("SELECT ABS(a) as v FROM t WHERE a=-5").first()['v'] == 5
def test_round(self):
assert self.c.execute("SELECT ROUND(b,1) as r FROM t WHERE a=100").first()['r'] == 3.1
def test_float_multiply(self):
val = self.c.execute("SELECT b*2 as v FROM t WHERE a=1").first()['v']
assert abs(val - 3.0) < 1e-9
def test_max_min_with_negatives(self):
row = self.c.execute("SELECT MAX(a) as mx, MIN(a) as mn FROM t").first()
assert row['mx'] == 100 and row['mn'] == -5
def test_sum_with_zero(self):
assert self.c.execute("SELECT SUM(a) as s FROM t WHERE a=0").scalar() == 0
def test_between_inclusive(self):
rv = self.c.execute("SELECT a FROM t WHERE a BETWEEN 1 AND 100 ORDER BY a")
vals = [r['a'] for r in rv]
assert vals == [1, 2, 100]
def test_negative_values_order(self):
rv = self.c.execute("SELECT a FROM t ORDER BY a ASC")
vals = [r['a'] for r in rv]
assert vals[0] == -5 and vals[-1] == 100
class TestJoins:
def setup_method(self):
self.d = tempfile.mkdtemp()
self.c = ApexClient(dirpath=self.d)
self.c.execute("CREATE TABLE emp (name STRING, dept_id INT)")
self.c.execute("INSERT INTO emp VALUES ('Alice',1),('Bob',2),('Carol',1),('Dave',NULL)")
self.c.execute("CREATE TABLE dept (id INT, dept_name STRING)")
self.c.execute("INSERT INTO dept VALUES (1,'Eng'),(2,'Sales'),(3,'HR')")
def teardown_method(self):
self.c.close()
shutil.rmtree(self.d, ignore_errors=True)
def test_inner_join_excludes_nulls(self):
rv = self.c.execute(
"SELECT e.name, d.dept_name FROM emp e JOIN dept d ON e.dept_id=d.id ORDER BY e.name")
names = [r['name'] for r in rv]
assert 'Alice' in names and 'Dave' not in names
def test_left_join_includes_unmatched(self):
rv = self.c.execute(
"SELECT e.name, d.dept_name FROM emp e LEFT JOIN dept d ON e.dept_id=d.id ORDER BY e.name")
assert len(rv) == 4
dave = next(r for r in rv if r['name'] == 'Dave')
assert dave['dept_name'] is None
def test_join_with_where_filter(self):
rv = self.c.execute("""
SELECT e.name FROM emp e JOIN dept d ON e.dept_id=d.id
WHERE d.dept_name='Eng' ORDER BY e.name""")
assert [r['name'] for r in rv] == ['Alice', 'Carol']
def test_self_join(self):
rv = xfail_sql(self.c,
"SELECT a.name as e1, b.name as e2 FROM emp a "
"JOIN emp b ON a.dept_id=b.dept_id AND a.name < b.name ORDER BY a.name")
pairs = [(r['e1'], r['e2']) for r in rv]
assert ('Alice', 'Carol') in pairs
def test_cross_join_count(self):
rv = xfail_sql(self.c,
"SELECT COUNT(*) as c FROM emp CROSS JOIN dept")
assert rv.scalar() == 12
def test_join_aggregate(self):
rv = self.c.execute("""
SELECT d.dept_name, COUNT(e.name) as cnt
FROM dept d LEFT JOIN emp e ON d.id=e.dept_id
GROUP BY d.dept_name ORDER BY d.dept_name""")
rows = {r['dept_name']: r['cnt'] for r in rv}
assert rows['Eng'] == 2 and rows['Sales'] == 1 and rows['HR'] == 0
class TestOuterJoinQualifiedKeys:
def setup_method(self):
self.d = tempfile.mkdtemp()
self.c = ApexClient(dirpath=self.d)
self.c.execute("CREATE TABLE l (id INT, name STRING)")
self.c.execute("CREATE TABLE r (id INT, tag STRING)")
self.c.execute("INSERT INTO l VALUES (1,'a'),(2,'b')")
self.c.execute("INSERT INTO r VALUES (2,'x'),(3,'y')")
def teardown_method(self):
self.c.close()
shutil.rmtree(self.d, ignore_errors=True)
def test_right_join_preserves_qualified_right_key(self):
rv = self.c.execute("""
SELECT l.id AS lid, r.id AS rid, l.name, r.tag
FROM l RIGHT JOIN r ON l.id = r.id
ORDER BY r.tag
""")
assert rv.to_dict() == [
{"lid": 2, "rid": 2, "name": "b", "tag": "x"},
{"lid": None, "rid": 3, "name": None, "tag": "y"},
]
def test_full_outer_join_preserves_both_join_keys(self):
rv = self.c.execute("""
SELECT l.id AS lid, r.id AS rid, l.name, r.tag
FROM l FULL OUTER JOIN r ON l.id = r.id
ORDER BY COALESCE(l.id, r.id)
""")
assert rv.to_dict() == [
{"lid": 1, "rid": None, "name": "a", "tag": None},
{"lid": 2, "rid": 2, "name": "b", "tag": "x"},
{"lid": None, "rid": 3, "name": None, "tag": "y"},
]
class TestGroupByHaving:
def setup_method(self):
self.d = tempfile.mkdtemp()
self.c = ApexClient(dirpath=self.d)
self.c.execute("CREATE TABLE sales (rep STRING, region STRING, amount INT)")
self.c.execute("""INSERT INTO sales VALUES
('Alice','North',100),('Alice','South',200),
('Bob','North',150),('Bob','North',50),
('Carol','South',300),('Dave','East',10)""")
def teardown_method(self):
self.c.close()
shutil.rmtree(self.d, ignore_errors=True)
def test_having_sum(self):
rv = self.c.execute("""
SELECT rep, SUM(amount) as total FROM sales
GROUP BY rep HAVING SUM(amount) > 150 ORDER BY total DESC""")
reps = [r['rep'] for r in rv]
assert 'Carol' in reps and 'Alice' in reps and 'Dave' not in reps
def test_having_count(self):
rv = self.c.execute("""
SELECT rep, COUNT(*) as cnt FROM sales
GROUP BY rep HAVING COUNT(*) > 1 ORDER BY rep""")
reps = [r['rep'] for r in rv]
assert 'Alice' in reps and 'Bob' in reps and 'Carol' not in reps
def test_group_by_two_columns(self):
rv = self.c.execute("""
SELECT rep, region, SUM(amount) as total FROM sales
GROUP BY rep, region ORDER BY rep, region""")
rows = {(r['rep'], r['region']): r['total'] for r in rv}
assert rows[('Alice', 'North')] == 100 and rows[('Bob', 'North')] == 200
def test_order_by_aggregate(self):
rv = self.c.execute("""
SELECT rep, SUM(amount) as total FROM sales GROUP BY rep ORDER BY SUM(amount) DESC""")
rows = rv.to_dict()
totals = [r['total'] for r in rows]
assert totals == sorted(totals, reverse=True), f"Expected DESC order, got {totals}"
assert rows[-1]['rep'] == 'Dave'
def test_having_avg(self):
rv = self.c.execute("""
SELECT region, AVG(amount) as avg_amt FROM sales
GROUP BY region HAVING AVG(amount) > 100 ORDER BY region""")
regions = [r['region'] for r in rv]
assert 'South' in regions
class TestLimitOffsetEdge:
def setup_method(self):
self.d = tempfile.mkdtemp()
self.c = ApexClient(dirpath=self.d)
self.c.execute("CREATE TABLE t (n INT)")
self.c.execute("INSERT INTO t VALUES " + ",".join(f"({i})" for i in range(10)))
def teardown_method(self):
self.c.close()
shutil.rmtree(self.d, ignore_errors=True)
def test_limit_exceeds_rows(self):
assert len(self.c.execute("SELECT n FROM t LIMIT 999")) == 10
def test_offset_skips_rows(self):
rv = self.c.execute("SELECT n FROM t ORDER BY n LIMIT 3 OFFSET 5")
assert [r['n'] for r in rv] == [5, 6, 7]
def test_offset_equals_row_count(self):
assert len(self.c.execute("SELECT n FROM t ORDER BY n LIMIT 5 OFFSET 10")) == 0
def test_offset_exceeds_row_count(self):
assert len(self.c.execute("SELECT n FROM t ORDER BY n LIMIT 5 OFFSET 100")) == 0
def test_offset_last_row(self):
rv = self.c.execute("SELECT n FROM t ORDER BY n LIMIT 1 OFFSET 9")
assert len(rv) == 1 and rv.first()['n'] == 9
def test_limit_zero(self):
rv = xfail_sql(self.c, "SELECT n FROM t ORDER BY n LIMIT 0")
assert len(rv) == 0
def test_distinct_with_limit(self):
self.c.execute("INSERT INTO t VALUES (0),(1),(2)")
rv = self.c.execute("SELECT DISTINCT n FROM t ORDER BY n LIMIT 3")
assert len(rv) == 3
class TestOrderByNullHandling:
def setup_method(self):
self.d = tempfile.mkdtemp()
self.c = ApexClient(dirpath=self.d)
self.c.execute("CREATE TABLE t (v INT)")
self.c.execute("INSERT INTO t VALUES (NULL),(2),(1),(NULL),(3)")
def teardown_method(self):
self.c.close()
shutil.rmtree(self.d, ignore_errors=True)
def test_order_by_nulls_first_last(self):
asc = self.c.execute("SELECT v FROM t ORDER BY v NULLS FIRST").to_dict()
desc = self.c.execute("SELECT v FROM t ORDER BY v DESC NULLS LAST").to_dict()
assert [r["v"] for r in asc] == [None, None, 1, 2, 3]
assert [r["v"] for r in desc] == [3, 2, 1, None, None]
class TestInsertSelect:
def setup_method(self):
self.d = tempfile.mkdtemp()
self.c = ApexClient(dirpath=self.d)
self.c.execute("CREATE TABLE src (name STRING, age INT)")
self.c.execute("INSERT INTO src VALUES ('Alice',25),('Bob',30),('Carol',22)")
self.c.execute("CREATE TABLE dst (name STRING, age INT)")
def teardown_method(self):
self.c.close()
shutil.rmtree(self.d, ignore_errors=True)
def test_insert_select_all(self):
self.c.execute("INSERT INTO dst SELECT name, age FROM src")
assert self.c.execute("SELECT COUNT(*) as c FROM dst").scalar() == 3
def test_insert_select_with_where(self):
self.c.execute("INSERT INTO dst SELECT name, age FROM src WHERE age >= 25")
assert self.c.execute("SELECT COUNT(*) as c FROM dst").scalar() == 2
def test_insert_select_with_transform(self):
self.c.execute("INSERT INTO dst SELECT name, age+1 as age FROM src WHERE name='Alice'")
row = self.c.execute("SELECT age FROM dst ORDER BY age").first()
assert row['age'] == 26
class TestPersistenceReopen:
def test_data_survives_close_reopen(self):
with tempfile.TemporaryDirectory() as d:
c1 = ApexClient(dirpath=d)
c1.create_table('t')
c1.store([{'x': i} for i in range(50)])
c1.flush()
c1.close()
c2 = ApexClient(dirpath=d)
c2.use_table('t')
assert c2.count_rows() == 50
c2.close()
def test_schema_survives_reopen(self):
with tempfile.TemporaryDirectory() as d:
c1 = ApexClient(dirpath=d)
c1.create_table('t', schema={'name': 'string', 'age': 'int64'})
c1.store([{'name': 'Alice', 'age': 25}])
c1.flush()
c1.close()
c2 = ApexClient(dirpath=d)
c2.use_table('t')
assert 'int' in c2.get_column_dtype('age').lower()
c2.close()
def test_delete_then_reopen_correct_count(self):
with tempfile.TemporaryDirectory() as d:
c1 = ApexClient(dirpath=d)
c1.create_table('t')
c1.store([{'x': i} for i in range(10)])
c1.execute("DELETE FROM t WHERE x > 4")
c1.flush()
c1.close()
c2 = ApexClient(dirpath=d)
c2.use_table('t')
assert c2.count_rows() == 5
c2.close()
def test_update_then_reopen_correct_values(self):
with tempfile.TemporaryDirectory() as d:
c1 = ApexClient(dirpath=d)
c1.create_table('t')
c1.store([{'x': 1, 'y': 'old'}])
c1.execute("UPDATE t SET y='new' WHERE x=1")
c1.flush()
c1.close()
c2 = ApexClient(dirpath=d)
c2.use_table('t')
assert c2.execute("SELECT y FROM t WHERE x=1").first()['y'] == 'new'
c2.close()
class TestErrorHandling:
def test_execute_after_close_raises(self):
with tempfile.TemporaryDirectory() as d:
c = ApexClient(dirpath=d)
c.create_table('t')
c.close()
with pytest.raises(RuntimeError):
c.execute("SELECT 1")
def test_store_without_table_raises(self):
with tempfile.TemporaryDirectory() as d:
c = ApexClient(dirpath=d)
with pytest.raises(RuntimeError):
c.store({'x': 1})
c.close()
def test_delete_no_args_raises(self):
with tempfile.TemporaryDirectory() as d:
c = ApexClient(dirpath=d)
c.create_table('t')
c.store([{'x': 1}])
with pytest.raises(ValueError):
c.delete()
c.close()
def test_bad_sql_raises(self):
with tempfile.TemporaryDirectory() as d:
c = ApexClient(dirpath=d)
c.create_table('t')
with pytest.raises(Exception):
c.execute("THIS IS NOT SQL AT ALL %%%")
c.close()
def test_select_nonexistent_table_raises(self):
with tempfile.TemporaryDirectory() as d:
c = ApexClient(dirpath=d)
with pytest.raises(Exception):
c.execute("SELECT * FROM no_such_table")
c.close()
def test_context_manager_closes_on_exception(self):
with tempfile.TemporaryDirectory() as d:
with pytest.raises(ValueError):
with ApexClient(dirpath=d) as c:
c.create_table('t')
raise ValueError("intentional")
assert c._is_closed
class TestCompression:
def test_get_compression_returns_string(self, tmp_client):
tmp_client.create_table('t')
assert isinstance(tmp_client.get_compression(), str)
def test_set_lz4_empty_table(self, tmp_client):
tmp_client.create_table('t')
assert isinstance(tmp_client.set_compression('lz4'), bool)
def test_set_zstd_empty_table(self, tmp_client):
tmp_client.create_table('t')
assert isinstance(tmp_client.set_compression('zstd'), bool)
def test_invalid_compression_raises(self, tmp_client):
tmp_client.create_table('t')
with pytest.raises(Exception):
tmp_client.set_compression('invalid_algo')
def test_lz4_data_roundtrip(self):
with tempfile.TemporaryDirectory() as d:
c = ApexClient(dirpath=d)
c.create_table('t')
c.set_compression('lz4')
c.store([{'x': i, 'y': f's{i}'} for i in range(100)])
c.flush()
assert c.count_rows() == 100
assert c.execute("SELECT SUM(x) as s FROM t").scalar() == sum(range(100))
c.close()
def test_compression_noop_on_nonempty_table(self):
with tempfile.TemporaryDirectory() as d:
c = ApexClient(dirpath=d)
c.create_table('t')
c.store([{'x': 1}])
result = c.set_compression('lz4')
assert result is False
c.close()
class TestDistinct:
def setup_method(self):
self.d = tempfile.mkdtemp()
self.c = ApexClient(dirpath=self.d)
self.c.execute("CREATE TABLE t (a STRING, b INT)")
self.c.execute("""INSERT INTO t VALUES
('x',1),('x',1),('x',2),('y',1),('y',1),('z',3)""")
def teardown_method(self):
self.c.close()
shutil.rmtree(self.d, ignore_errors=True)
def test_distinct_single_col(self):
rv = self.c.execute("SELECT DISTINCT a FROM t ORDER BY a")
assert [r['a'] for r in rv] == ['x', 'y', 'z']
def test_distinct_two_cols(self):
rv = self.c.execute("SELECT DISTINCT a, b FROM t ORDER BY a, b")
rows = [(r['a'], r['b']) for r in rv]
assert ('x', 1) in rows and ('x', 2) in rows
assert len([r for r in rows if r[0] == 'x']) == 2
def test_count_distinct(self):
assert self.c.execute("SELECT COUNT(DISTINCT a) as c FROM t").scalar() == 3
def test_count_distinct_with_filter(self):
assert self.c.execute("SELECT COUNT(DISTINCT b) as c FROM t WHERE a='x'").scalar() == 2
class TestOrderBy:
def setup_method(self):
self.d = tempfile.mkdtemp()
self.c = ApexClient(dirpath=self.d)
self.c.execute("CREATE TABLE t (a STRING, b INT, c FLOAT)")
self.c.execute("""INSERT INTO t VALUES
('z',3,1.1),('a',1,3.3),('m',2,2.2),('a',5,0.5),('m',NULL,1.5)""")
def teardown_method(self):
self.c.close()
shutil.rmtree(self.d, ignore_errors=True)
def test_order_by_string_asc(self):
rv = self.c.execute("SELECT a FROM t ORDER BY a ASC")
vals = [r['a'] for r in rv]
assert vals[0] == 'a' and vals[-1] == 'z'
def test_order_by_string_desc(self):
rv = self.c.execute("SELECT a FROM t ORDER BY a DESC")
assert rv.to_dict()[0]['a'] == 'z'
def test_order_by_two_cols(self):
rv = self.c.execute("SELECT a, b FROM t ORDER BY a ASC, b DESC")
rows = [(r['a'], r['b']) for r in rv]
a_rows = [r for r in rows if r[0] == 'a']
assert a_rows[0][1] == 5
def test_order_by_null_last_asc(self):
rv = self.c.execute("SELECT b FROM t ORDER BY b ASC")
vals = [r['b'] for r in rv]
non_null = [v for v in vals if v is not None]
assert non_null == sorted(non_null)
def test_order_by_float(self):
rv = self.c.execute("SELECT c FROM t ORDER BY c ASC")
vals = [r['c'] for r in rv]
non_null = [v for v in vals if v is not None]
assert non_null == sorted(non_null)