Skip to content

feat: add test for streaming data plane #18

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 3 commits into from
Oct 17, 2025
Merged

feat: add test for streaming data plane #18

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
36f2e0b
feat: add test for streaming data plane
paullatzelsperger Oct 17, 2025
4bf9494
use async test assertions
paullatzelsperger Oct 17, 2025
2517bc3
filter for integration test
paullatzelsperger Oct 17, 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
25 changes: 0 additions & 25 deletions .dockerignore
View file
Open in desktop

This file was deleted.

37 changes: 35 additions & 2 deletions .github/workflows/test.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,10 @@ on:
pull_request:
branches: [ "main" ]

concurrency:
group: ${{ github.workflow }}-${{ github.ref }}
cancel-in-progress: true

jobs:
build:

Expand All @@ -33,5 +37,34 @@ jobs:
run: dotnet restore
- name: Build
run: dotnet build --no-restore
- name: Test
run: dotnet test --no-build --verbosity normal
- name: Unit/Integration Tests
run: dotnet test --no-build --verbosity normal --filter "Category!=Integration"

e2e-test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Setup .NET
uses: actions/setup-dotnet@v4
with:
dotnet-version: 9.0.x

- uses: docker/setup-compose-action@v1

- name: Docker Compose Build
env:
NUGET_USERNAME: ${{ github.actor }}
NUGET_PASSWORD: ${{ secrets.GITHUB_TOKEN }}
run: |
docker compose up --build -d
working-directory: ./Samples/Streaming

- name: Execute E2E Tests
run: |
dotnet restore Samples/Streaming/TestRunner/TestRunner.csproj
dotnet test Samples/Streaming/TestRunner/TestRunner.csproj --filter "Category=Integration"

# - name: Cleanup
# run: |
# docker-compose down -v

