test: add comprehensive test coverage for extracted backend crates

Add 296 tests across unit and integration test suites to cover
the newly extracted crates from the recent refactor commits.

Unit tests (270):
- windmill-trigger-postgres (96): hex codec, bool parsing, type
  conversion, relation tracking, replication message parsing,
  publication data validation
- windmill-trigger-http (92): HMAC signature verification for
  GitHub/Slack/Stripe/TikTok/Twitch/Zoom webhooks, API key auth,
  Basic Auth, route validation, HTTP method/request type serde
- windmill-api-jobs (39): SQL query builder for job listing/counting
  with filters, pagination, label handling
- windmill-trigger (31): TriggerMode serde, query pagination,
  BaseTriggerData backward compat, HandlerAction, ServerState
- windmill-common webhook (7): WebhookMessage serialization tags
- worker nativets/postgresql (5): nativets job execution with
  args/objects/datetime, postgresql query execution

Integration tests (26):
- backend/tests/triggers.rs: capture config CRUD, capture payload
  operations, capture API endpoints, HTTP trigger CRUD with mode
  filtering, all trigger types DB schema validation (websocket,
  kafka, postgres, nats, sqs), schedule operations

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

backend/windmill-trigger-postgres/src/bool.rs

@@ -22,3 +22,36 @@ pub fn parse_bool(s: &str) -> Result<bool, ParseBoolError> {
         _ => Err(ParseBoolError::InvalidInput(s.to_string())),
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_parse_true() {
+        assert_eq!(parse_bool("t").unwrap(), true);
+    }
+
+    #[test]
+    fn test_parse_false() {
+        assert_eq!(parse_bool("f").unwrap(), false);
+    }
+
+    #[test]
+    fn test_invalid_true_string() {
+        assert!(matches!(
+            parse_bool("true"),
+            Err(ParseBoolError::InvalidInput(s)) if s == "true"
+        ));
+    }
+
+    #[test]
+    fn test_invalid_empty() {
+        assert!(parse_bool("").is_err());
+    }
+
+    #[test]
+    fn test_invalid_uppercase() {
+        assert!(parse_bool("T").is_err());
+    }
+}

backend/windmill-trigger-postgres/src/converter.rs

