Skip to content

feat: add /:id/complete endpoint #16

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
paullatzelsperger merged 6 commits into from
Oct 15, 2025
Merged

feat: add /:id/complete endpoint #16

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
a562de8
feat: add /:id/complete endpoint
paullatzelsperger Oct 14, 2025
bbd8131
complete -> completed
paullatzelsperger Oct 15, 2025
d5828e3
startById is only valid for consumer DFs
paullatzelsperger Oct 15, 2025
5b0aa8f
update source/dest -> dataAddress
paullatzelsperger Oct 15, 2025
1447cee
made dataAddress optional
paullatzelsperger Oct 15, 2025
1875ff1
made dataAddress optional
paullatzelsperger Oct 15, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
140 changes: 97 additions & 43 deletions DataPlane.Sdk.Api.Test/DataPlaneSignalingApiControllerTest.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -143,7 +143,7 @@ public async Task Prepare_Success()
DatasetId = "test-asset",
ParticipantId = TestUser,
AgreementId = "test-agreement",
DestinationDataAddress = new DataAddress("test-type"),
DataAddress = new DataAddress("test-type"),
TransferType = new TransferType
{
DestinationType = "test-type",
Expand Down Expand Up @@ -174,7 +174,7 @@ public async Task Prepare_WhenReturnsSync_Success()
DatasetId = "test-asset",
ParticipantId = TestUser,
AgreementId = "test-agreement",
DestinationDataAddress = new DataAddress("test-type"),
DataAddress = new DataAddress("test-type"),
TransferType = new TransferType
{
DestinationType = "test-type",
Expand Down Expand Up @@ -205,7 +205,7 @@ public async Task Prepare_WhenReturnsAsync_Success()
DatasetId = "test-asset",
ParticipantId = TestUser,
AgreementId = "test-agreement",
DestinationDataAddress = new DataAddress("test-type"),
DataAddress = new DataAddress("test-type"),
TransferType = new TransferType
{
DestinationType = "test-type",
Expand Down Expand Up @@ -238,7 +238,7 @@ public async Task Prepare_WhenSdkReturnsInvalidState_Expect400()
DatasetId = "test-asset",
ParticipantId = TestUser,
AgreementId = "test-agreement",
DestinationDataAddress = new DataAddress("test-type"),
DataAddress = new DataAddress("test-type"),
TransferType = new TransferType
{
DestinationType = "test-type",
Expand All @@ -263,7 +263,7 @@ public async Task Prepare_WhenDataflowExists_Expect409()
DatasetId = "test-asset",
ParticipantId = TestUser,
AgreementId = "test-agreement",
DestinationDataAddress = new DataAddress("test-type"),
DataAddress = new DataAddress("test-type"),
TransferType = new TransferType
{
DestinationType = "test-type",
Expand Down Expand Up @@ -359,11 +359,7 @@ public async Task Start_Success()
DatasetId = "test-asset",
ParticipantId = TestUser,
AgreementId = "test-agreement",
SourceDataAddress = new DataAddress("test-type")
{
Properties = { ["test-key"] = "test-value" }
},
DestinationDataAddress = new DataAddress("test-type")
DataAddress = new DataAddress("test-type")
{
Properties = { ["test-key"] = "test-value" }
},
Expand Down Expand Up @@ -396,11 +392,7 @@ public async Task Start_WhenAlreadyExists_ExpectConflict()
DatasetId = "test-asset",
ParticipantId = TestUser,
AgreementId = "test-agreement",
SourceDataAddress = new DataAddress("test-type")
{
Properties = { ["test-key"] = "test-value" }
},
DestinationDataAddress = new DataAddress("test-type")
DataAddress = new DataAddress("test-type")
{
Properties = { ["test-key"] = "test-value" }
},
Expand Down Expand Up @@ -432,11 +424,7 @@ public async Task Start_SdkHandlerWrongState_ExpectBadRequest()
DatasetId = "test-asset",
ParticipantId = TestUser,
AgreementId = "test-agreement",
SourceDataAddress = new DataAddress("test-type")
{
Properties = { ["test-key"] = "test-value" }
},
DestinationDataAddress = new DataAddress("test-type")
DataAddress = new DataAddress("test-type")
{
Properties = { ["test-key"] = "test-value" }
},
Expand Down Expand Up @@ -470,8 +458,7 @@ public async Task Start_WhenSdkReturnsStarted_Success()
DatasetId = "test-asset",
ParticipantId = TestUser,
AgreementId = "test-agreement",
SourceDataAddress = new DataAddress("test-type"),
DestinationDataAddress = new DataAddress("test-type"),
DataAddress = new DataAddress("test-type"),
TransferType = new TransferType
{
DestinationType = "test-type",
Expand Down Expand Up @@ -501,8 +488,7 @@ public async Task Start_WhenSdkReturnsStarting_Success()
DatasetId = "test-asset",
ParticipantId = TestUser,
AgreementId = "test-agreement",
SourceDataAddress = new DataAddress("test-type"),
DestinationDataAddress = new DataAddress("test-type"),
DataAddress = new DataAddress("test-type"),
TransferType = new TransferType
{
DestinationType = "test-type",
Expand All @@ -523,24 +509,25 @@ public async Task Start_WhenSdkReturnsStarting_Success()

#endregion

#region StartById
#region Notify-Started

[Fact]
public async Task StartById_Success()
{
Sdk.OnStart = null;
var flow = CreateDataFlow();
flow.IsConsumer = true;
DataFlowContext.DataFlows.Add(flow);
await DataFlowContext.SaveChangesAsync();

var startMsg = new DataFlowStartByIdMessage
var startMsg = new DataFlowStartedNotificationMessage
{
SourceDataAddress = new DataAddress("test-type")
DataAddress = new DataAddress("test-type")
{
Properties = { ["test-key"] = "test-value" }
}
};
var response = await HttpClient.PostAsJsonAsync($"/api/v1/{TestUser}/dataflows/{flow.Id}/start", startMsg);
var response = await HttpClient.PostAsJsonAsync($"/api/v1/{TestUser}/dataflows/{flow.Id}/started", startMsg);
response.StatusCode.ShouldBe(HttpStatusCode.OK);
var body = await response.Content.ReadFromJsonAsync<DataFlowResponseMessage>();
body.ShouldNotBeNull();
Expand All @@ -553,14 +540,14 @@ public async Task StartById_WhenNotFound_ExpectError()
{
Sdk.OnStart = null;

var startMsg = new DataFlowStartByIdMessage
var startMsg = new DataFlowStartedNotificationMessage
{
SourceDataAddress = new DataAddress("test-type")
DataAddress = new DataAddress("test-type")
{
Properties = { ["test-key"] = "test-value" }
}
};
var response = await HttpClient.PostAsJsonAsync($"/api/v1/{TestUser}/dataflows/not-exist/start", startMsg);
var response = await HttpClient.PostAsJsonAsync($"/api/v1/{TestUser}/dataflows/not-exist/started", startMsg);
response.StatusCode.ShouldBe(HttpStatusCode.NotFound);
}

Expand All @@ -573,14 +560,14 @@ public async Task StartById_InvalidState_ExpectConflict()
DataFlowContext.DataFlows.Add(flow);
await DataFlowContext.SaveChangesAsync();

var startMsg = new DataFlowStartByIdMessage
var startMsg = new DataFlowStartedNotificationMessage
{
SourceDataAddress = new DataAddress("test-type")
DataAddress = new DataAddress("test-type")
{
Properties = { ["test-key"] = "test-value" }
}
};
var response = await HttpClient.PostAsJsonAsync($"/api/v1/{TestUser}/dataflows/{flow.Id}/start", startMsg);
var response = await HttpClient.PostAsJsonAsync($"/api/v1/{TestUser}/dataflows/{flow.Id}/started", startMsg);
response.StatusCode.ShouldBe(HttpStatusCode.Conflict);
}

Expand All @@ -593,17 +580,18 @@ public async Task StartById_WhenSdkReturnsStarting_Success()
return StatusResult<DataFlow>.Success(df);
};
var flow = CreateDataFlow();
flow.IsConsumer = true;
DataFlowContext.DataFlows.Add(flow);
await DataFlowContext.SaveChangesAsync();

var startMsg = new DataFlowStartByIdMessage
var startMsg = new DataFlowStartedNotificationMessage
{
SourceDataAddress = new DataAddress("test-type")
DataAddress = new DataAddress("test-type")
{
Properties = { ["test-key"] = "test-value" }
}
};
var response = await HttpClient.PostAsJsonAsync($"/api/v1/{TestUser}/dataflows/{flow.Id}/start", startMsg);
var response = await HttpClient.PostAsJsonAsync($"/api/v1/{TestUser}/dataflows/{flow.Id}/started", startMsg);
response.StatusCode.ShouldBe(HttpStatusCode.Accepted);
response.Headers.Location.ShouldNotBeNull()
.ToString().ShouldEndWith($"/api/v1/{TestUser}/dataflows/{flow.Id}");
Expand All @@ -622,17 +610,18 @@ public async Task StartById_WhenSdkReturnsStarted_Success()
return StatusResult<DataFlow>.Success(df);
};
var flow = CreateDataFlow();
flow.IsConsumer = true;
DataFlowContext.DataFlows.Add(flow);
await DataFlowContext.SaveChangesAsync();

var startMsg = new DataFlowStartByIdMessage
var startMsg = new DataFlowStartedNotificationMessage
{
SourceDataAddress = new DataAddress("test-type")
DataAddress = new DataAddress("test-type")
{
Properties = { ["test-key"] = "test-value" }
}
};
var response = await HttpClient.PostAsJsonAsync($"/api/v1/{TestUser}/dataflows/{flow.Id}/start", startMsg);
var response = await HttpClient.PostAsJsonAsync($"/api/v1/{TestUser}/dataflows/{flow.Id}/started", startMsg);
response.StatusCode.ShouldBe(HttpStatusCode.OK);
var body = await response.Content.ReadFromJsonAsync<DataFlowResponseMessage>();
body.ShouldNotBeNull();
Expand All @@ -652,14 +641,79 @@ public async Task StartById_SdkHandlerWrongState_ExpectBadRequest()
DataFlowContext.DataFlows.Add(flow);
await DataFlowContext.SaveChangesAsync();

var startMsg = new DataFlowStartByIdMessage
var startMsg = new DataFlowStartedNotificationMessage
{
SourceDataAddress = new DataAddress("test-type")
DataAddress = new DataAddress("test-type")
{
Properties = { ["test-key"] = "test-value" }
}
};
var response = await HttpClient.PostAsJsonAsync($"/api/v1/{TestUser}/dataflows/{flow.Id}/start", startMsg);
var response = await HttpClient.PostAsJsonAsync($"/api/v1/{TestUser}/dataflows/{flow.Id}/started", startMsg);
response.StatusCode.ShouldBe(HttpStatusCode.Conflict);
}

#endregion

#region Complete

[Fact]
public async Task Complete_Success()
{
Sdk.OnComplete = null;
var flow = CreateDataFlow();
flow.State = DataFlowState.Started;
DataFlowContext.DataFlows.Add(flow);
await DataFlowContext.SaveChangesAsync();

var response = await HttpClient.PostAsync($"/api/v1/{TestUser}/dataflows/{flow.Id}/completed", null);
response.StatusCode.ShouldBe(HttpStatusCode.OK);
}

[Fact]
public async Task Complete_WhenNotFound_ExpectNotFound()
{
Sdk.OnComplete = null;

var response = await HttpClient.PostAsync($"/api/v1/{TestUser}/dataflows/not-exist/completed", null);
response.StatusCode.ShouldBe(HttpStatusCode.NotFound);
}

[Fact]
public async Task Complete_WithBody_ExpectBadRequest()
{
Sdk.OnComplete = null;

var response = await HttpClient.PostAsync($"/api/v1/{TestUser}/dataflows/some-flow/completed", new StringContent("{\"foo\": \"bar\"}"));
response.StatusCode.ShouldBe(HttpStatusCode.BadRequest);
}

[Fact]
public async Task Complete_WhenSdkReportsError_ExpectBadRequest()
{
Sdk.OnComplete = df => StatusResult.Failed(new StatusFailure
{
Message = "test error",
Reason = FailureReason.ServiceUnavailable
});

var flow = CreateDataFlow();
flow.State = DataFlowState.Started;
DataFlowContext.DataFlows.Add(flow);
await DataFlowContext.SaveChangesAsync();

var response = await HttpClient.PostAsync($"/api/v1/{TestUser}/dataflows/{flow.Id}/completed", null);
response.StatusCode.ShouldBe(HttpStatusCode.ServiceUnavailable);
}

[Fact]
public async Task Complete_WhenFlowInWrongState_ExpectBadRequest()
{
var flow = CreateDataFlow();
flow.State = DataFlowState.Terminated;
DataFlowContext.DataFlows.Add(flow);
await DataFlowContext.SaveChangesAsync();

var response = await HttpClient.PostAsync($"/api/v1/{TestUser}/dataflows/{flow.Id}/completed", null);
response.StatusCode.ShouldBe(HttpStatusCode.Conflict);
}

Expand Down
23 changes: 21 additions & 2 deletions DataPlane.Sdk.Api/Controllers/DataPlaneSignalingApiController.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,8 +82,9 @@ public async Task<IActionResult> Start([FromRoute] string participantContextId,
}

[Authorize]
[HttpPost("{dataFlowId}/start")]
public async Task<IActionResult> StartById([FromRoute] string participantContextId, [FromRoute] string dataFlowId, DataFlowStartByIdMessage startMessage)
[HttpPost("{dataFlowId}/started")]
public async Task<IActionResult> StartById([FromRoute] string participantContextId, [FromRoute] string dataFlowId,
DataFlowStartedNotificationMessage startMessage)
{
if (!(await authorizationService.AuthorizeAsync(User, new ResourceTuple(participantContextId, dataFlowId), "DataFlowAccess")).Succeeded)
{
Expand Down Expand Up @@ -144,6 +145,24 @@ public async Task<IActionResult> Terminate([FromRoute] string participantContext
return statusResult.IsSucceeded ? Ok() : StatusCode((int)statusResult.Failure!.Reason, statusResult);
}

[Authorize]
[HttpPost("{dataFlowId}/completed")]
public async Task<IActionResult> Complete([FromRoute] string participantContextId, [FromRoute] string dataFlowId)
{
if (!(await authorizationService.AuthorizeAsync(User, new ResourceTuple(participantContextId, dataFlowId), "DataFlowAccess")).Succeeded)
{
return Forbid();
}

if (Request.ContentLength > 0)
{
return BadRequest("Request body is not allowed for this endpoint");
}

var result = await signalingService.CompleteAsync(dataFlowId);
return result.IsSucceeded ? Ok() : StatusCode((int)result.Failure!.Reason, result);
}

[Authorize]
[HttpGet("{dataFlowId}/status")]
public async Task<IActionResult> GetStatus([FromRoute] string dataFlowId, [FromRoute] string participantContextId)
Expand Down
2 changes: 1 addition & 1 deletion DataPlane.Sdk.Core.Test/Data/DataFlowContextTest.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ public async Task SaveAsync_ShouldAddNewDataFlow()

entry.Entity.Id.ShouldBe(dataFlow.Id);
entry.Entity.ShouldBeEquivalentTo(dataFlow);
entry.Entity.Destination.Properties["test-key"].ShouldBeEquivalentTo("test-value");
entry.Entity.Destination?.Properties["test-key"].ShouldBeEquivalentTo("test-value");
}

[Fact]
Expand Down
Loading
Loading