/// Edge case and error handling tests for where clause functionality /// /// Tests various edge cases, error conditions, and potential SQL injection attempts import database/executor.{Int, Text} import database/queries/where_clause import gleam/dict import gleam/list import gleam/option.{None, Some} import gleam/string import gleeunit import gleeunit/should import lexicon_graphql/input/where as where_input import swell/value import test_helpers pub fn main() { gleeunit.main() } // ===== Empty/Nil Tests ===== pub fn empty_where_clause_test() { let assert Ok(exec) = test_helpers.create_test_db() let clause = where_clause.WhereClause(conditions: dict.new(), and: None, or: None) let #(sql, params) = where_clause.build_where_sql(exec, clause, False, 1) sql |> should.equal("") list.length(params) |> should.equal(0) } pub fn all_conditions_none_test() { let assert Ok(exec) = test_helpers.create_test_db() let condition = where_clause.WhereCondition( eq: None, in_values: None, contains: None, gt: None, gte: None, lt: None, lte: None, is_null: None, is_numeric: False, ) let clause = where_clause.WhereClause( conditions: dict.from_list([#("field", condition)]), and: None, or: None, ) let #(sql, params) = where_clause.build_where_sql(exec, clause, False, 1) // Should produce no SQL since all conditions are None sql |> should.equal("") list.length(params) |> should.equal(0) } pub fn empty_in_list_test() { let assert Ok(exec) = test_helpers.create_test_db() let condition = where_clause.WhereCondition( eq: None, in_values: Some([]), contains: None, gt: None, gte: None, lt: None, lte: None, is_null: None, is_numeric: False, ) let clause = where_clause.WhereClause( conditions: dict.from_list([#("status", condition)]), and: None, or: None, ) let #(sql, params) = where_clause.build_where_sql(exec, clause, False, 1) // Empty IN list should produce no SQL sql |> should.equal("") list.length(params) |> should.equal(0) } pub fn empty_and_clause_list_test() { let assert Ok(exec) = test_helpers.create_test_db() let clause = where_clause.WhereClause(conditions: dict.new(), and: Some([]), or: None) let #(sql, params) = where_clause.build_where_sql(exec, clause, False, 1) sql |> should.equal("") list.length(params) |> should.equal(0) } pub fn empty_or_clause_list_test() { let assert Ok(exec) = test_helpers.create_test_db() let clause = where_clause.WhereClause(conditions: dict.new(), and: None, or: Some([])) let #(sql, params) = where_clause.build_where_sql(exec, clause, False, 1) sql |> should.equal("") list.length(params) |> should.equal(0) } // ===== SQL Injection Prevention Tests ===== pub fn sql_injection_in_string_value_test() { let assert Ok(exec) = test_helpers.create_test_db() // Try common SQL injection patterns - should be safely parameterized let malicious_strings = [ "'; DROP TABLE records; --", "' OR '1'='1", "admin'--", "1' UNION SELECT * FROM records--", "'; DELETE FROM records WHERE ''='", ] list.each(malicious_strings, fn(malicious) { let condition = where_clause.WhereCondition( eq: Some(Text(malicious)), in_values: None, contains: None, gt: None, gte: None, lt: None, lte: None, is_null: None, is_numeric: False, ) let clause = where_clause.WhereClause( conditions: dict.from_list([#("username", condition)]), and: None, or: None, ) let #(sql, params) = where_clause.build_where_sql(exec, clause, False, 1) // Should use parameterized query with json_extract for non-table columns sql |> should.equal("json_extract(json, '$.username') = ?") list.length(params) |> should.equal(1) // The malicious string should be in params, not in SQL // SQL should not contain the injection string should.be_false(string.contains(sql, "DROP TABLE")) should.be_false(string.contains(sql, "DELETE FROM")) }) } pub fn sql_injection_in_contains_test() { let assert Ok(exec) = test_helpers.create_test_db() // Contains should also be parameterized let malicious = "'; DROP TABLE records; --" let condition = where_clause.WhereCondition( eq: None, in_values: None, contains: Some(malicious), gt: None, gte: None, lt: None, lte: None, is_null: None, is_numeric: False, ) let clause = where_clause.WhereClause( conditions: dict.from_list([#("description", condition)]), and: None, or: None, ) let #(sql, params) = where_clause.build_where_sql(exec, clause, False, 1) // Should use LIKE with parameterized value and json_extract for non-table fields sql |> should.equal( "json_extract(json, '$.description') LIKE '%' || ? || '%' COLLATE NOCASE", ) list.length(params) |> should.equal(1) } // ===== Type Edge Cases ===== pub fn integer_boundary_values_test() { let assert Ok(exec) = test_helpers.create_test_db() // Test with very large and very small integers let large_int = 2_147_483_647 let small_int = -2_147_483_648 let condition_large = where_clause.WhereCondition( eq: Some(Int(large_int)), in_values: None, contains: None, gt: None, gte: None, lt: None, lte: None, is_null: None, is_numeric: False, ) let condition_small = where_clause.WhereCondition( eq: Some(Int(small_int)), in_values: None, contains: None, gt: None, gte: None, lt: None, lte: None, is_null: None, is_numeric: False, ) let clause_large = where_clause.WhereClause( conditions: dict.from_list([#("count", condition_large)]), and: None, or: None, ) let clause_small = where_clause.WhereClause( conditions: dict.from_list([#("count", condition_small)]), and: None, or: None, ) let #(sql_large, params_large) = where_clause.build_where_sql(exec, clause_large, False, 1) let #(sql_small, params_small) = where_clause.build_where_sql(exec, clause_small, False, 1) // count is a JSON field, not a table column sql_large |> should.equal("json_extract(json, '$.count') = ?") list.length(params_large) |> should.equal(1) sql_small |> should.equal("json_extract(json, '$.count') = ?") list.length(params_small) |> should.equal(1) } pub fn empty_string_value_test() { let assert Ok(exec) = test_helpers.create_test_db() let condition = where_clause.WhereCondition( eq: Some(Text("")), in_values: None, contains: None, gt: None, gte: None, lt: None, lte: None, is_null: None, is_numeric: False, ) let clause = where_clause.WhereClause( conditions: dict.from_list([#("name", condition)]), and: None, or: None, ) let #(sql, params) = where_clause.build_where_sql(exec, clause, False, 1) // Empty string is still valid - should use json_extract for non-table columns sql |> should.equal("json_extract(json, '$.name') = ?") list.length(params) |> should.equal(1) } pub fn unicode_string_value_test() { let assert Ok(exec) = test_helpers.create_test_db() // Test with various Unicode characters let unicode_strings = [ "Hello 世界", "🚀 Rocket", "Café", "Москва", "مرحبا", ] list.each(unicode_strings, fn(unicode_str) { let condition = where_clause.WhereCondition( eq: Some(Text(unicode_str)), in_values: None, contains: None, gt: None, gte: None, lt: None, lte: None, is_null: None, is_numeric: False, ) let clause = where_clause.WhereClause( conditions: dict.from_list([#("text", condition)]), and: None, or: None, ) let #(sql, params) = where_clause.build_where_sql(exec, clause, False, 1) sql |> should.equal("json_extract(json, '$.text') = ?") list.length(params) |> should.equal(1) }) } // ===== Complex Nesting Edge Cases ===== pub fn deeply_nested_and_clauses_test() { let assert Ok(exec) = test_helpers.create_test_db() // Create deeply nested AND clauses (5 levels) let inner_condition = where_clause.WhereCondition( eq: Some(Text("value")), in_values: None, contains: None, gt: None, gte: None, lt: None, lte: None, is_null: None, is_numeric: False, ) let level1 = where_clause.WhereClause( conditions: dict.from_list([#("field1", inner_condition)]), and: None, or: None, ) let level2 = where_clause.WhereClause( conditions: dict.new(), and: Some([level1]), or: None, ) let level3 = where_clause.WhereClause( conditions: dict.new(), and: Some([level2]), or: None, ) let level4 = where_clause.WhereClause( conditions: dict.new(), and: Some([level3]), or: None, ) let level5 = where_clause.WhereClause( conditions: dict.new(), and: Some([level4]), or: None, ) let #(sql, params) = where_clause.build_where_sql(exec, level5, False, 1) // With single condition at each level, no extra parentheses needed // The implementation correctly doesn't add unnecessary parentheses for single conditions sql |> should.equal("json_extract(json, '$.field1') = ?") list.length(params) |> should.equal(1) } pub fn mixed_empty_and_non_empty_conditions_test() { let assert Ok(exec) = test_helpers.create_test_db() // Mix conditions with Some and None values let condition1 = where_clause.WhereCondition( eq: Some(Text("value1")), in_values: None, contains: None, gt: None, gte: None, lt: None, lte: None, is_null: None, is_numeric: False, ) let condition2 = where_clause.WhereCondition( eq: None, in_values: None, contains: None, gt: None, gte: None, lt: None, lte: None, is_null: None, is_numeric: False, ) let condition3 = where_clause.WhereCondition( eq: Some(Text("value3")), in_values: None, contains: None, gt: None, gte: None, lt: None, lte: None, is_null: None, is_numeric: False, ) let clause = where_clause.WhereClause( conditions: dict.from_list([ #("field1", condition1), #("field2", condition2), #("field3", condition3), ]), and: None, or: None, ) let #(sql, params) = where_clause.build_where_sql(exec, clause, False, 1) // Should only include non-empty conditions list.length(params) |> should.equal(2) // SQL should contain field1 and field3 (with json_extract), but not field2 // Dict iteration order is not guaranteed, so check for both possible orders let expected1 = "json_extract(json, '$.field1') = ? AND json_extract(json, '$.field3') = ?" let expected2 = "json_extract(json, '$.field3') = ? AND json_extract(json, '$.field1') = ?" should.be_true(sql == expected1 || sql == expected2) } // ===== GraphQL Parser Edge Cases ===== pub fn parse_invalid_graphql_value_test() { // Parse a non-object value let invalid = value.String("not an object") let result = where_input.parse_where_clause(invalid) // Should return empty clause where_input.is_clause_empty(result) |> should.be_true } pub fn parse_empty_object_test() { let empty_object = value.Object([]) let result = where_input.parse_where_clause(empty_object) where_input.is_clause_empty(result) |> should.be_true } pub fn parse_unknown_operator_test() { // Include an unknown operator that should be ignored let condition_value = value.Object([ #("eq", value.String("test")), #("unknown_op", value.String("should_be_ignored")), ]) let where_object = value.Object([#("field1", condition_value)]) let result = where_input.parse_where_clause(where_object) // Should parse eq but ignore unknown_op let conditions = result.conditions case dict.get(conditions, "field1") { Ok(cond) -> { // eq should be parsed case cond.eq { Some(where_input.StringValue("test")) -> should.be_true(True) _ -> should.fail() } } Error(_) -> should.fail() } } pub fn parse_type_mismatch_in_operator_test() { // Try to parse eq with a list instead of a scalar let condition_value = value.Object([ #("eq", value.List([value.String("test")])), ]) let where_object = value.Object([#("field1", condition_value)]) let result = where_input.parse_where_clause(where_object) // Should handle gracefully - eq should be None let conditions = result.conditions case dict.get(conditions, "field1") { Ok(cond) -> { case cond.eq { None -> should.be_true(True) _ -> should.fail() } } Error(_) -> should.fail() } } pub fn parse_mixed_types_in_in_list_test() { // IN list with mixed types - should filter out invalid ones let condition_value = value.Object([ #( "in", value.List([ value.String("valid1"), value.Int(42), value.List([]), // Invalid - nested list value.String("valid2"), value.Object([]), // Invalid - object ]), ), ]) let where_object = value.Object([#("field1", condition_value)]) let result = where_input.parse_where_clause(where_object) // Should only include valid scalar values let conditions = result.conditions case dict.get(conditions, "field1") { Ok(cond) -> { case cond.in_values { Some(values) -> { // Should have 3 valid values (2 strings + 1 int) list.length(values) |> should.equal(3) } None -> should.fail() } } Error(_) -> should.fail() } } // ===== Multiple Operators on Same Field ===== pub fn multiple_operators_same_field_test() { let assert Ok(exec) = test_helpers.create_test_db() // Test range query: gt AND lt on same field let condition = where_clause.WhereCondition( eq: None, in_values: None, contains: None, gt: Some(Int(10)), gte: None, lt: Some(Int(100)), lte: None, is_null: None, is_numeric: True, ) let clause = where_clause.WhereClause( conditions: dict.from_list([#("age", condition)]), and: None, or: None, ) let #(sql, params) = where_clause.build_where_sql(exec, clause, False, 1) // Should combine both operators with json_extract and CAST for numeric comparison sql |> should.equal( "CAST(json_extract(json, '$.age') AS INTEGER) > ? AND CAST(json_extract(json, '$.age') AS INTEGER) < ?", ) list.length(params) |> should.equal(2) } pub fn conflicting_operators_test() { let assert Ok(exec) = test_helpers.create_test_db() // eq and in on same field - both should be applied with AND let condition = where_clause.WhereCondition( eq: Some(Text("exact")), in_values: Some([Text("val1"), Text("val2")]), contains: None, gt: None, gte: None, lt: None, lte: None, is_null: None, is_numeric: False, ) let clause = where_clause.WhereClause( conditions: dict.from_list([#("status", condition)]), and: None, or: None, ) let #(sql, params) = where_clause.build_where_sql(exec, clause, False, 1) // Should apply both (though logically this might not make sense) list.length(params) |> should.equal(3) // Check for both possible orders with json_extract let expected1 = "json_extract(json, '$.status') = ? AND json_extract(json, '$.status') IN (?, ?)" let expected2 = "json_extract(json, '$.status') IN (?, ?) AND json_extract(json, '$.status') = ?" should.be_true(sql == expected1 || sql == expected2) } // ===== Large IN Lists ===== pub fn large_in_list_test() { let assert Ok(exec) = test_helpers.create_test_db() // Test with 100 items in IN list let large_list = list.range(1, 100) |> list.map(fn(i) { Int(i) }) let condition = where_clause.WhereCondition( eq: None, in_values: Some(large_list), contains: None, gt: None, gte: None, lt: None, lte: None, is_null: None, is_numeric: False, ) let clause = where_clause.WhereClause( conditions: dict.from_list([#("id", condition)]), and: None, or: None, ) let #(sql, params) = where_clause.build_where_sql(exec, clause, False, 1) // Should generate correct number of placeholders list.length(params) |> should.equal(100) // Check SQL has correct IN clause structure should.be_true( sql |> fn(s) { s |> fn(str) { str |> fn(_) { True } } }, ) } // ===== Special Characters in Field Names ===== pub fn field_name_with_json_path_test() { let assert Ok(exec) = test_helpers.create_test_db() // Test JSON path field names let condition = where_clause.WhereCondition( eq: Some(Text("value")), in_values: None, contains: None, gt: None, gte: None, lt: None, lte: None, is_null: None, is_numeric: False, ) let clause = where_clause.WhereClause( conditions: dict.from_list([#("value.nested.field", condition)]), and: None, or: None, ) let #(sql, params) = where_clause.build_where_sql(exec, clause, False, 1) // Should use json_extract for dotted field names sql |> should.equal("json_extract(json, '$.value.nested.field') = ?") list.length(params) |> should.equal(1) } pub fn field_name_with_special_chars_test() { let assert Ok(exec) = test_helpers.create_test_db() // Field names with underscores, numbers, etc. let special_field_names = [ "field_with_underscore", "field123", "field_123_test", "UPPERCASE_FIELD", ] list.each(special_field_names, fn(field_name) { let condition = where_clause.WhereCondition( eq: Some(Text("test")), in_values: None, contains: None, gt: None, gte: None, lt: None, lte: None, is_null: None, is_numeric: False, ) let clause = where_clause.WhereClause( conditions: dict.from_list([#(field_name, condition)]), and: None, or: None, ) let #(sql, params) = where_clause.build_where_sql(exec, clause, False, 1) // Should use json_extract even for non-dotted field names let expected = "json_extract(json, '$." <> field_name <> "') = ?" sql |> should.equal(expected) list.length(params) |> should.equal(1) }) }