test(config): schema

1 files changed, 63 insertions, 129 deletions

diff --git a/lib/api-config/src/schema/mod.rs b/lib/api-config/src/schema/mod.rs
index f626e87..d16ee9f 100644
--- a/lib/api-config/src/schema/mod.rs
+++ b/lib/api-config/src/schema/mod.rs
@@ -1,7 +1,10 @@
-mod create;
-pub use create::CreateSchema;
+mod create_schema;
+mod delete_schema;
+mod get_schema;
+mod implementation;
+mod update_schema;
+pub use create_schema::CreateSchema;
 use sqlx::PgPool;
-use tracing::debug;
 
 use async_trait::async_trait;
 use serde::{Deserialize, Serialize};
@@ -96,133 +99,64 @@ pub trait SchemaDriver: Send + Sync {
     ) -> Result<Vec<TransactionSchema>, ConfigurationError>;
 }
 
-#[async_trait]
-impl SchemaDriver for SchemaService {
-    #[tracing::instrument(skip(self, schema))]
-    async fn create_schema(
-        &self,
-        kind: &str,
-        version: &str,
-        schema: &serde_json::Value,
-    ) -> Result<TransactionSchema, crate::ConfigurationError> {
-        debug!("creating transaction schema");
-        sqlx::query_as!(
-            TransactionSchema,
-            "insert into transaction_schema (schema_type, schema_version, schema) values ($1, $2, $3)
-            returning *
-            ",
-            kind,
-            version,
-            sqlx::types::Json(&schema) as _
-        )
-        .fetch_one(&self.database)
-        .await
-        .map_err(|e| e.into())
-    }
+#[cfg(test)]
+mod tests {
+    use sqlx::PgPool;
+
+    use crate::schema::{SchemaDriver, SchemaService};
+
+    #[sqlx::test(
+        migrator = "crate::MIGRATOR",
+        fixtures(path = "../../tests/fixtures", scripts("schema"))
+    )]
+    async fn schema(pool: PgPool) -> anyhow::Result<()> {
+        let driver = SchemaService { database: pool };
+
+        // 2. Define Fixtures
+        let kind = "fin_tx_v1";
+        let version = "1.0.0";
+        let min_schema = serde_json::json!({
+            "type": "object",
+            "required": ["amount", "currency"],
+            "properties": {
+                "amount": { "type": "integer" },
+                "ccy": { "type": "string" }
+            }
+        });
+
+        // CREATE
+        let created = driver
+            .create_schema(kind, version, &min_schema)
+            .await
+            .expect("Create failed");
+        assert_eq!(created.schema_type, kind);
+
+        // GET
+        let found = driver
+            .get_schema("payment", "1.0.0")
+            .await
+            .expect("Get failed")
+            .expect("Schema missing");
+        assert_eq!(found.schema["required"], min_schema["required"]);
+
+        // UPDATE
+        let updated_json = serde_json::json!({"type": "object", "note": "updated"});
+        let updated = driver
+            .update_schema(kind, version, &updated_json)
+            .await
+            .expect("Update failed")
+            .expect("No row to update");
+        assert_eq!(updated.schema["note"], "updated");
+
+        // DELETE
+        driver
+            .delete_schema(kind, version)
+            .await
+            .expect("Delete failed");
+
+        let final_check = driver.get_schema(kind, version).await.unwrap();
+        assert!(final_check.is_none());
 
-    #[tracing::instrument(skip(self))]
-    async fn delete_schema(
-        &self,
-        kind: &str,
-        version: &str,
-    ) -> Result<(), crate::ConfigurationError> {
-        debug!("deleting transaction schema");
-        sqlx::query!(
-            "delete from transaction_schema where schema_type = $1 and schema_version = $2",
-            kind,
-            version,
-        )
-        .execute(&self.database)
-        .await?;
         Ok(())
     }
-
-    #[tracing::instrument(skip(self))]
-    async fn get_schema(
-        &self,
-        kind: &str,
-        version: &str,
-    ) -> Result<Option<TransactionSchema>, crate::ConfigurationError> {
-        debug!("getting transaction schema");
-        let result = sqlx::query_as!(
-            TransactionSchema,
-            "select 
-            *
-            from transaction_schema where schema_type = $1 and schema_version = $2",
-            kind,
-            version,
-        )
-        .fetch_optional(&self.database)
-        .await?;
-
-        Ok(result)
-    }
-
-    #[tracing::instrument(skip(self, schema))]
-    async fn update_schema(
-        &self,
-        kind: &str,
-        version: &str,
-        schema: &serde_json::Value,
-    ) -> Result<Option<TransactionSchema>, crate::ConfigurationError> {
-        debug!("updating transaction schema");
-        sqlx::query_as!(
-            TransactionSchema,
-            "
-            update
-                transaction_schema
-            set 
-                schema = $3
-            where 
-                schema_type = $1 
-                and schema_version = $2
-            returning *
-        ",
-            kind,
-            version,
-            sqlx::types::Json(&schema) as _
-        )
-        .fetch_optional(&self.database)
-        .await
-        .map_err(|e| e.into())
-    }
-
-    #[tracing::instrument(skip(self))]
-    async fn get_schemas(
-        &self,
-        limit: i64,
-        first: Option<i64>,
-        after: Option<&str>,
-    ) -> Result<Vec<TransactionSchema>, ConfigurationError> {
-        debug!("getting transaction schemas");
-        let limit = first.unwrap_or(limit);
-        let mut last_type = String::default();
-        let mut last_version = String::default();
-
-        if let Some(s) = after {
-            let parts: Vec<&str> = s.split(',').collect();
-            if parts.len() == 2 {
-                last_type = parts[0].to_string();
-                last_version = parts[1].to_string();
-            }
-        }
-
-        let rows = sqlx::query_as!(
-            TransactionSchema,
-            "
-            select *
-            from transaction_schema
-            where ($1 = '' or (schema_type, schema_version) > ($1, $2))
-            order by schema_type asc, schema_version asc
-            limit $3
-        ",
-            &last_type,
-            &last_version,
-            limit + 1
-        )
-        .fetch_all(&self.database)
-        .await?;
-
-        Ok(rows)
-    }
 }