@@ -253,3 +253,244 @@ impl Converter {
         Ok(arr)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    // --- Scalar type conversions ---
+
+    #[test]
+    fn test_bool_true() {
+        let result = Converter::try_from_str(Some(Type::BOOL), "t").unwrap();
+        assert_eq!(result, Value::Bool(true));
+    }
+
+    #[test]
+    fn test_bool_false() {
+        let result = Converter::try_from_str(Some(Type::BOOL), "f").unwrap();
+        assert_eq!(result, Value::Bool(false));
+    }
+
+    #[test]
+    fn test_text() {
+        let result = Converter::try_from_str(Some(Type::TEXT), "hello world").unwrap();
+        assert_eq!(result, Value::String("hello world".to_string()));
+    }
+
+    #[test]
+    fn test_varchar() {
+        let result = Converter::try_from_str(Some(Type::VARCHAR), "test").unwrap();
+        assert_eq!(result, Value::String("test".to_string()));
+    }
+
+    #[test]
+    fn test_int2() {
+        let result = Converter::try_from_str(Some(Type::INT2), "42").unwrap();
+        assert_eq!(result, json!(42));
+    }
+
+    #[test]
+    fn test_int4() {
+        let result = Converter::try_from_str(Some(Type::INT4), "-100").unwrap();
+        assert_eq!(result, json!(-100));
+    }
+
+    #[test]
+    fn test_int8() {
+        let result = Converter::try_from_str(Some(Type::INT8), "9999999999").unwrap();
+        assert_eq!(result, json!(9999999999i64));
+    }
+
+    #[test]
+    fn test_float4() {
+        let result = Converter::try_from_str(Some(Type::FLOAT4), "3.14").unwrap();
+        assert!(result.as_f64().unwrap() - 3.14 < 0.001);
+    }
+
+    #[test]
+    fn test_float8() {
+        let result = Converter::try_from_str(Some(Type::FLOAT8), "2.718281828").unwrap();
+        assert!(result.as_f64().unwrap() - 2.718281828 < 0.0001);
+    }
+
+    #[test]
+    fn test_numeric() {
+        let result = Converter::try_from_str(Some(Type::NUMERIC), "123.456").unwrap();
+        assert_eq!(result.to_string(), "123.456");
+    }
+
+    #[test]
+    fn test_uuid() {
+        let result =
+            Converter::try_from_str(Some(Type::UUID), "550e8400-e29b-41d4-a716-446655440000")
+                .unwrap();
+        assert_eq!(
+            result,
+            Value::String("550e8400-e29b-41d4-a716-446655440000".to_string())
+        );
+    }
+
+    #[test]
+    fn test_json() {
+        let result =
+            Converter::try_from_str(Some(Type::JSON), r#"{"key": "value", "n": 1}"#).unwrap();
+        assert_eq!(result, json!({"key": "value", "n": 1}));
+    }
+
+    #[test]
+    fn test_jsonb() {
+        let result = Converter::try_from_str(Some(Type::JSONB), r#"[1,2,3]"#).unwrap();
+        assert_eq!(result, json!([1, 2, 3]));
+    }
+
+    #[test]
+    fn test_date() {
+        let result = Converter::try_from_str(Some(Type::DATE), "2024-01-15").unwrap();
+        assert_eq!(result, Value::String("2024-01-15".to_string()));
+    }
+
+    #[test]
+    fn test_time() {
+        let result = Converter::try_from_str(Some(Type::TIME), "14:30:00.0").unwrap();
+        assert_eq!(result, Value::String("14:30:00".to_string()));
+    }
+
+    #[test]
+    fn test_timestamp() {
+        let result =
+            Converter::try_from_str(Some(Type::TIMESTAMP), "2024-01-15 14:30:00.0").unwrap();
+        assert_eq!(
+            result,
+            Value::String("2024-01-15 14:30:00".to_string())
+        );
+    }
+
+    #[test]
+    fn test_timestamptz() {
+        let result = Converter::try_from_str(
+            Some(Type::TIMESTAMPTZ),
+            "2024-01-15 14:30:00.0+00",
+        )
+        .unwrap();
+        assert!(result.as_str().unwrap().contains("2024-01-15"));
+    }
+
+    #[test]
+    fn test_bytea() {
+        let result = Converter::try_from_str(Some(Type::BYTEA), "\\x48656c6c6f").unwrap();
+        assert_eq!(result, json!([72, 101, 108, 108, 111]));
+    }
+
+    #[test]
+    fn test_oid() {
+        let result = Converter::try_from_str(Some(Type::OID), "12345").unwrap();
+        assert_eq!(result, json!(12345u32));
+    }
+
+    #[test]
+    fn test_none_type_defaults_to_text() {
+        let result = Converter::try_from_str(None, "anything").unwrap();
+        assert_eq!(result, Value::String("anything".to_string()));
+    }
+
+    // --- Array type conversions ---
+
+    #[test]
+    fn test_int4_array() {
+        let result = Converter::try_from_str(Some(Type::INT4_ARRAY), "{1,2,3}").unwrap();
+        assert_eq!(result, json!([1, 2, 3]));
+    }
+
+    #[test]
+    fn test_text_array() {
+        let result = Converter::try_from_str(Some(Type::TEXT_ARRAY), "{hello,world}").unwrap();
+        assert_eq!(result, json!(["hello", "world"]));
+    }
+
+    #[test]
+    fn test_bool_array() {
+        let result = Converter::try_from_str(Some(Type::BOOL_ARRAY), "{t,f,t}").unwrap();
+        assert_eq!(result, json!([true, false, true]));
+    }
+
+    #[test]
+    fn test_array_with_null() {
+        let result = Converter::try_from_str(Some(Type::INT4_ARRAY), "{1,NULL,3}").unwrap();
+        assert_eq!(result, json!([1, null, 3]));
+    }
+
+    #[test]
+    fn test_array_with_quoted_strings() {
+        let result =
+            Converter::try_from_str(Some(Type::TEXT_ARRAY), r#"{"hello, world","test"}"#).unwrap();
+        assert_eq!(result, json!(["hello, world", "test"]));
+    }
+
+    #[test]
+    fn test_empty_array() {
+        let result = Converter::try_from_str(Some(Type::INT4_ARRAY), "{}").unwrap();
+        assert_eq!(result, json!([]));
+    }
+
+    #[test]
+    fn test_uuid_array() {
+        let result = Converter::try_from_str(
+            Some(Type::UUID_ARRAY),
+            "{550e8400-e29b-41d4-a716-446655440000,6ba7b810-9dad-11d1-80b4-00c04fd430c8}",
+        )
+        .unwrap();
+        assert_eq!(
+            result,
+            json!([
+                "550e8400-e29b-41d4-a716-446655440000",
+                "6ba7b810-9dad-11d1-80b4-00c04fd430c8"
+            ])
+        );
+    }
+
+    // --- Error cases ---
+
+    #[test]
+    fn test_invalid_int() {
+        assert!(Converter::try_from_str(Some(Type::INT4), "not_a_number").is_err());
+    }
+
+    #[test]
+    fn test_invalid_bool() {
+        assert!(Converter::try_from_str(Some(Type::BOOL), "yes").is_err());
+    }
+
+    #[test]
+    fn test_invalid_uuid() {
+        assert!(Converter::try_from_str(Some(Type::UUID), "not-a-uuid").is_err());
+    }
+
+    #[test]
+    fn test_invalid_json() {
+        assert!(Converter::try_from_str(Some(Type::JSON), "not json").is_err());
+    }
+
+    #[test]
+    fn test_array_missing_braces() {
+        assert!(Converter::try_from_str(Some(Type::INT4_ARRAY), "1,2,3").is_err());
+    }
+
+    #[test]
+    fn test_array_too_short() {
+        assert!(Converter::try_from_str(Some(Type::INT4_ARRAY), "{").is_err());
+    }
+
+    #[test]
+    fn test_array_with_escaped_backslash() {
+        let result =
+            Converter::try_from_str(Some(Type::TEXT_ARRAY), r#"{"a\\b","c"}"#).unwrap();
+        assert_eq!(result, json!(["a\\b", "c"]));
+    }
+
+    #[test]
+    fn test_float_nan_rejected() {
+        assert!(Converter::try_from_str(Some(Type::FLOAT4), "NaN").is_err());
+    }
+}

backend/windmill-trigger-postgres/src/hex.rs

@@ -41,3 +41,86 @@ pub fn from_bytea_hex(s: &str) -> Result<Vec<u8>, ByteaHexParseError> {
 
     Ok(result)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_valid_bytea_hex() {
+        assert_eq!(from_bytea_hex("\\x48656c6c6f").unwrap(), b"Hello");
+    }
+
+    #[test]
+    fn test_empty_bytea_hex() {
+        assert_eq!(from_bytea_hex("\\x").unwrap(), Vec::<u8>::new());
+    }
+
+    #[test]
+    fn test_single_byte() {
+        assert_eq!(from_bytea_hex("\\xff").unwrap(), vec![0xff]);
+    }
+
+    #[test]
+    fn test_all_zeros() {
+        assert_eq!(from_bytea_hex("\\x000000").unwrap(), vec![0, 0, 0]);
+    }
+
+    #[test]
+    fn test_missing_prefix() {
+        assert!(matches!(
+            from_bytea_hex("48656c6c6f"),
+            Err(ByteaHexParseError::InvalidPrefix)
+        ));
+    }
+
+    #[test]
+    fn test_wrong_prefix() {
+        assert!(matches!(
+            from_bytea_hex("0x48656c6c6f"),
+            Err(ByteaHexParseError::InvalidPrefix)
+        ));
+    }
+
+    #[test]
+    fn test_too_short() {
+        assert!(matches!(
+            from_bytea_hex("\\"),
+            Err(ByteaHexParseError::InvalidPrefix)
+        ));
+    }
+
+    #[test]
+    fn test_empty_string() {
+        assert!(matches!(
+            from_bytea_hex(""),
+            Err(ByteaHexParseError::InvalidPrefix)
+        ));
+    }
+
+    #[test]
+    fn test_odd_digits() {
+        assert!(matches!(
+            from_bytea_hex("\\xabc"),
+            Err(ByteaHexParseError::OddNumerOfDigits)
+        ));
+    }
+
+    #[test]
+    fn test_invalid_hex_chars() {
+        assert!(matches!(
+            from_bytea_hex("\\xzz"),
+            Err(ByteaHexParseError::ParseInt(_))
+        ));
+    }
+
+    #[test]
+    fn test_uppercase_hex() {
+        assert_eq!(from_bytea_hex("\\xABCD").unwrap(), vec![0xab, 0xcd]);
+    }
+
+    #[test]
+    fn test_mixed_case_hex() {
+        assert_eq!(from_bytea_hex("\\xAbCd").unwrap(), vec![0xab, 0xcd]);
+    }
+}

backend/windmill-trigger-postgres/src/lib.rs

@@ -550,3 +550,163 @@ pub fn generate_random_string() -> String {
 
     format!("{}_{}", timestamp, random_part)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_valid_publication_data() {
+        let json = r#"{
+            "table_to_track": [
+                {
+                    "schema_name": "public",
+                    "table_to_track": [
+                        {"table_name": "users"}
+                    ]
+                }
+            ],
+            "transaction_to_track": ["insert", "update"]
+        }"#;
+        let result: std::result::Result<PublicationData, _> = serde_json::from_str(json);
+        assert!(result.is_ok());
+    }
+
+    #[test]
+    fn test_publication_data_empty_schema_name() {
+        let json = r#"{
+            "table_to_track": [
+                {
+                    "schema_name": "",
+                    "table_to_track": []
+                }
+            ],
+            "transaction_to_track": ["insert"]
+        }"#;
+        let result: std::result::Result<PublicationData, _> = serde_json::from_str(json);
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn test_publication_data_empty_table_name() {
+        let json = r#"{
+            "table_to_track": [
+                {
+                    "schema_name": "public",
+                    "table_to_track": [
+                        {"table_name": "  "}
+                    ]
+                }
+            ],
+            "transaction_to_track": ["insert"]
+        }"#;
+        let result: std::result::Result<PublicationData, _> = serde_json::from_str(json);
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn test_publication_data_invalid_transaction_type() {
+        let json = r#"{
+            "transaction_to_track": ["insert", "truncate"]
+        }"#;
+        let result: std::result::Result<PublicationData, _> = serde_json::from_str(json);
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn test_publication_data_too_many_transaction_types() {
+        let json = r#"{
+            "transaction_to_track": ["insert", "update", "delete", "insert"]
+        }"#;
+        let result: std::result::Result<PublicationData, _> = serde_json::from_str(json);
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn test_publication_data_duplicate_schema_names() {
+        let json = r#"{
+            "table_to_track": [
+                {"schema_name": "public", "table_to_track": [{"table_name": "a"}]},
+                {"schema_name": "public", "table_to_track": [{"table_name": "b"}]}
+            ],
+            "transaction_to_track": ["insert"]
+        }"#;
+        let result: std::result::Result<PublicationData, _> = serde_json::from_str(json);
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn test_publication_data_all_tables_in_schema() {
+        let json = r#"{
+            "table_to_track": [
+                {"schema_name": "public", "table_to_track": []}
+            ],
+            "transaction_to_track": ["insert"]
+        }"#;
+        let result: std::result::Result<PublicationData, _> = serde_json::from_str(json);
+        assert!(result.is_ok());
+    }
+
+    #[test]
+    fn test_publication_data_incompatible_tracking() {
+        let json = r#"{
+            "table_to_track": [
+                {"schema_name": "schema1", "table_to_track": []},
+                {"schema_name": "schema2", "table_to_track": [{"table_name": "t1", "columns_name": ["col1"]}]}
+            ],
+            "transaction_to_track": ["insert"]
+        }"#;
+        let result: std::result::Result<PublicationData, _> = serde_json::from_str(json);
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn test_postgres_config_serialization() {
+        let config = PostgresConfig {
+            postgres_resource_path: "f/db/postgres".to_string(),
+            replication_slot_name: "slot_1".to_string(),
+            publication_name: "pub_1".to_string(),
+            basic_mode: Some(false),
+        };
+        let json = serde_json::to_value(&config).unwrap();
+        assert_eq!(json["postgres_resource_path"], "f/db/postgres");
+        assert_eq!(json["replication_slot_name"], "slot_1");
+    }
+
+    #[test]
+    fn test_generate_random_string_format() {
+        let s = generate_random_string();
+        assert!(s.contains('_'));
+        let parts: Vec<&str> = s.split('_').collect();
+        assert_eq!(parts.len(), 2);
+        assert_eq!(parts[1].len(), 10);
+    }
+
+    #[test]
+    fn test_relations_add_table() {
+        let mut rel = Relations::new("public".to_string(), vec![]);
+        rel.add_new_table(TableToTrack::new("users".to_string(), None, None));
+        assert_eq!(rel.table_to_track.len(), 1);
+        assert_eq!(rel.table_to_track[0].table_name, "users");
+    }
+
+    #[test]
+    fn test_table_to_track_with_where_clause() {
+        let tt = TableToTrack::new(
+            "orders".to_string(),
+            Some("status = 'active'".to_string()),
+            None,
+        );
+        assert_eq!(tt.where_clause, Some("status = 'active'".to_string()));
+    }
+
+    #[test]
+    fn test_table_to_track_with_columns() {
+        let tt = TableToTrack::new(
+            "users".to_string(),
+            None,
+            Some(vec!["id".to_string(), "email".to_string()]),
+        );
+        assert_eq!(tt.columns_name.as_ref().unwrap().len(), 2);
+    }
+}

backend/windmill-trigger-postgres/src/mapper.rs

@@ -133,3 +133,90 @@ export async function main(
         )
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_postgres_to_typescript_bool() {
+        assert_eq!(postgres_to_typescript_type(Some(Type::BOOL)), "boolean");
+    }
+
+    #[test]
+    fn test_postgres_to_typescript_text_types() {
+        assert_eq!(postgres_to_typescript_type(Some(Type::TEXT)), "string");
+        assert_eq!(postgres_to_typescript_type(Some(Type::VARCHAR)), "string");
+        assert_eq!(postgres_to_typescript_type(Some(Type::CHAR)), "string");
+    }
+
+    #[test]
+    fn test_postgres_to_typescript_number_types() {
+        assert_eq!(postgres_to_typescript_type(Some(Type::INT2)), "number");
+        assert_eq!(postgres_to_typescript_type(Some(Type::INT4)), "number");
+        assert_eq!(postgres_to_typescript_type(Some(Type::INT8)), "number");
+        assert_eq!(postgres_to_typescript_type(Some(Type::FLOAT4)), "number");
+        assert_eq!(postgres_to_typescript_type(Some(Type::FLOAT8)), "number");
+        assert_eq!(postgres_to_typescript_type(Some(Type::NUMERIC)), "number");
+    }
+
+    #[test]
+    fn test_postgres_to_typescript_array_types() {
+        assert_eq!(
+            postgres_to_typescript_type(Some(Type::INT4_ARRAY)),
+            "Array<number>"
+        );
+        assert_eq!(
+            postgres_to_typescript_type(Some(Type::TEXT_ARRAY)),
+            "Array<string>"
+        );
+        assert_eq!(
+            postgres_to_typescript_type(Some(Type::BOOL_ARRAY)),
+            "Array<boolean>"
+        );
+    }
+
+    #[test]
+    fn test_postgres_to_typescript_date_types() {
+        assert_eq!(postgres_to_typescript_type(Some(Type::DATE)), "string");
+        assert_eq!(postgres_to_typescript_type(Some(Type::TIMESTAMP)), "string");
+        assert_eq!(postgres_to_typescript_type(Some(Type::TIMESTAMPTZ)), "string");
+        assert_eq!(postgres_to_typescript_type(Some(Type::UUID)), "string");
+    }
+
+    #[test]
+    fn test_postgres_to_typescript_json() {
+        assert_eq!(postgres_to_typescript_type(Some(Type::JSON)), "unknown");
+        assert_eq!(postgres_to_typescript_type(Some(Type::JSONB)), "unknown");
+    }
+
+    #[test]
+    fn test_postgres_to_typescript_none() {
+        assert_eq!(postgres_to_typescript_type(None), "string");
+    }
+
+    #[test]
+    fn test_into_body_struct_typescript() {
+        let fields = vec![
+            MappingInfo::new("id".to_string(), Some(Type::INT4), false),
+            MappingInfo::new("name".to_string(), Some(Type::TEXT), true),
+        ];
+        let result = into_body_struct(Language::Typescript, fields);
+        assert!(result.contains("id: number,"));
+        assert!(result.contains("name?: string,"));
+    }
+
+    #[test]
+    fn test_empty_template() {
+        let mapper = Mapper::new(HashMap::new(), Language::Typescript);
+        let template = mapper.get_template();
+        assert!(template.contains("row: any"));
+    }
+
+    #[test]
+    fn test_mapping_info_nullable_field() {
+        let info = MappingInfo::new("email".to_string(), Some(Type::VARCHAR), true);
+        assert!(info.is_nullable);
+        assert_eq!(info.column_name, "email");
+    }
+}

backend/windmill-trigger-postgres/src/relation.rs

@@ -72,3 +72,117 @@ impl RelationConverter {
         Ok(object)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use bytes::Bytes;
+    use rust_postgres::types::Type;
+    use serde_json::json;
+
+    use super::super::replication_message::{Column, ReplicaIdentity, RelationBody};
+
+    fn make_relation(o_id: Oid, columns: Vec<Column>) -> RelationBody {
+        RelationBody::new(None, o_id, "public".to_string(), "test".to_string(), ReplicaIdentity::Default, columns)
+    }
+
+    fn text_col(name: &str, typ: Option<Type>) -> Column {
+        Column::new(0, name.to_string(), typ, -1)
+    }
+
+    #[test]
+    fn test_row_to_json_text_columns() {
+        let mut converter = RelationConverter::new();
+        converter.add_relation(make_relation(1, vec![
+            text_col("id", Some(Type::INT4)),
+            text_col("name", Some(Type::TEXT)),
+        ]));
+
+        let tuple = vec![
+            TupleData::Text(Bytes::from("42")),
+            TupleData::Text(Bytes::from("Alice")),
+        ];
+
+        let result = converter.row_to_json((1, tuple)).unwrap();
+        assert_eq!(result["id"], json!(42));
+        assert_eq!(result["name"], json!("Alice"));
+    }
+
+    #[test]
+    fn test_row_to_json_with_null() {
+        let mut converter = RelationConverter::new();
+        converter.add_relation(make_relation(1, vec![
+            text_col("id", Some(Type::INT4)),
+            text_col("email", Some(Type::TEXT)),
+        ]));
+
+        let tuple = vec![
+            TupleData::Text(Bytes::from("1")),
+            TupleData::Null,
+        ];
+
+        let result = converter.row_to_json((1, tuple)).unwrap();
+        assert_eq!(result["id"], json!(1));
+        assert_eq!(result["email"], Value::Null);
+    }
+
+    #[test]
+    fn test_row_to_json_with_unchanged_toast() {
+        let mut converter = RelationConverter::new();
+        converter.add_relation(make_relation(1, vec![
+            text_col("data", Some(Type::TEXT)),
+        ]));
+
+        let tuple = vec![TupleData::UnchangedToast];
+        let result = converter.row_to_json((1, tuple)).unwrap();
+        assert_eq!(result["data"], Value::Null);
+    }
+
+    #[test]
+    fn test_row_to_json_binary_rejected() {
+        let mut converter = RelationConverter::new();
+        converter.add_relation(make_relation(1, vec![
+            text_col("data", Some(Type::BYTEA)),
+        ]));
+
+        let tuple = vec![TupleData::Binary(Bytes::from("data"))];
+        assert!(matches!(
+            converter.row_to_json((1, tuple)),
+            Err(RelationConversionError::BinaryFormatNotSupported)
+        ));
+    }
+
+    #[test]
+    fn test_missing_relation() {
+        let converter = RelationConverter::new();
+        let tuple = vec![TupleData::Null];
+        assert!(matches!(
+            converter.row_to_json((999, tuple)),
+            Err(RelationConversionError::FailToFindMatchingTable)
+        ));
+    }
+
+    #[test]
+    fn test_multiple_relations() {
+        let mut converter = RelationConverter::new();
+        converter.add_relation(make_relation(1, vec![text_col("a", Some(Type::TEXT))]));
+        converter.add_relation(make_relation(2, vec![text_col("b", Some(Type::INT4))]));
+
+        let result1 = converter.row_to_json((1, vec![TupleData::Text(Bytes::from("hello"))])).unwrap();
+        let result2 = converter.row_to_json((2, vec![TupleData::Text(Bytes::from("42"))])).unwrap();
+
+        assert_eq!(result1["a"], json!("hello"));
+        assert_eq!(result2["b"], json!(42));
+    }
+
+    #[test]
+    fn test_row_to_json_bool_column() {
+        let mut converter = RelationConverter::new();
+        converter.add_relation(make_relation(1, vec![
+            text_col("active", Some(Type::BOOL)),
+        ]));
+
+        let result = converter.row_to_json((1, vec![TupleData::Text(Bytes::from("t"))])).unwrap();
+        assert_eq!(result["active"], json!(true));
+    }
+}

backend/windmill-trigger-postgres/src/replication_message.rs

@@ -508,3 +508,309 @@ impl ReplicationMessage {
         Ok(replication_message)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn build_keepalive(wal_end: u64, timestamp: i64, reply: bool) -> Bytes {
+        let mut buf = Vec::new();
+        buf.push(PRIMARY_KEEPALIVE_BYTE);
+        buf.extend_from_slice(&wal_end.to_be_bytes());
+        buf.extend_from_slice(&timestamp.to_be_bytes());
+        buf.push(if reply { 1 } else { 0 });
+        Bytes::from(buf)
+    }
+
+    fn build_xlog_data(wal_start: u64, wal_end: u64, timestamp: i64, data: &[u8]) -> Bytes {
+        let mut buf = Vec::new();
+        buf.push(X_LOG_DATA_BYTE);
+        buf.extend_from_slice(&wal_start.to_be_bytes());
+        buf.extend_from_slice(&wal_end.to_be_bytes());
+        buf.extend_from_slice(&timestamp.to_be_bytes());
+        buf.extend_from_slice(data);
+        Bytes::from(buf)
+    }
+
+    #[test]
+    fn test_parse_keepalive_with_reply() {
+        let buf = build_keepalive(100, 200, true);
+        match ReplicationMessage::parse(buf).unwrap() {
+            ReplicationMessage::PrimaryKeepAlive(body) => {
+                assert_eq!(body.wal_end, 100);
+                assert_eq!(body.timestamp, 200);
+                assert!(body.reply);
+            }
+            _ => panic!("expected PrimaryKeepAlive"),
+        }
+    }
+
+    #[test]
+    fn test_parse_keepalive_without_reply() {
+        let buf = build_keepalive(500, 1000, false);
+        match ReplicationMessage::parse(buf).unwrap() {
+            ReplicationMessage::PrimaryKeepAlive(body) => {
+                assert_eq!(body.wal_end, 500);
+                assert_eq!(body.timestamp, 1000);
+                assert!(!body.reply);
+            }
+            _ => panic!("expected PrimaryKeepAlive"),
+        }
+    }
+
+    #[test]
+    fn test_parse_xlog_data() {
+        let payload = b"test payload";
+        let buf = build_xlog_data(10, 20, 30, payload);
+        match ReplicationMessage::parse(buf).unwrap() {
+            ReplicationMessage::XLogData(body) => {
+                assert_eq!(body.wal_start, 10);
+                assert_eq!(body.wal_end, 20);
+                assert_eq!(body.timestamp, 30);
+                assert_eq!(&body.data[..], payload);
+            }
+            _ => panic!("expected XLogData"),
+        }
+    }
+
+    #[test]
+    fn test_parse_unknown_byte() {
+        let buf = Bytes::from(vec![0xFF, 0, 0, 0, 0, 0, 0, 0, 0]);
+        assert!(ReplicationMessage::parse(buf).is_err());
+    }
+
+    fn build_begin_message() -> Vec<u8> {
+        let mut buf = Vec::new();
+        buf.push(BEGIN_BYTE);
+        buf.extend_from_slice(&0i64.to_be_bytes()); // lsn
+        buf.extend_from_slice(&0i64.to_be_bytes()); // timestamp
+        buf.extend_from_slice(&0i32.to_be_bytes()); // xid
+        buf
+    }
+
+    fn build_commit_message() -> Vec<u8> {
+        let mut buf = Vec::new();
+        buf.push(COMMIT_BYTE);
+        buf.push(0); // flags
+        buf.extend_from_slice(&0u64.to_be_bytes()); // lsn
+        buf.extend_from_slice(&0u64.to_be_bytes()); // end_lsn
+        buf.extend_from_slice(&0i64.to_be_bytes()); // timestamp
+        buf
+    }
+
+    fn build_insert_message(o_id: u32, tuple_data: &[(u8, &[u8])]) -> Vec<u8> {
+        let mut buf = Vec::new();
+        buf.push(INSERT_BYTE);
+        buf.extend_from_slice(&o_id.to_be_bytes());
+        buf.push(TUPLE_NEW_BYTE);
+        buf.extend_from_slice(&(tuple_data.len() as i16).to_be_bytes());
+        for (tag, data) in tuple_data {
+            buf.push(*tag);
+            match *tag {
+                TUPLE_DATA_TEXT_BYTE | TUPLE_DATA_BINARY_BYTE => {
+                    buf.extend_from_slice(&(data.len() as i32).to_be_bytes());
+                    buf.extend_from_slice(data);
+                }
+                _ => {}
+            }
+        }
+        buf
+    }
+
+    fn build_delete_message(o_id: u32, tuple_data: &[(u8, &[u8])]) -> Vec<u8> {
+        let mut buf = Vec::new();
+        buf.push(DELETE_BYTE);
+        buf.extend_from_slice(&o_id.to_be_bytes());
+        buf.push(TUPLE_OLD_BYTE);
+        buf.extend_from_slice(&(tuple_data.len() as i16).to_be_bytes());
+        for (tag, data) in tuple_data {
+            buf.push(*tag);
+            match *tag {
+                TUPLE_DATA_TEXT_BYTE | TUPLE_DATA_BINARY_BYTE => {
+                    buf.extend_from_slice(&(data.len() as i32).to_be_bytes());
+                    buf.extend_from_slice(data);
+                }
+                _ => {}
+            }
+        }
+        buf
+    }
+
+    fn settings(streaming: bool) -> LogicalReplicationSettings {
+        LogicalReplicationSettings { streaming }
+    }
+
+    #[test]
+    fn test_parse_begin() {
+        let data = Bytes::from(build_begin_message());
+        let body = XLogDataBody::new(0, 0, 0, data);
+        match body.parse(&settings(false)).unwrap() {
+            LogicalReplicationMessage::Begin => {}
+            other => panic!("expected Begin, got {:?}", other),
+        }
+    }
+
+    #[test]
+    fn test_parse_commit() {
+        let data = Bytes::from(build_commit_message());
+        let body = XLogDataBody::new(0, 0, 0, data);
+        match body.parse(&settings(false)).unwrap() {
+            LogicalReplicationMessage::Commit => {}
+            other => panic!("expected Commit, got {:?}", other),
+        }
+    }
+
+    #[test]
+    fn test_parse_insert_with_text_tuple() {
+        let data = Bytes::from(build_insert_message(
+            42,
+            &[(TUPLE_DATA_TEXT_BYTE, b"hello")],
+        ));
+        let body = XLogDataBody::new(0, 0, 0, data);
+        match body.parse(&settings(false)).unwrap() {
+            LogicalReplicationMessage::Insert(insert) => {
+                assert_eq!(insert.o_id, 42);
+                assert_eq!(insert.tuple.len(), 1);
+                match &insert.tuple[0] {
+                    TupleData::Text(b) => assert_eq!(&b[..], b"hello"),
+                    other => panic!("expected Text, got {:?}", other),
+                }
+            }
+            other => panic!("expected Insert, got {:?}", other),
+        }
+    }
+
+    #[test]
+    fn test_parse_insert_with_null_tuple() {
+        let data = Bytes::from(build_insert_message(
+            10,
+            &[(TUPLE_DATA_NULL_BYTE, &[])],
+        ));
+        let body = XLogDataBody::new(0, 0, 0, data);
+        match body.parse(&settings(false)).unwrap() {
+            LogicalReplicationMessage::Insert(insert) => {
+                assert_eq!(insert.tuple.len(), 1);
+                assert!(matches!(insert.tuple[0], TupleData::Null));
+            }
+            other => panic!("expected Insert, got {:?}", other),
+        }
+    }
+
+    #[test]
+    fn test_parse_insert_with_toast_tuple() {
+        let data = Bytes::from(build_insert_message(
+            10,
+            &[(TUPLE_DATA_TOAST_BYTE, &[])],
+        ));
+        let body = XLogDataBody::new(0, 0, 0, data);
+        match body.parse(&settings(false)).unwrap() {
+            LogicalReplicationMessage::Insert(insert) => {
+                assert!(matches!(insert.tuple[0], TupleData::UnchangedToast));
+            }
+            other => panic!("expected Insert, got {:?}", other),
+        }
+    }
+
+    #[test]
+    fn test_parse_insert_multiple_columns() {
+        let data = Bytes::from(build_insert_message(
+            1,
+            &[
+                (TUPLE_DATA_TEXT_BYTE, b"col1"),
+                (TUPLE_DATA_NULL_BYTE, &[]),
+                (TUPLE_DATA_TEXT_BYTE, b"col3"),
+            ],
+        ));
+        let body = XLogDataBody::new(0, 0, 0, data);
+        match body.parse(&settings(false)).unwrap() {
+            LogicalReplicationMessage::Insert(insert) => {
+                assert_eq!(insert.tuple.len(), 3);
+                assert!(matches!(&insert.tuple[0], TupleData::Text(_)));
+                assert!(matches!(insert.tuple[1], TupleData::Null));
+                assert!(matches!(&insert.tuple[2], TupleData::Text(_)));
+            }
+            other => panic!("expected Insert, got {:?}", other),
+        }
+    }
+
+    #[test]
+    fn test_parse_delete_with_old_tuple() {
+        let data = Bytes::from(build_delete_message(
+            99,
+            &[(TUPLE_DATA_TEXT_BYTE, b"old_val")],
+        ));
+        let body = XLogDataBody::new(0, 0, 0, data);
+        match body.parse(&settings(false)).unwrap() {
+            LogicalReplicationMessage::Delete(delete) => {
+                assert_eq!(delete.o_id, 99);
+                assert!(delete.old_tuple.is_some());
+                assert!(delete.key_tuple.is_none());
+            }
+            other => panic!("expected Delete, got {:?}", other),
+        }
+    }
+
+    #[test]
+    fn test_parse_relation() {
+        let mut buf = Vec::new();
+        buf.push(RELATION_BYTE);
+        buf.extend_from_slice(&100u32.to_be_bytes()); // o_id
+        buf.extend_from_slice(b"public\0"); // namespace
+        buf.extend_from_slice(b"users\0"); // name
+        buf.push(REPLICA_IDENTITY_DEFAULT_BYTE as u8); // replica identity
+        buf.extend_from_slice(&1i16.to_be_bytes()); // num columns
+        // column: flags=0, name="id", type_oid=23 (INT4), type_modifier=-1
+        buf.push(0); // flags
+        buf.extend_from_slice(b"id\0"); // name
+        buf.extend_from_slice(&23u32.to_be_bytes()); // type_oid (INT4)
+        buf.extend_from_slice(&(-1i32).to_be_bytes()); // type_modifier
+
+        let data = Bytes::from(buf);
+        let body = XLogDataBody::new(0, 0, 0, data);
+        match body.parse(&settings(false)).unwrap() {
+            LogicalReplicationMessage::Relation(rel) => {
+                assert_eq!(rel.o_id, 100);
+                assert_eq!(rel.namespace, "public");
+                assert_eq!(rel.name, "users");
+                assert_eq!(rel.columns.len(), 1);
+                assert_eq!(rel.columns[0].name, "id");
+                assert!(matches!(rel.columns[0].type_o_id, Some(Type::INT4)));
+            }
+            other => panic!("expected Relation, got {:?}", other),
+        }
+    }
+
+    #[test]
+    fn test_parse_insert_with_streaming_transaction_id() {
+        let mut buf = Vec::new();
+        buf.push(INSERT_BYTE);
+        buf.extend_from_slice(&42i32.to_be_bytes()); // transaction_id
+        buf.extend_from_slice(&10u32.to_be_bytes()); // o_id
+        buf.push(TUPLE_NEW_BYTE);
+        buf.extend_from_slice(&0i16.to_be_bytes()); // 0 columns
+
+        let data = Bytes::from(buf);
+        let body = XLogDataBody::new(0, 0, 0, data);
+        match body.parse(&settings(true)).unwrap() {
+            LogicalReplicationMessage::Insert(insert) => {
+                assert_eq!(insert.transaction_id, Some(42));
+                assert_eq!(insert.o_id, 10);
+            }
+            other => panic!("expected Insert, got {:?}", other),
+        }
+    }
+
+    #[test]
+    fn test_unknown_tuple_data_byte() {
+        let mut buf = Vec::new();
+        buf.push(INSERT_BYTE);
+        buf.extend_from_slice(&1u32.to_be_bytes()); // o_id
+        buf.push(TUPLE_NEW_BYTE);
+        buf.extend_from_slice(&1i16.to_be_bytes()); // 1 column
+        buf.push(0xFF); // invalid tuple data byte
+
+        let data = Bytes::from(buf);
+        let body = XLogDataBody::new(0, 0, 0, data);
+        assert!(body.parse(&settings(false)).is_err());
+    }
+}