bsodfather/crank
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add soap adapter foundation

Browse Source
This commit is contained in:
a.tolmachev
2026-04-07 00:00:19 +03:00
parent 31fbdfdc02
commit a6388e4353
24 changed files with 1346 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/admin-api/Cargo.toml
+1
View File
@@ -12,6 +12,7 @@ axum.workspace = true
axum-extra.workspace = true
base64.workspace = true
crank-adapter-grpc = { path = "../../crates/crank-adapter-grpc", features = ["test-support"] }
crank-adapter-soap = { path = "../../crates/crank-adapter-soap" }
crank-core = { path = "../../crates/crank-core" }
crank-mapping = { path = "../../crates/crank-mapping" }
crank-proto = { path = "../../crates/crank-proto" }
apps/admin-api/src/app.rs
+199 -3
View File
@@ -24,8 +24,9 @@ use crate::{
operations::{
archive_operation, create_operation, create_version, delete_operation,
export_operation, generate_draft, get_operation, get_operation_version,
list_grpc_services, list_operations, publish_operation, run_test, update_operation,
upload_descriptor_set, upload_input_json, upload_output_json, upload_proto_descriptor,
list_grpc_services, list_operations, list_soap_services, publish_operation, run_test,
update_operation, upload_descriptor_set, upload_input_json, upload_output_json,
upload_proto_descriptor, upload_wsdl_descriptor, upload_xsd_descriptor,
},
secrets::{create_secret, delete_secret, get_secret, list_secrets, rotate_secret},
streaming::{
@@ -77,10 +78,22 @@ pub fn build_app(state: AppState) -> Router {
"/operations/{operation_id}/descriptors/descriptor-set",
post(upload_descriptor_set),
)
.route(
"/operations/{operation_id}/descriptors/wsdl",
post(upload_wsdl_descriptor),
)
.route(
"/operations/{operation_id}/descriptors/xsd",
post(upload_xsd_descriptor),
)
.route(
"/operations/{operation_id}/grpc/services",
get(list_grpc_services),
)
.route(
"/operations/{operation_id}/soap/services",
get(list_soap_services),
)
.route(
"/operations/{operation_id}/drafts/generate",
post(generate_draft),
@@ -209,7 +222,8 @@ mod tests {
use crank_adapter_grpc::test_support as grpc_test_support;
use crank_core::{
DescriptorId, ExecutionConfig, GraphqlOperationType, GraphqlTarget, GrpcTarget, HttpMethod,
MembershipRole, Protocol, RestTarget, SecretKind, Target, ToolDescription, WorkspaceId,
MembershipRole, Protocol, RestTarget, SecretKind, SoapBindingStyle, SoapOperationMetadata,
SoapTarget, SoapVersion, Target, ToolDescription, WorkspaceId,
};
use crank_mapping::{MappingRule, MappingSet};
use crank_registry::PostgresRegistry;
@@ -1427,6 +1441,77 @@ mod tests {
assert_eq!(test_run["response_preview"]["message"], "hello");
}
#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn uploads_wsdl_and_tests_soap_operation() {
let registry = test_registry().await;
let storage_root = test_storage_root("soap_runtime");
let endpoint = spawn_soap_server().await;
let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
let client = authorized_client(&base_url).await;
let created = client
.post(format!("{base_url}/operations"))
.json(&test_soap_operation_payload(
&endpoint,
"crm_create_lead_soap",
))
.send()
.await
.unwrap()
.json::<Value>()
.await
.unwrap();
let operation_id = created["operation_id"].as_str().unwrap().to_owned();
let uploaded = client
.post(format!(
"{base_url}/operations/{operation_id}/descriptors/wsdl"
))
.header("x-file-name", "lead.wsdl")
.body(SOAP_TEST_WSDL)
.send()
.await
.unwrap()
.json::<Value>()
.await
.unwrap();
let services = client
.get(format!(
"{base_url}/operations/{operation_id}/soap/services"
))
.send()
.await
.unwrap()
.json::<Value>()
.await
.unwrap();
let test_run = client
.post(format!("{base_url}/operations/{operation_id}/test-runs"))
.json(&json!({
"version": 1,
"input": { "email": "user@example.com" }
}))
.send()
.await
.unwrap()
.json::<Value>()
.await
.unwrap();
assert_eq!(uploaded["version"], 1);
assert_eq!(services["services"][0]["service_name"], "LeadService");
assert_eq!(services["services"][0]["ports"][0]["port_name"], "LeadPort");
assert_eq!(
services["services"][0]["ports"][0]["operations"][0]["operation_name"],
"CreateLead"
);
assert_eq!(test_run["ok"], true);
assert_eq!(test_run["response_preview"]["id"], "lead_123");
}
#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn manages_auth_profiles_and_yaml_upsert() {
@@ -1779,6 +1864,18 @@ mod tests {
format!("http://{}", address)
}
async fn spawn_soap_server() -> String {
let app = Router::new().route("/", post(soap_handler));
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
tokio::spawn(async move {
axum::serve(listener, app).await.unwrap();
});
format!("http://{}", address)
}
async fn spawn_admin_api(app: Router) -> TestServer {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
@@ -1867,6 +1964,23 @@ mod tests {
}))
}
async fn soap_handler(body: String) -> (axum::http::StatusCode, String) {
assert!(body.contains("<email>user@example.com</email>"));
(
axum::http::StatusCode::OK,
r#"<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/">
<soap:Body>
<CreateLeadResponse>
<id>lead_123</id>
<status>created</status>
</CreateLeadResponse>
</soap:Body>
</soap:Envelope>"#
.to_owned(),
)
}
async fn test_registry() -> PostgresRegistry {
let database_url = env::var("TEST_DATABASE_URL")
.unwrap_or_else(|_| "postgres://crank:crank@127.0.0.1:5432/crank".to_owned());
@@ -2098,6 +2212,88 @@ mod tests {
}
}
fn test_soap_operation_payload(endpoint: &str, name: &str) -> OperationPayload {
OperationPayload {
name: name.to_owned(),
display_name: "Create Lead SOAP".to_owned(),
category: "sales".to_owned(),
protocol: Protocol::Soap,
target: Target::Soap(SoapTarget {
wsdl_ref: "sample_wsdl".into(),
service_name: "LeadService".to_owned(),
port_name: "LeadPort".to_owned(),
operation_name: "CreateLead".to_owned(),
endpoint_override: Some(endpoint.to_owned()),
soap_version: SoapVersion::Soap11,
soap_action: Some("urn:createLead".to_owned()),
binding_style: SoapBindingStyle::DocumentLiteral,
headers: Vec::new(),
fault_contract: None,
metadata: SoapOperationMetadata {
input_part_names: vec!["CreateLeadRequest".to_owned()],
output_part_names: vec!["CreateLeadResponse".to_owned()],
namespaces: vec!["urn:crm".to_owned()],
},
}),
input_schema: object_schema("email"),
output_schema: object_schema("id"),
input_mapping: MappingSet {
rules: vec![MappingRule {
source: "$.mcp.email".to_owned(),
target: "$.request.body.email".to_owned(),
required: true,
default_value: None,
transform: None,
condition: None,
notes: None,
}],
},
output_mapping: MappingSet {
rules: vec![MappingRule {
source: "$.response.body.id".to_owned(),
target: "$.output.id".to_owned(),
required: true,
default_value: None,
transform: None,
condition: None,
notes: None,
}],
},
execution_config: ExecutionConfig {
timeout_ms: 1_000,
retry_policy: None,
auth_profile_ref: None,
headers: BTreeMap::new(),
protocol_options: None,
streaming: None,
},
tool_description: ToolDescription {
title: "Create Lead SOAP".to_owned(),
description: "Creates a CRM lead through SOAP".to_owned(),
tags: vec!["crm".to_owned(), "soap".to_owned()],
examples: Vec::new(),
},
}
}
const SOAP_TEST_WSDL: &str = r#"<?xml version="1.0"?>
<definitions xmlns="http://schemas.xmlsoap.org/wsdl/"
xmlns:soap="http://schemas.xmlsoap.org/wsdl/soap/"
xmlns:tns="urn:crm"
targetNamespace="urn:crm">
<binding name="LeadBinding" type="tns:LeadPortType">
<soap:binding transport="http://schemas.xmlsoap.org/soap/http" style="document"/>
<operation name="CreateLead">
<soap:operation soapAction="urn:createLead"/>
</operation>
</binding>
<service name="LeadService">
<port name="LeadPort" binding="tns:LeadBinding">
<soap:address location="https://soap.example.com/lead"/>
</port>
</service>
</definitions>"#;
fn object_schema(field_name: &str) -> Schema {
Schema {
kind: SchemaKind::Object,
apps/admin-api/src/error.rs
+1
View File
@@ -238,6 +238,7 @@ fn runtime_test_failure_code(error: &RuntimeError) -> &'static str {
RuntimeError::GraphqlAdapter(_) => "runtime_graphql_error",
RuntimeError::GrpcAdapter(_) => "runtime_grpc_error",
RuntimeError::RestAdapter(_) => "runtime_rest_error",
RuntimeError::SoapAdapter(_) => "runtime_soap_error",
RuntimeError::WebsocketAdapter(_) => "runtime_websocket_error",
RuntimeError::UnsupportedProtocol { .. } => "runtime_protocol_error",
RuntimeError::InvalidPreparedRequest { .. } => "runtime_request_error",
apps/admin-api/src/routes/operations.rs
+56
View File
@@ -269,6 +269,46 @@ pub async fn upload_descriptor_set(
Ok(Json(json!(descriptor)))
}
pub async fn upload_wsdl_descriptor(
Path(path): Path<WorkspaceOperationPath>,
State(state): State<AppState>,
headers: HeaderMap,
body: Bytes,
) -> Result<Json<Value>, ApiError> {
let source_name = header_file_name(&headers);
let descriptor = state
.service
.upload_wsdl_file(
&path.workspace_id.as_str().into(),
&path.operation_id.as_str().into(),
source_name.as_deref(),
body.as_ref(),
)
.await?;
Ok(Json(json!(descriptor)))
}
pub async fn upload_xsd_descriptor(
Path(path): Path<WorkspaceOperationPath>,
State(state): State<AppState>,
headers: HeaderMap,
body: Bytes,
) -> Result<Json<Value>, ApiError> {
let source_name = header_file_name(&headers);
let descriptor = state
.service
.upload_xsd_file(
&path.workspace_id.as_str().into(),
&path.operation_id.as_str().into(),
source_name.as_deref(),
body.as_ref(),
)
.await?;
Ok(Json(json!(descriptor)))
}
pub async fn list_grpc_services(
Path(path): Path<WorkspaceOperationPath>,
Query(query): Query<ExportQuery>,
@@ -285,6 +325,22 @@ pub async fn list_grpc_services(
Ok(Json(json!({ "services": services })))
}
pub async fn list_soap_services(
Path(path): Path<WorkspaceOperationPath>,
Query(query): Query<ExportQuery>,
State(state): State<AppState>,
) -> Result<Json<Value>, ApiError> {
let services = state
.service
.list_soap_services(
&path.workspace_id.as_str().into(),
&path.operation_id.as_str().into(),
query.version,
)
.await?;
Ok(Json(json!({ "services": services })))
}
pub async fn generate_draft(
Path(path): Path<WorkspaceOperationPath>,
State(state): State<AppState>,
apps/admin-api/src/service.rs
+151
View File
@@ -6,6 +6,7 @@ use base64::{
engine::general_purpose::{STANDARD, URL_SAFE_NO_PAD},
};
use crank_adapter_grpc::test_support as grpc_test_support;
use crank_adapter_soap::{SoapServiceSummary, inspect_wsdl};
use crank_core::{
Agent, AgentId, AgentOperationBinding, AgentStatus, AgentVersion, AggregationMode,
AsyncJobHandle, AuthConfig, AuthKind, AuthProfile, AuthProfileId, ConfigExport, ExecutionMode,
@@ -563,6 +564,8 @@ pub struct GrpcMethodSummary {
pub output_schema: Schema,
}
pub type SoapServiceCatalog = Vec<SoapServiceSummary>;
fn default_operation_category() -> String {
"general".to_owned()
}
@@ -2599,6 +2602,153 @@ impl AdminService {
.collect::<Result<Vec<_>, _>>()
}
#[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str(), source_name = source_name.unwrap_or("service.wsdl")))]
pub async fn upload_wsdl_file(
&self,
workspace_id: &WorkspaceId,
operation_id: &OperationId,
source_name: Option<&str>,
payload: &[u8],
) -> Result<DescriptorUploadResponse, ApiError> {
let summary = self.get_operation(workspace_id, operation_id).await?;
if summary.protocol != Protocol::Soap {
return Err(ApiError::validation(
"wsdl upload is only allowed for soap operations",
));
}
let services =
inspect_wsdl(payload).map_err(|error| ApiError::validation(error.to_string()))?;
let descriptor_id = crank_core::DescriptorId::new(new_prefixed_id("desc"));
let storage_ref = self
.storage
.write_descriptor(
operation_id,
summary.current_draft_version,
crank_registry::DescriptorKind::WsdlUpload,
&descriptor_id,
source_name,
payload,
)
.await?;
self.registry
.save_descriptor_metadata(SaveDescriptorMetadataRequest {
descriptor: &crank_registry::DescriptorMetadata {
id: descriptor_id.clone(),
operation_id: Some(operation_id.clone()),
version: Some(summary.current_draft_version),
descriptor_kind: crank_registry::DescriptorKind::WsdlUpload,
storage_ref,
source_name: source_name.map(ToOwned::to_owned),
package_index: Some(
serde_json::to_value(&services)
.map_err(|error| ApiError::internal(error.to_string()))?,
),
created_at: now_string()?,
},
})
.await?;
info!(
operation_id = %operation_id.as_str(),
descriptor_id = %descriptor_id.as_str(),
version = summary.current_draft_version,
"wsdl uploaded"
);
Ok(DescriptorUploadResponse {
descriptor_id: descriptor_id.as_str().to_owned(),
version: summary.current_draft_version,
})
}
#[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str(), source_name = source_name.unwrap_or("schema.xsd")))]
pub async fn upload_xsd_file(
&self,
workspace_id: &WorkspaceId,
operation_id: &OperationId,
source_name: Option<&str>,
payload: &[u8],
) -> Result<DescriptorUploadResponse, ApiError> {
let summary = self.get_operation(workspace_id, operation_id).await?;
if summary.protocol != Protocol::Soap {
return Err(ApiError::validation(
"xsd upload is only allowed for soap operations",
));
}
let descriptor_id = crank_core::DescriptorId::new(new_prefixed_id("desc"));
let storage_ref = self
.storage
.write_descriptor(
operation_id,
summary.current_draft_version,
crank_registry::DescriptorKind::XsdUpload,
&descriptor_id,
source_name,
payload,
)
.await?;
self.registry
.save_descriptor_metadata(SaveDescriptorMetadataRequest {
descriptor: &crank_registry::DescriptorMetadata {
id: descriptor_id.clone(),
operation_id: Some(operation_id.clone()),
version: Some(summary.current_draft_version),
descriptor_kind: crank_registry::DescriptorKind::XsdUpload,
storage_ref,
source_name: source_name.map(ToOwned::to_owned),
package_index: None,
created_at: now_string()?,
},
})
.await?;
info!(
operation_id = %operation_id.as_str(),
descriptor_id = %descriptor_id.as_str(),
version = summary.current_draft_version,
"xsd uploaded"
);
Ok(DescriptorUploadResponse {
descriptor_id: descriptor_id.as_str().to_owned(),
version: summary.current_draft_version,
})
}
#[instrument(skip(self), fields(operation_id = %operation_id.as_str(), version))]
pub async fn list_soap_services(
&self,
workspace_id: &WorkspaceId,
operation_id: &OperationId,
version: Option<u32>,
) -> Result<SoapServiceCatalog, ApiError> {
let summary = self.get_operation(workspace_id, operation_id).await?;
if summary.protocol != Protocol::Soap {
return Err(ApiError::validation(
"soap services are only available for soap operations",
));
}
let version = version.unwrap_or(summary.current_draft_version);
let descriptor = self
.registry
.list_descriptor_metadata(operation_id, version)
.await?
.into_iter()
.rev()
.find(|descriptor| {
descriptor.descriptor_kind == crank_registry::DescriptorKind::WsdlUpload
})
.ok_or_else(|| ApiError::not_found("wsdl was not uploaded"))?;
let package_index = descriptor
.package_index
.ok_or_else(|| ApiError::not_found("wsdl inspection metadata is not available"))?;
serde_json::from_value(package_index).map_err(|error| ApiError::internal(error.to_string()))
}
#[instrument(skip(self))]
pub async fn get_auth_profile(
&self,
@@ -4017,6 +4167,7 @@ fn runtime_error_code(error: &RuntimeError) -> &'static str {
RuntimeError::GraphqlAdapter(_) => "graphql_error",
RuntimeError::GrpcAdapter(_) => "grpc_error",
RuntimeError::RestAdapter(_) => "rest_error",
RuntimeError::SoapAdapter(_) => "soap_error",
RuntimeError::WebsocketAdapter(_) => "websocket_error",
RuntimeError::UnsupportedProtocol { .. } => "unsupported_protocol",
RuntimeError::MissingStreamingConfig { .. } => "streaming_config_error",
apps/admin-api/src/storage.rs
+2
View File
@@ -73,6 +73,8 @@ impl LocalArtifactStorage {
DescriptorKind::ReflectionSnapshot => {
source_name.unwrap_or("reflection.bin").to_owned()
}
DescriptorKind::WsdlUpload => source_name.unwrap_or("service.wsdl").to_owned(),
DescriptorKind::XsdUpload => source_name.unwrap_or("schema.xsd").to_owned(),
};
let path = self
.root