Testing processes

Task methods, gateway conditions, DMN decisions and service handlers are ordinary Go functions. Test them with go test. For an entire process, use the real queue and engine with a dedicated PostgreSQL test database.

Level	Checks	Database
Generation	Diagram, conditions, model types and signatures	No
Unit tests	Methods, conditions, decisions, stubbed handlers	No
End-to-end	Transitions, tasks, retries and instance outcomes	PostgreSQL

Examples below are included from the site's orders example, which CI builds and tests. Original source comments and assertion messages may be in Russian.

Data model and dependencies#

The order process validates an order, sends large orders to manager approval, then notifies the customer. A mailer interface makes the external dependency replaceable.

// orders/internal/bpmgen/model/data_model.go
// DataModel - переменные процессов. С ней работают методы и скрипты задач, обработчики jobType,
// условия шлюзов и решения DMN: любая переменная, которую они используют, должна быть полем модели.
// Поле получает переменную процесса по тегу json, например:
//
//	Amount float64 `json:"amount"`
//
// Вложенные структуры дают пути: client.account.number в BPMN и DMN - поле data.Client.Account.Number.
// После метода поля модели записываются поверх переменных процесса. Поле с omitempty
// в нулевом значении не меняет переменную
type DataModel struct {
	Amount   float64 `json:"amount"`
	Email    string  `json:"email"`
	Checked  bool    `json:"checked"`
	Notified bool    `json:"notified"`
}

// orders/internal/bpmgen/services/services.go
// Services - внешние сервисы, которые получают методы task и обработчики jobType.
// Заполняются при старте и передаются в сгенерированные Processes и Queues.
// Поля удобнее делать интерфейсами, чтобы в тестах подставлять заглушки
type Services struct {
	Mailer Mailer
}

// Mailer - отправка писем клиентам
type Mailer interface {
	Send(ctx context.Context, to, subject, body string) error
}

Test a task method#

Call the method with a model and inspect its fields; no engine or database is needed.

// orders/internal/bpmgen/order_approval/method_check_order.go
// CheckOrder - метод задачи CheckOrder «Проверить заказ»
func CheckOrder(ctx context.Context, svc *services.Services, data *model.DataModel) error {
	if data.Amount <= 0 {
		return fmt.Errorf("сумма заказа должна быть положительной: %v", data.Amount)
	}
	data.Checked = true
	return nil
}

// orders/internal/bpmgen/order_approval/method_check_order_test.go
func TestCheckOrder(t *testing.T) {
	ctx, svc := context.Background(), &services.Services{}

	data := &model.DataModel{Amount: 500}
	if err := CheckOrder(ctx, svc, data); err != nil {
		t.Fatal(err)
	}
	if !data.Checked {
		t.Error("заказ не отмечен проверенным")
	}
	if err := CheckOrder(ctx, svc, &model.DataModel{}); err == nil {
		t.Error("пустой заказ прошёл проверку")
	}
}

Test a gateway condition#

The expression =amount > 1000 becomes a Go function. Test boundary cases from the same package because the function is unexported.

// orders/internal/bpmgen/order_approval/cond_is_big.go
// isBig возвращает id следующего элемента после IsBig
func isBig(data *model.DataModel) string {
	// в процессе: amount > 1000
	if data.Amount > 1000 {
		return "Flow_big"
	}
	return "Flow_small"
}

// orders/internal/bpmgen/order_approval/method_check_order_test.go
// Условие шлюза сгенерировано в Go-функцию isBig: её можно проверить из пакета процесса.
func TestIsBig(t *testing.T) {
	tests := []struct {
		amount float64
		want   string
	}{
		{amount: 1000, want: "Flow_small"},
		{amount: 1000.01, want: "Flow_big"},
		{amount: 5000, want: "Flow_big"},
	}
	for _, tt := range tests {
		if got := isBig(&model.DataModel{Amount: tt.amount}); got != tt.want {
			t.Errorf("isBig(%v) = %s, want %s", tt.amount, got, tt.want)
		}
	}
}

Test a service handler#

Use a fake mailer that records calls or returns an error. Construct bpm.Job with the fields the handler needs.

// orders/internal/bpmgen/handlers/notify_client.go
// NotifyClient - обработчик service task с jobType "notifyClient" во всех процессах, где он встречается
// (job.ProcessName, job.ElementId). Вызывается асинхронно из очереди задач и при ошибке повторяется,
// поэтому должен быть идемпотентным (ключ - job.InstanceUid + job.ElementId)
func NotifyClient(ctx context.Context, svc *services.Services, job bpm.Job, data *model.DataModel) error {
	body := fmt.Sprintf("Заказ на сумму %.2f принят", data.Amount)
	if err := svc.Mailer.Send(ctx, data.Email, "Заказ "+job.BusinessKey, body); err != nil {
		return fmt.Errorf("письмо клиенту: %w", err)
	}
	data.Notified = true
	return nil
}

