Spaces:
Runtime error
Runtime error
File size: 4,401 Bytes
b339b93 | 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 | """Tests for schema definitions."""
from duckdb_loader.schema import (
ALL_COLUMNS,
CONTRIBUTIONS_COLUMNS,
_get_column_type,
create_indexes,
create_schema,
)
class TestColumnDefinitions:
"""Tests for column definitions."""
def test_contributions_columns_not_empty(self):
"""Default columns list is not empty."""
assert len(CONTRIBUTIONS_COLUMNS) > 0
def test_all_columns_superset(self):
"""ALL_COLUMNS contains all CONTRIBUTIONS_COLUMNS."""
for col in CONTRIBUTIONS_COLUMNS:
assert col in ALL_COLUMNS, f"Missing column: {col}"
def test_all_columns_count(self):
"""ALL_COLUMNS has reasonable count (sanity check)."""
assert len(ALL_COLUMNS) >= len(CONTRIBUTIONS_COLUMNS)
assert len(ALL_COLUMNS) >= 40 # Sanity check for catastrophic deletion
def test_core_columns_present(self):
"""Core columns are in default set."""
core = ["cycle", "amount", "date", "contributor.name", "recipient.name"]
for col in core:
assert col in CONTRIBUTIONS_COLUMNS
class TestColumnTypes:
"""Tests for column type mapping."""
def test_integer_columns(self):
"""Integer columns map correctly."""
assert _get_column_type("cycle") == "INTEGER"
def test_float_columns(self):
"""Float columns map correctly."""
assert _get_column_type("amount") == "DOUBLE"
assert _get_column_type("contributor.cfscore") == "DOUBLE"
assert _get_column_type("latitude") == "DOUBLE"
def test_string_columns(self):
"""String columns default to VARCHAR."""
assert _get_column_type("contributor.name") == "VARCHAR"
assert _get_column_type("unknown.column") == "VARCHAR"
class TestCreateSchema:
"""Tests for schema creation."""
def test_create_schema_default_columns(self, temp_duckdb_path):
"""Creates schema with default columns."""
import duckdb
conn = duckdb.connect(str(temp_duckdb_path))
create_schema(conn)
# Verify table exists
result = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='contributions'"
).fetchone()
assert result is not None
# Verify columns
cols = conn.execute("PRAGMA table_info(contributions)").fetchall()
col_names = [c[1] for c in cols]
assert "cycle" in col_names
assert "amount" in col_names
conn.close()
def test_create_schema_custom_columns(self, temp_duckdb_path):
"""Creates schema with custom columns."""
import duckdb
conn = duckdb.connect(str(temp_duckdb_path))
custom_cols = ["cycle", "amount", "contributor.name"]
create_schema(conn, columns=custom_cols)
cols = conn.execute("PRAGMA table_info(contributions)").fetchall()
col_names = [c[1] for c in cols]
assert len(col_names) == 3
assert "cycle" in col_names
conn.close()
def test_create_schema_custom_table_name(self, temp_duckdb_path):
"""Creates schema with custom table name."""
import duckdb
conn = duckdb.connect(str(temp_duckdb_path))
create_schema(conn, table_name="my_contributions")
result = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='my_contributions'"
).fetchone()
assert result is not None
conn.close()
class TestCreateIndexes:
"""Tests for index creation."""
def test_create_indexes(self, temp_duckdb_path):
"""Creates indexes on table."""
import duckdb
conn = duckdb.connect(str(temp_duckdb_path))
create_schema(conn)
create_indexes(conn)
# DuckDB stores index info differently, just verify no errors
conn.close()
def test_create_indexes_missing_columns(self, temp_duckdb_path):
"""Index creation handles missing columns gracefully."""
import duckdb
conn = duckdb.connect(str(temp_duckdb_path))
# Create with minimal columns
create_schema(conn, columns=["cycle", "amount"])
# Should not raise despite missing indexed columns
create_indexes(conn)
conn.close()
|