12 changes: 11 additions & 1 deletion Samples/Streaming/Consumer/Extensions.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,17 @@ public static void AddDataPlaneSdk(this IServiceCollection services, IConfigurat
dataService.Start(NatsDataAddress.Create(dataFlow.Destination)).Wait();
return StatusResult<DataFlow>.Success(dataFlow);
},
OnTerminate = df => StatusResult.Success(),
OnTerminate = df =>
{
if (df.Destination == null)
{
return StatusResult.FromCode(400, "DataFlow.Destination cannot be null");
}

var dataService = services.BuildServiceProvider().GetRequiredService<NatsSubscriber>();
dataService.Stop(NatsDataAddress.Create(df.Destination)).Wait();
return StatusResult.Success();
},
OnSuspend = _ => StatusResult.Success(),
OnPrepare = f =>
{
Expand Down
191 changes: 66 additions & 125 deletions Samples/Streaming/Resources/Postman/StreamingPull.postman_collection.json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,10 +1,10 @@
{
"info": {
"_postman_id": "84fd9da2-29f2-405d-9f51-f28fdf4c532b",
"_postman_id": "a5ec8438-2ba7-45df-bec0-c9006f64e347",
"name": "Streaming Pull",
"schema": "https://schema.getpostman.com/json/collection/v2.1.0/collection.json",
"_exporter_id": "647585",
"_collection_link": "https://galactic-star-228409.postman.co/workspace/Dataplane-work~d820e7a7-8273-4d79-986e-96754a31f467/collection/647585-84fd9da2-29f2-405d-9f51-f28fdf4c532b?action=share&source=collection_link&creator=647585"
"_collection_link": "https://galactic-star-228409.postman.co/workspace/Dataplane-work~d820e7a7-8273-4d79-986e-96754a31f467/collection/647585-a5ec8438-2ba7-45df-bec0-c9006f64e347?action=share&source=collection_link&creator=647585"
},
"item": [
{
Expand Down Expand Up @@ -38,20 +38,18 @@
"header": [],
"body": {
"mode": "raw",
"raw": "{\n \"processId\": \"{{DATAFLOW_ID}}\",\n \"datasetId\": \"test-asset\",\n \"participantId\": \"{{PARTICIPANT_ID}}\",\n \"agreementId\": \"test-agreement\",\n \"transferType\": {\n \"flowType\": \"pull\",\n \"destinationType\": \"NatsStream\"\n }\n}",
"raw": "{\n \"processId\": \"{{DATAFLOW_ID}}\",\n \"datasetId\": \"test-asset\",\n \"participantId\": \"{{PARTICIPANT_ID}}\",\n \"agreementId\": \"test-agreement\",\n \n \"transferType\": {\n \"flowType\": \"pull\",\n \"destinationType\": \"NatsStream\"\n }\n}",
"options": {
"raw": {
"language": "json"
}
}
},
"url": {
"raw": "http://localhost:8081/api/v1/{{PARTICIPANT_ID}}/dataflows/prepare",
"protocol": "http",
"raw": "{{CONSUMER_HOST}}/api/v1/{{PARTICIPANT_ID}}/dataflows/prepare",
"host": [
"localhost"
"{{CONSUMER_HOST}}"
],
"port": "8081",
"path": [
"api",
"v1",
Expand Down Expand Up @@ -102,9 +100,9 @@
}
},
"url": {
"raw": "{{DATAPLANE_HOST}}/api/v1/{{PARTICIPANT_ID}}/dataflows/start",
"raw": "{{PROVIDER_HOST}}/api/v1/{{PARTICIPANT_ID}}/dataflows/start",
"host": [
"{{DATAPLANE_HOST}}"
"{{PROVIDER_HOST}}"
],
"path": [
"api",
Expand All @@ -119,117 +117,6 @@
},
{
"name": "Consumer - Notify Started",
"event": [
{
"listen": "test",
"script": {
"exec": [
"if (pm.response.code < 300 && pm.response.code >= 200) {",
" const responseJson = pm.response.json();",
" pm.expect(responseJson).to.have.property('dataAddress');",
" pm.expect(responseJson.dataAddress).to.have.property('properties');",
" pm.expect(responseJson.dataAddress.properties).to.have.property('url');",
" pm.expect(responseJson.dataAddress.properties.url).to.exist;",
"",
" const url = responseJson.dataAddress.properties.url;",
" pm.collectionVariables.set(\"PUBLIC_URL\", url);",
"",
" pm.expect(responseJson.dataAddress.properties).to.have.property('token');",
" const token = responseJson.dataAddress.properties.token;",
" pm.collectionVariables.set(\"API_KEY\", token);",
"}",
"",
"pm.test(\"Status code is >=200 and <300\", function () {",
" pm.expect(pm.response.code < 300 && pm.response.code >= 200).to.be.true",
"});",
"pm.test(\"Public URL is set\", function(){",
" pm.expect(pm.collectionVariables.get(\"PUBLIC_URL\")).not.to.be.undefined",
"})",
"",
"",
"pm.test(\"API is set\", function(){",
" pm.expect(pm.collectionVariables.get(\"API_KEY\")).not.to.be.undefined",
"})",
""
],
"type": "text/javascript",
"packages": {},
"requests": {}
}
},
{
"listen": "prerequest",
"script": {
"exec": [
""
],
"type": "text/javascript",
"packages": {},
"requests": {}
}
}
],
"request": {
"method": "POST",
"header": [],
"body": {
"mode": "raw",
"raw": "{\n \"processId\": \"{{DATAFLOW_ID}}\",\n \"datasetId\": \"test-asset\",\n \"participantId\": \"{{PARTICIPANT_ID}}\",\n \"agreementId\": \"test-agreement\",\n \"dataAddress\": {\n \"@type\": \"HttpData\",\n \"properties\":{\n \"baseUrl\": \"https://jsonplaceholder.typicode.com/comments/22\"\n }\n },\n \"transferType\": {\n \"flowType\": \"pull\",\n \"destinationType\": \"HttpData\"\n }\n}",
"options": {
"raw": {
"language": "json"
}
}
},
"url": {
"raw": "{{DATAPLANE_HOST}}/api/v1/{{PARTICIPANT_ID}}/dataflows/{{CONSUMER_DATAFLOW_ID}}//started",
"host": [
"{{DATAPLANE_HOST}}"
],
"path": [
"api",
"v1",
"{{PARTICIPANT_ID}}",
"dataflows",
"{{CONSUMER_DATAFLOW_ID}}",
"",
"started"
]
}
},
"response": []
},
{
"name": "Access Public API",
"request": {
"auth": {
"type": "apikey",
"apikey": [
{
"key": "value",
"value": "{{API_KEY}}",
"type": "string"
},
{
"key": "key",
"value": "x-api-key",
"type": "string"
}
]
},
"method": "GET",
"header": [],
"url": {
"raw": "{{PUBLIC_URL}}",
"host": [
"{{PUBLIC_URL}}"
]
}
},
"response": []
},
{
"name": "Complete Dataflow",
"event": [
{
"listen": "test",
Expand Down Expand Up @@ -259,32 +146,34 @@
"header": [],
"body": {
"mode": "raw",
"raw": "",
"raw": "{\n \"processId\": \"{{DATAFLOW_ID}}\",\n \"datasetId\": \"test-asset\",\n \"participantId\": \"{{PARTICIPANT_ID}}\",\n \"agreementId\": \"test-agreement\",\n \"dataAddress\": {\n \"properties\": {\n \"endpointType\": \"https://example.com/natsdp/v1/nats\",\n \"endpoint\": \"nats://nats:4222\",\n \"endpointProperties\": [\n {\n \"key\": \"channel\",\n \"type\": \"string\",\n \"value\": \"{{DATAFLOW_ID}}.forward\"\n },\n {\n \"key\": \"replyChannel\",\n \"type\": \"string\",\n \"value\": \"{{DATAFLOW_ID}}.reply\"\n }\n ]\n },\n \"@context\": {\n \"edc\": \"https://w3id.org/edc/v0.0.1/ns/\"\n },\n \"@type\": \"NatsStream\"\n },\n \"transferType\": {\n \"flowType\": \"pull\",\n \"destinationType\": \"HttpData\"\n }\n}",
"options": {
"raw": {
"language": "json"
}
}
},
"url": {
"raw": "{{DATAPLANE_HOST}}/api/v1/{{PARTICIPANT_ID}}/dataflows/{{DATAFLOW_ID}}/completed",
"raw": "http://localhost:8081/api/v1/{{PARTICIPANT_ID}}/dataflows/{{DATAFLOW_ID}}/started",
"protocol": "http",
"host": [
"{{DATAPLANE_HOST}}"
"localhost"
],
"port": "8081",
"path": [
"api",
"v1",
"{{PARTICIPANT_ID}}",
"dataflows",
"{{DATAFLOW_ID}}",
"completed"
"started"
]
}
},
"response": []
},
{
"name": "Terminate Dataflow",
"name": "Provider - Terminate Dataflow",
"event": [
{
"listen": "test",
Expand Down Expand Up @@ -338,5 +227,57 @@
},
"response": []
}
],
"auth": {
"type": "oauth2",
"oauth2": [
{
"key": "addTokenTo",
"value": "header",
"type": "string"
}
]
},
"event": [
{
"listen": "prerequest",
"script": {
"type": "text/javascript",
"packages": {},
"requests": {},
"exec": [
""
]
}
},
{
"listen": "test",
"script": {
"type": "text/javascript",
"packages": {},
"requests": {},
"exec": [
""
]
}
}
],
"variable": [
{
"key": "PARTICIPANT_ID",
"value": "dataplane-signaling-api"
},
{
"key": "PROVIDER_HOST",
"value": "http://localhost:8080"
},
{
"key": "DATAFLOW_ID",
"value": "test-flow-1"
},
{
"key": "CONSUMER_HOST",
"value": "http://localhost:8081"
}
]
}
Loading
Loading