// orders/internal/bpmgen/handlers/notify_client_test.go
package handlers

import (
	"context"
	"errors"
	"testing"

	"example.com/orders/internal/bpmgen/model"
	"example.com/orders/internal/bpmgen/services"
	"github.com/nicolas-vp/go-bpm/bpm"
)

// fakeMailer - заглушка services.Mailer: запоминает письма или возвращает заданную ошибку.
type fakeMailer struct {
	err  error
	sent []string
}

func (m *fakeMailer) Send(ctx context.Context, to, subject, body string) error {
	if m.err != nil {
		return m.err
	}
	m.sent = append(m.sent, to+": "+subject)
	return nil
}

func TestNotifyClient(t *testing.T) {
	mailer := &fakeMailer{}
	svc := &services.Services{Mailer: mailer}
	job := bpm.Job{InstanceUid: "01a17411-e69b-7ea7", ElementId: "Notify", BusinessKey: "order-2"}
	data := &model.DataModel{Amount: 500, Email: "client@example.com"}

	if err := NotifyClient(context.Background(), svc, job, data); err != nil {
		t.Fatal(err)
	}
	if !data.Notified || len(mailer.sent) != 1 || mailer.sent[0] != "client@example.com: Заказ order-2" {
		t.Errorf("notified = %v, sent = %v", data.Notified, mailer.sent)
	}
}

// Ошибка почты - ошибка обработчика: очередь повторит задачу, а клиент не отмечен уведомлённым.
func TestNotifyClientMailerDown(t *testing.T) {
	svc := &services.Services{Mailer: &fakeMailer{err: errors.New("smtp: connection refused")}}
	data := &model.DataModel{Amount: 500, Email: "client@example.com"}

	if err := NotifyClient(context.Background(), svc, bpm.Job{}, data); err == nil {
		t.Fatal("ошибка почты потерялась")
	}
	if data.Notified {
		t.Error("клиент отмечен уведомлённым без письма")
	}
}

Test a DMN decision#

Test both the output and matched rule IDs, which later appear in console history. Business rule methods can also be called directly: bpm.RecordDecision skips history outside an engine execution context.

// go-bpm/examples/demo/bpmgen/test_decision/decision_decision_1.go
// decideDecision1 - решение Decision_1 «Test Decision» из businessRule.dmn, hit policy UNIQUE.
// Возвращает значение выхода c первого подошедшего правила; если ни одно не подошло - нулевое значение и пустой список правил.
// Второе значение - id сработавших правил для истории
func decideDecision1(data *model.DataModel) (string, []string) {
	a := data.A // Некая переменная a
	b := data.B // Некая переменная b

	// DecisionRule_04e4bk8: "foo" | "bar" => "correct foobar"
	if a == "foo" && b == "bar" {
		return "correct foobar", []string{"DecisionRule_04e4bk8"}
	}
	// DecisionRule_1a2n073: "some" | - => "some a"
	if a == "some" {
		return "some a", []string{"DecisionRule_1a2n073"}
	}
	// DecisionRule_0t56a5i: - | "some" => "some b"
	if b == "some" {
		return "some b", []string{"DecisionRule_0t56a5i"}
	}
	// DecisionRule_1kb19kv: - | - => "default"
	return "default", []string{"DecisionRule_1kb19kv"}
}

// demo/decision_test.go
package test_decision

import (
	"context"
	"slices"
	"testing"

	"github.com/nicolas-vp/go-bpm/examples/demo/bpmgen/model"
	"github.com/nicolas-vp/go-bpm/examples/demo/bpmgen/services"
)

// Таблица Decision_1 (hit policy UNIQUE) сгенерирована в decideDecision1: каждое правило - ветка if.
func TestDecideDecision1(t *testing.T) {
	tests := []struct {
		a, b, want, rule string
	}{
		{a: "foo", b: "bar", want: "correct foobar", rule: "DecisionRule_04e4bk8"},
		{a: "some", b: "x", want: "some a", rule: "DecisionRule_1a2n073"},
		{a: "z", b: "some", want: "some b", rule: "DecisionRule_0t56a5i"},
		{a: "z", b: "z", want: "default", rule: "DecisionRule_1kb19kv"},
	}
	for _, tt := range tests {
		got, rules := decideDecision1(&model.DataModel{A: tt.a, B: tt.b})
		if got != tt.want || !slices.Equal(rules, []string{tt.rule}) {
			t.Errorf("a=%s, b=%s: %q %v, want %q [%s]", tt.a, tt.b, got, rules, tt.want, tt.rule)
		}
	}
}

