test(webhook): e2e dispatcher → embedded HttpServer + HMAC verify
Закрывает Phase 5 deferred item. Завершает webhooks v2 покрытие. WebhookE2ETest (2 tests): Test 1: success path - Стартует JDK HttpServer на random localhost port (no extra deps) - Создаёт WebhookSubscription указывающий на /hook - POST record в dictionary → outbox event - Ждёт что receiver получит POST в течение 20s - Verify body содержит business_key - Verify HMAC signature header валиден (HmacSigner.sign(secret, body) matches X-Ordinis-Signature) - Verify Ordinis headers (X-Ordinis-Event-Type, X-Ordinis-Scope=PUBLIC) - Verify DB delivery row.status = delivered + lastStatusCode = 200 Test 2: retry path - Receiver всегда 500 - POST record → outbox → dispatcher tries deliver - Verify delivery transitions to status=retrying + lastStatusCode=500 - Не ждём DLQ (1000 attempts × 1m backoff = недели) SSRF guard через ordinis.webhook.allowed-hosts=localhost,127.0.0.1 test override (production deny private CIDR остаётся включённым). Test config: - ordinis.outbox.enabled=true (для outbox poller) - ordinis.webhook.enabled=true (для dispatcher) - match-interval-ms=200, send-interval-ms=200 (faster tests) Webhooks v2 — все 5 phases закрыты.
This commit is contained in:
Zimin A.N.
+239
@@ -0,0 +1,239 @@
|
||||
package cloud.nstart.terravault.ordinis.app.e2e;
|
||||
|
||||
import cloud.nstart.terravault.ordinis.domain.outbox.OutboxEventRepository;
|
||||
import cloud.nstart.terravault.ordinis.domain.webhook.WebhookDelivery;
|
||||
import cloud.nstart.terravault.ordinis.domain.webhook.WebhookDeliveryRepository;
|
||||
import cloud.nstart.terravault.ordinis.domain.webhook.WebhookSubscription;
|
||||
import cloud.nstart.terravault.ordinis.domain.webhook.WebhookSubscriptionRepository;
|
||||
import cloud.nstart.terravault.ordinis.outbox.webhook.HmacSigner;
|
||||
import com.fasterxml.jackson.databind.JsonNode;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.sun.net.httpserver.HttpServer;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
import org.springframework.http.MediaType;
|
||||
import org.springframework.test.context.ActiveProfiles;
|
||||
import org.springframework.test.context.DynamicPropertyRegistry;
|
||||
import org.springframework.test.context.DynamicPropertySource;
|
||||
import org.springframework.test.web.servlet.MockMvc;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.time.Duration;
|
||||
import java.util.List;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.ConcurrentLinkedQueue;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.awaitility.Awaitility.await;
|
||||
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
|
||||
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
|
||||
|
||||
/**
|
||||
* End-to-end webhook delivery test:
|
||||
* <ol>
|
||||
* <li>Стартуем embedded {@link HttpServer} на random localhost port —
|
||||
* imitate consumer's webhook receiver.</li>
|
||||
* <li>Создаём WebhookSubscription указывающий на этот endpoint.</li>
|
||||
* <li>POST a record — попадает в outbox_events.</li>
|
||||
* <li>WebhookDispatcher должен matchTick → INSERT delivery →
|
||||
* sendTick → POST на наш HttpServer → mark delivered.</li>
|
||||
* <li>Проверяем: receiver получил body + valid HMAC signature +
|
||||
* Ordinis headers; delivery row mark'нут как delivered.</li>
|
||||
* </ol>
|
||||
*
|
||||
* <p>SSRF guard разрешает localhost через {@code allowed-hosts} config.
|
||||
*/
|
||||
@SpringBootTest(
|
||||
webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
|
||||
properties = {
|
||||
"ordinis.outbox.enabled=true",
|
||||
"ordinis.outbox.poll-interval-ms=200",
|
||||
"ordinis.outbox.batch-size=20",
|
||||
"ordinis.webhook.enabled=true",
|
||||
"ordinis.webhook.match-interval-ms=200",
|
||||
"ordinis.webhook.send-interval-ms=200",
|
||||
"ordinis.webhook.allowed-hosts=localhost,127.0.0.1",
|
||||
})
|
||||
@AutoConfigureMockMvc
|
||||
@ActiveProfiles("test")
|
||||
class WebhookE2ETest {
|
||||
|
||||
@DynamicPropertySource
|
||||
static void props(DynamicPropertyRegistry registry) {
|
||||
E2ESupport.registerProperties(registry);
|
||||
}
|
||||
|
||||
@Autowired MockMvc mvc;
|
||||
@Autowired ObjectMapper om;
|
||||
@Autowired OutboxEventRepository outboxRepo;
|
||||
@Autowired WebhookSubscriptionRepository subRepo;
|
||||
@Autowired WebhookDeliveryRepository deliveryRepo;
|
||||
|
||||
private HttpServer receiver;
|
||||
private final ConcurrentLinkedQueue<ReceivedRequest> received = new ConcurrentLinkedQueue<>();
|
||||
|
||||
record ReceivedRequest(String path, String body, String signatureHeader, String eventTypeHeader, String scopeHeader) {}
|
||||
|
||||
@BeforeEach
|
||||
void startReceiver() throws IOException {
|
||||
receiver = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
|
||||
receiver.createContext("/hook", exchange -> {
|
||||
byte[] body = exchange.getRequestBody().readAllBytes();
|
||||
received.add(new ReceivedRequest(
|
||||
exchange.getRequestURI().getPath(),
|
||||
new String(body, StandardCharsets.UTF_8),
|
||||
exchange.getRequestHeaders().getFirst(HmacSigner.HEADER_NAME),
|
||||
exchange.getRequestHeaders().getFirst("X-Ordinis-Event-Type"),
|
||||
exchange.getRequestHeaders().getFirst("X-Ordinis-Scope")));
|
||||
exchange.sendResponseHeaders(200, -1);
|
||||
exchange.close();
|
||||
});
|
||||
receiver.start();
|
||||
}
|
||||
|
||||
@AfterEach
|
||||
void stopReceiver() {
|
||||
if (receiver != null) receiver.stop(0);
|
||||
}
|
||||
|
||||
@Test
|
||||
void recordCreate_dispatchedToWebhookReceiver_withValidHmac() throws Exception {
|
||||
int port = receiver.getAddress().getPort();
|
||||
String hookUrl = "http://127.0.0.1:" + port + "/hook";
|
||||
|
||||
// 1. Создаём dict
|
||||
String dictName = "wh_" + UUID.randomUUID().toString().replace("-", "").substring(0, 8);
|
||||
JsonNode schema = om.readTree("""
|
||||
{"$schema":"http://json-schema.org/draft-07/schema#","type":"object",
|
||||
"additionalProperties":false,"x-id-source":"code","required":["code"],
|
||||
"properties":{"code":{"type":"string","x-unique":true}}}""");
|
||||
var dictBody = om.createObjectNode()
|
||||
.put("name", dictName)
|
||||
.put("scope", "PUBLIC")
|
||||
.put("schemaVersion", "1.0.0")
|
||||
.put("bundle", "test");
|
||||
dictBody.set("schemaJson", schema);
|
||||
mvc.perform(post("/api/v1/dictionaries")
|
||||
.contentType(MediaType.APPLICATION_JSON)
|
||||
.content(om.writeValueAsBytes(dictBody)))
|
||||
.andExpect(status().is2xxSuccessful());
|
||||
|
||||
// 2. Создаём webhook subscription напрямую через repository (CRUD endpoint
|
||||
// требует RESTRICTED scope от JWT, в тесте без auth — проще через repo)
|
||||
String secret = "test-secret-" + UUID.randomUUID();
|
||||
WebhookSubscription sub = new WebhookSubscription(
|
||||
"e2e-test-receiver", hookUrl, secret, "test-user");
|
||||
sub = subRepo.save(sub);
|
||||
UUID subId = sub.getId();
|
||||
|
||||
long deliveriesBefore = deliveryRepo.count();
|
||||
|
||||
// 3. POST record → outbox_events row created
|
||||
String needle = "EVT_" + UUID.randomUUID().toString().replace("-", "").substring(0, 12).toUpperCase();
|
||||
var recBody = om.createObjectNode().put("businessKey", needle);
|
||||
recBody.set("data", om.createObjectNode().put("code", needle));
|
||||
mvc.perform(post("/api/v1/dictionaries/{n}/records", dictName)
|
||||
.contentType(MediaType.APPLICATION_JSON)
|
||||
.content(om.writeValueAsBytes(recBody)))
|
||||
.andExpect(status().is2xxSuccessful());
|
||||
|
||||
// 4. Wait for delivery: receiver получил POST request
|
||||
await().atMost(20, TimeUnit.SECONDS)
|
||||
.pollInterval(Duration.ofMillis(300))
|
||||
.until(() -> received.stream().anyMatch(r -> r.body().contains(needle)));
|
||||
|
||||
// 5. Verify request properties
|
||||
ReceivedRequest req = received.stream()
|
||||
.filter(r -> r.body().contains(needle))
|
||||
.findFirst()
|
||||
.orElseThrow();
|
||||
|
||||
assertThat(req.path()).isEqualTo("/hook");
|
||||
assertThat(req.scopeHeader()).isEqualTo("PUBLIC");
|
||||
assertThat(req.eventTypeHeader()).isNotBlank();
|
||||
|
||||
// HMAC signature валиден: receiver мог бы verify через свой secret
|
||||
String expectedSig = HmacSigner.sign(secret, req.body().getBytes(StandardCharsets.UTF_8));
|
||||
assertThat(req.signatureHeader()).isEqualTo(expectedSig);
|
||||
|
||||
// 6. Verify DB delivery row marked delivered
|
||||
await().atMost(10, TimeUnit.SECONDS)
|
||||
.pollInterval(Duration.ofMillis(300))
|
||||
.until(() -> deliveryRepo.count() > deliveriesBefore);
|
||||
|
||||
List<WebhookDelivery> all = deliveryRepo.findBySubscriptionIdOrderByCreatedAtDesc(
|
||||
subId, org.springframework.data.domain.PageRequest.of(0, 10)).getContent();
|
||||
assertThat(all).isNotEmpty();
|
||||
WebhookDelivery latest = all.get(0);
|
||||
await().atMost(10, TimeUnit.SECONDS)
|
||||
.pollInterval(Duration.ofMillis(300))
|
||||
.until(() -> {
|
||||
WebhookDelivery d = deliveryRepo.findById(latest.getId()).orElseThrow();
|
||||
return d.getStatus() == WebhookDelivery.Status.delivered;
|
||||
});
|
||||
|
||||
WebhookDelivery delivered = deliveryRepo.findById(latest.getId()).orElseThrow();
|
||||
assertThat(delivered.getStatus()).isEqualTo(WebhookDelivery.Status.delivered);
|
||||
assertThat(delivered.getLastStatusCode()).isEqualTo(200);
|
||||
assertThat(delivered.getDeliveredAt()).isNotNull();
|
||||
assertThat(delivered.getLastError()).isNull();
|
||||
}
|
||||
|
||||
@Test
|
||||
void receiverReturning500_deliveryRetried_notDelivered() throws Exception {
|
||||
// Стартуем ALT receiver который всегда 500
|
||||
HttpServer failing = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
|
||||
failing.createContext("/fail", exchange -> {
|
||||
byte[] body = exchange.getRequestBody().readAllBytes(); // drain
|
||||
assertThat(body.length).isGreaterThan(0);
|
||||
exchange.sendResponseHeaders(500, -1);
|
||||
exchange.close();
|
||||
});
|
||||
failing.start();
|
||||
try {
|
||||
int port = failing.getAddress().getPort();
|
||||
String hookUrl = "http://127.0.0.1:" + port + "/fail";
|
||||
|
||||
// dict
|
||||
String dictName = "whf_" + UUID.randomUUID().toString().replace("-", "").substring(0, 8);
|
||||
JsonNode schema = om.readTree("""
|
||||
{"$schema":"http://json-schema.org/draft-07/schema#","type":"object",
|
||||
"additionalProperties":false,"x-id-source":"code","required":["code"],
|
||||
"properties":{"code":{"type":"string","x-unique":true}}}""");
|
||||
var dictBody = om.createObjectNode()
|
||||
.put("name", dictName).put("scope", "PUBLIC")
|
||||
.put("schemaVersion", "1.0.0").put("bundle", "test");
|
||||
dictBody.set("schemaJson", schema);
|
||||
mvc.perform(post("/api/v1/dictionaries").contentType(MediaType.APPLICATION_JSON)
|
||||
.content(om.writeValueAsBytes(dictBody))).andExpect(status().is2xxSuccessful());
|
||||
|
||||
WebhookSubscription sub = subRepo.save(new WebhookSubscription(
|
||||
"e2e-failing", hookUrl, "secret-fail", "test-user"));
|
||||
|
||||
String needle = "EVT_" + UUID.randomUUID().toString().replace("-", "").substring(0, 12);
|
||||
var recBody = om.createObjectNode().put("businessKey", needle);
|
||||
recBody.set("data", om.createObjectNode().put("code", needle));
|
||||
mvc.perform(post("/api/v1/dictionaries/{n}/records", dictName)
|
||||
.contentType(MediaType.APPLICATION_JSON)
|
||||
.content(om.writeValueAsBytes(recBody))).andExpect(status().is2xxSuccessful());
|
||||
|
||||
// Ждём что delivery с status=retrying появится (HTTP 500 → markFailed)
|
||||
await().atMost(20, TimeUnit.SECONDS)
|
||||
.pollInterval(Duration.ofMillis(300))
|
||||
.until(() -> deliveryRepo.findBySubscriptionIdOrderByCreatedAtDesc(
|
||||
sub.getId(), org.springframework.data.domain.PageRequest.of(0, 10))
|
||||
.getContent().stream()
|
||||
.anyMatch(d -> d.getStatus() == WebhookDelivery.Status.retrying
|
||||
&& d.getLastStatusCode() != null && d.getLastStatusCode() == 500));
|
||||
} finally {
|
||||
failing.stop(0);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user