yamlbase 0.7.0

A lightweight SQL server that serves YAML-defined tables over standard SQL protocols
Documentation 
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

#!/usr/bin/env python3
"""
Test the exact queries mentioned in the feedback to verify our implementation.
"""

import psycopg2
import sys

def test_feedback_queries():
    """Test the exact queries mentioned in the user feedback."""
    print("Testing YamlBase Enhanced SQL Features...")
    
    try:
        # Connect to yamlbase
        conn = psycopg2.connect(
            host="localhost",
            port=5434,
            dbname="sql_features_test",
            user="admin",
            password="password123"
        )
        
        with conn.cursor() as cur:
            # Test 1: UPPER with string literal (reported as working)
            print("\n1. Testing UPPER with string literal:")
            cur.execute("SELECT UPPER('test') AS test_upper")
            result = cur.fetchone()
            print(f"   Result: {result[0]} (Expected: TEST)")
            assert result[0] == "TEST", "UPPER with literal failed"
            
            # Test 2: UPPER with column identifier (reported as failing)
            print("\n2. Testing UPPER with column identifier:")
            try:
                cur.execute("SELECT UPPER(FIRST_NAME) FROM EMPLOYEES")
                results = cur.fetchall()
                print(f"   Success! Found {len(results)} rows")
                for i, row in enumerate(results[:3]):  # Show first 3
                    print(f"   Row {i}: {row[0]}")
            except Exception as e:
                print(f"   FAILED: {e}")
                
            # Test 3: Complex JOIN (reported as failing)
            print("\n3. Testing complex JOIN:")
            try:
                cur.execute("""
                    SELECT p.PROJECT_ID, r.FIRST_NAME 
                    FROM PROJECTS p 
                    LEFT JOIN EMPLOYEES r ON p.MANAGER_ID = r.EMPLOYEE_ID
                """)
                results = cur.fetchall()
                print(f"   Success! Found {len(results)} rows")
                for i, row in enumerate(results[:3]):  # Show first 3
                    print(f"   Row {i}: Project {row[0]}, Manager: {row[1]}")
            except Exception as e:
                print(f"   FAILED: {e}")
                
            # Test 4: UPPER in JOIN condition (reported as failing)
            print("\n4. Testing UPPER in JOIN condition:")
            try:
                cur.execute("""
                    SELECT a.EMPLOYEE_ID, a.PROJECT_ROLE, r.FIRST_NAME, r.LAST_NAME
                    FROM PROJECT_ASSIGNMENTS a
                    LEFT JOIN EMPLOYEES r ON UPPER(a.EMPLOYEE_ID) = UPPER(r.EMPLOYEE_ID)
                    WHERE a.PROJECT_ID = 'PRJ001'
                """)
                results = cur.fetchall()
                print(f"   Success! Found {len(results)} rows")
                for row in results:
                    print(f"   Employee {row[0]}: {row[2]} {row[3]} ({row[1]})")
            except Exception as e:
                print(f"   FAILED: {e}")
                
            # Test 5: LOWER function
            print("\n5. Testing LOWER function:")
            try:
                cur.execute("SELECT EMPLOYEE_ID, LOWER(EMAIL) FROM EMPLOYEES WHERE DEPARTMENT = 'Engineering'")
                results = cur.fetchall()
                print(f"   Success! Found {len(results)} rows")
                for row in results[:2]:
                    print(f"   {row[0]}: {row[1]}")
            except Exception as e:
                print(f"   FAILED: {e}")
                
            # Test 6: TRIM function
            print("\n6. Testing TRIM function:")
            try:
                cur.execute("SELECT TRIM(FIRST_NAME), TRIM(LAST_NAME) FROM EMPLOYEES WHERE EMPLOYEE_ID = 'WBI001'")
                result = cur.fetchone()
                print(f"   Success! {result[0]} {result[1]}")
            except Exception as e:
                print(f"   FAILED: {e}")
                
            # Test 7: Nested functions
            print("\n7. Testing nested functions:")
            try:
                cur.execute("SELECT UPPER(TRIM(FIRST_NAME)) FROM EMPLOYEES WHERE EMPLOYEE_ID = 'WBI001'")
                result = cur.fetchone()
                print(f"   Success! {result[0]}")
            except Exception as e:
                print(f"   FAILED: {e}")
                
            # Test 8: CHAR type support
            print("\n8. Testing CHAR type support:")
            try:
                cur.execute("SELECT EMPLOYEE_ID, ACTIVE_FLAG FROM EMPLOYEES WHERE ACTIVE_FLAG = 'Y'")
                results = cur.fetchall()
                print(f"   Success! Found {len(results)} active employees")
            except Exception as e:
                print(f"   FAILED: {e}")
                
        conn.close()
        print("\n✅ Test completed!")
        
    except psycopg2.OperationalError as e:
        print(f"❌ Connection failed: {e}")
        print("\nMake sure YamlBase is running with:")
        print("cargo run -- -f examples/sql_features_test.yaml --protocol postgres -p 5434")
        sys.exit(1)

if __name__ == "__main__":
    test_feedback_queries()