// Задача бизнес-правила кладёт результат решения в модель. Вне движка запись в историю правил пропускается.
func TestBusinessRuleTask(t *testing.T) {
	data := &model.DataModel{A: "foo", B: "bar"}
	if err := Activity0xzljrt(context.Background(), &services.Services{}, data); err != nil {
		t.Fatal(err)
	}
	if data.C != "correct foobar" {
		t.Errorf("c = %q", data.C)
	}
}

Test the whole process#

Unit tests cannot establish that gateway routes, assignment groups and retries work together. End-to-end tests start the real engine and queue on PostgreSQL, with stubbed external services.

StartProcess only enqueues a start, so poll instance status or user tasks with a timeout rather than assuming immediate completion.

// orders/internal/e2e/order_test.go
func TestSmallOrder(t *testing.T) {
	env := start(t, &fakeMailer{})

	id, err := env.engine.StartProcess(env.ctx, "OrderApproval",
		map[string]any{"amount": 500, "email": "client@example.com"}, engine.WithBusinessKey("order-1"))
	if err != nil {
		t.Fatal(err)
	}
	// маленький заказ идёт мимо согласования сразу к уведомлению
	env.waitStatus(id, engine.InstanceCompleted)
	if sent := env.mailer.Sent(); len(sent) != 1 || sent[0] != "client@example.com: Заказ order-1" {
		t.Errorf("sent = %v", sent)
	}
}

func TestBigOrderNeedsApproval(t *testing.T) {
	env := start(t, &fakeMailer{})

	id, err := env.engine.StartProcess(env.ctx, "OrderApproval",
		map[string]any{"amount": 5000, "email": "client@example.com"}, engine.WithBusinessKey("order-2"))
	if err != nil {
		t.Fatal(err)
	}
	// крупный заказ ждёт менеджера: задача в очереди группы managers, письма ещё нет
	task := env.waitTask(id, "Approve")
	if len(env.mailer.Sent()) != 0 {
		t.Fatal("письмо ушло до согласования")
	}
	if err := env.tasks.Claim(env.ctx, task.TaskUid, "bob", []string{"accounting"}); !errors.Is(err, engine.ErrNotCandidate) {
		t.Errorf("взял задачу не из своей группы: err = %v", err)
	}
	if err := env.tasks.Claim(env.ctx, task.TaskUid, "bob", []string{"managers"}); err != nil {
		t.Fatal(err)
	}
	if err := env.tasks.Complete(env.ctx, task.TaskUid, "bob", nil); err != nil {
		t.Fatal(err)
	}
	env.waitStatus(id, engine.InstanceCompleted)
	if len(env.mailer.Sent()) != 1 {
		t.Errorf("sent = %v", env.mailer.Sent())
	}
}

func TestMailerDown(t *testing.T) {
	env := start(t, &fakeMailer{err: errors.New("smtp: connection refused")})

	id, err := env.engine.StartProcess(env.ctx, "OrderApproval", map[string]any{"amount": 500})
	if err != nil {
		t.Fatal(err)
	}
	// у notifyClient retries="3": после трёх неудачных попыток экземпляр упал
	env.waitStatus(id, engine.InstanceFailed)
}

func TestInvalidOrder(t *testing.T) {
	env := start(t, &fakeMailer{})

	id, err := env.engine.StartProcess(env.ctx, "OrderApproval", map[string]any{"amount": 0})
	if err != nil {
		t.Fatal(err)
	}
	// CheckOrder отвергает пустой заказ: шаг повторяется SystemMaxAttempts раз, затем экземпляр падает
	env.waitStatus(id, engine.InstanceFailed)
}

The environment uses two attempts and a 50 ms retry delay to make failure tests fast:

// orders/internal/e2e/order_test.go
// testEnv - движок процессов на тестовой базе.
type testEnv struct {
	t      *testing.T
	ctx    context.Context
	engine *engine.Engine
	tasks  *engine.UserTaskService
	reader *engine.HistoryReader
	mailer *fakeMailer
}

func start(t *testing.T, mailer *fakeMailer) *testEnv {
	t.Helper()
	ctx := context.Background()
	pool := testPool(t)
	log := slog.New(slog.DiscardHandler)
	on := func() bool { return true }

	svc := &services.Services{Mailer: mailer}
	processes := bpmgen.Processes(svc)
	q := queue.New(pool, queue.Options{PollInterval: 100 * time.Millisecond, Logger: log})
	e, err := engine.New(q, engine.Config{
		SystemMaxAttempts: 2,                     // упавший метод - быстро в failed
		RetryDelay:        50 * time.Millisecond, // повторы без долгих пауз
		History:           engine.HistoryConfig{ToDB: on},
		Logger:            log,
	}, processes, bpmgen.Queues(svc))
	if err != nil {
		t.Fatal(err)
	}
	q.Start(ctx)
	t.Cleanup(q.Stop)
	return &testEnv{
		t:      t,
		ctx:    ctx,
		engine: e,
		tasks:  engine.NewUserTaskService(pool, e),
		reader: engine.NewHistoryReader(pool, processes, bpmgen.Decisions),
		mailer: mailer,
	}
}

// testPool подключается к GO_BPM_TEST_DSN, применяет миграции и очищает таблицы движка.
func testPool(t *testing.T) *pgxpool.Pool {
	t.Helper()
	dsn := os.Getenv("GO_BPM_TEST_DSN")
	if dsn == "" {
		t.Skip("GO_BPM_TEST_DSN не задан")
	}
	ctx := context.Background()
	pool, err := pgxpool.New(ctx, dsn)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(pool.Close)
	if err := migrations.Up(ctx, pool); err != nil {
		t.Fatal(err)
	}
	_, err = pool.Exec(ctx, `truncate bpm_queue, bpm_parallel_sync, bpm_wait, bpm_history, bpm_rules_history,
		bpm_process_instance, bpm_user_task`)
	if err != nil {
		t.Fatal(err)
	}
	return pool
}

The mailer fake uses a mutex because handlers execute in worker goroutines:

// orders/internal/e2e/order_test.go
// waitStatus ждёт, пока экземпляр придёт в статус want; процесс идёт асинхронно, через очередь.
func (env *testEnv) waitStatus(id uuid.UUID, want string) {
	env.t.Helper()
	deadline := time.Now().Add(10 * time.Second)
	for {
		instance, err := env.reader.GetInstance(env.ctx, id)
		if err != nil {
			env.t.Fatal(err)
		}
		if instance.Status == want {
			return
		}
		if instance.Status != engine.InstanceRunning || time.Now().After(deadline) {
			env.t.Fatalf("экземпляр %s: статус %s, ждали %s", id, instance.Status, want)
		}
		time.Sleep(50 * time.Millisecond)
	}
}

// waitTask ждёт открытую задачу пользователя на элементе elementId.
func (env *testEnv) waitTask(id uuid.UUID, elementId string) engine.UserTask {
	env.t.Helper()
	deadline := time.Now().Add(10 * time.Second)
	for {
		tasks, err := env.tasks.ListTasks(env.ctx, engine.TaskFilter{InstanceUid: id})
		if err != nil {
			env.t.Fatal(err)
		}
		for _, task := range tasks {
			if task.ElementId == elementId {
				return task
			}
		}
		if time.Now().After(deadline) {
			env.t.Fatalf("экземпляр %s: нет задачи %s, открыты %+v", id, elementId, tasks)
		}
		time.Sleep(50 * time.Millisecond)
	}
}

// fakeMailer - заглушка services.Mailer. Обработчик вызывают воркеры очереди, поэтому под мьютексом.
type fakeMailer struct {
	err  error
	mu   sync.Mutex
	sent []string
}

func (m *fakeMailer) Send(ctx context.Context, to, subject, body string) error {
	if m.err != nil {
		return m.err
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	m.sent = append(m.sent, to+": "+subject)
	return nil
}

func (m *fakeMailer) Sent() []string {
	m.mu.Lock()
	defer m.mu.Unlock()
	return append([]string(nil), m.sent...)
}

Run the example#

Without GO_BPM_TEST_DSN, database tests skip. From examples/orders:

docker run -d --name orders-test-pg -e POSTGRES_PASSWORD=postgres -e POSTGRES_DB=orders_test -p 5433:5432 postgres:18
GO_BPM_TEST_DSN='postgres://postgres:postgres@localhost:5433/orders_test?sslmode=disable' go test ./...

The database is cleared before tests. Use a dedicated test database. Tests share tables, so avoid t.Parallel() and use go test -p 1 if database tests span packages.

CI and engine tests#

In CI, provide a PostgreSQL service and set GO_BPM_TEST_DSN, then run generation, git diff --exit-code and tests. The diff check catches diagrams changed without regenerating code.

The engine repository includes integration coverage for demo processes, user task claim/unclaim/complete, transactional starts, boundary timers and errors, decisions, call activities and generator validation. See running and operating for commands.