From 5bd6402887021f98d54e9bbc16e4aecb2126c419 Mon Sep 17 00:00:00 2001 From: Ovsep Avakian Date: Thu, 23 Jul 2026 17:06:20 +0300 Subject: [PATCH] Restructure the project --- .agent/docs/design.md | 2 +- .agent/docs/hooks.md | 5 +- .agent/docs/testing.md | 4 +- component/activation.go | 2 +- component/component.go | 2 +- component/doc.go | 8 ++ component/hooks.go | 2 +- component/plugin.go | 10 +-- cycle/doc.go | 7 ++ doc.go | 10 +++ docs/wiki/999.-Internals.md | 2 +- example_test.go | 56 ++++++++++++ fmesh.go | 4 +- fmesh_test.go | 18 +--- hooks.go | 2 +- .../chainability/chainability_test.go | 35 +++----- .../component_hooks/basic_test.go | 58 +++++-------- integration_tests/computation/basic_test.go | 85 +++++-------------- integration_tests/constraints/time_test.go | 31 ++----- .../errorhandling/error_returns_test.go | 5 +- .../errorhandling/orphaned_component_test.go | 3 +- integration_tests/hooks/component_test.go | 19 +++-- integration_tests/hooks/fmesh_test.go | 51 +++++------ integration_tests/hooks/port_test.go | 69 ++++++--------- integration_tests/labels/transform_test.go | 30 ++----- integration_tests/meta/meta_test.go | 52 +++--------- .../fan_out_concurrency_integration_test.go | 17 ++-- integration_tests/piping/fan_test.go | 43 ++++------ integration_tests/ports/port_creation_test.go | 70 +++++---------- .../ports/waiting_for_inputs_test.go | 13 +-- integration_tests/state/basic_test.go | 31 ++----- {hook => internal/hook}/hook_group.go | 0 {hook => internal/hook}/hook_group_test.go | 0 internal/testutil/testutil.go | 61 +++++++++++++ meta/doc.go | 7 ++ port/doc.go | 8 ++ port/hooks.go | 6 +- port/port.go | 4 +- port/port_test.go | 3 + runtime.go | 8 +- 40 files changed, 399 insertions(+), 444 deletions(-) create mode 100644 component/doc.go create mode 100644 cycle/doc.go create mode 100644 doc.go create mode 100644 example_test.go rename {hook => internal/hook}/hook_group.go (100%) rename {hook => internal/hook}/hook_group_test.go (100%) create mode 100644 internal/testutil/testutil.go create mode 100644 meta/doc.go create mode 100644 port/doc.go diff --git a/.agent/docs/design.md b/.agent/docs/design.md index a7c57d4..1e61c9d 100644 --- a/.agent/docs/design.md +++ b/.agent/docs/design.md @@ -25,7 +25,7 @@ Architecture overview (the concept → type → package table and the execution - **`meta`** — `Labels` (string k/v) and `Scalars` (string→float64). `Keys()`/`Values()` return sorted slices for determinism. `Merge(other)` is the one non-mutating method on both types. `Every(pred)` on empty = `true` (vacuous truth). `ForEach` returns `error`. Constructors: `NewLabels()`, `NewScalars()`. - **`port`** — `Flush()` fans out then clears source. `PipeTo` is output→input only. Both return `error`. `PipeTo` validates direction at call time. - **`component`** — `State` is `map[string]any`, persistent across cycles and across `Run`s (see [runtime.md](runtime.md)). Constructors use functional options: `component.New(name, opts...) (*Component, error)`. Ports come in two creation styles: name-based (`WithInputs`/`AddInputs`, `WithIndexedInputs("i", 1, 3)` → `i1..i3`) and attach-based (`AttachInputPorts` for pre-built `port.NewInput` ports with options). `LoopbackPipe(out, in)` wires a component to itself (such a mesh never stops naturally). `ErrWaitingForInputs`/`ErrWaitingForInputsKeep` are scheduler control-flow sentinels, not failures. -- **`hook`** — generic `hook.Group[T]`, ordered, fail-fast `Trigger`. Three hook levels (mesh/component/port); see [hooks.md](hooks.md). +- **`hook`** — lives at `internal/hook` (not public API). Generic `hook.Group[T]`, ordered, fail-fast `Trigger`. Three hook levels (mesh/component/port); see [hooks.md](hooks.md). - **`cycle`** — has its own `Any`/`Every`/`Count` on its collection type, independent of `signal.Group`. ## Metadata tiers on groups/collections diff --git a/.agent/docs/hooks.md b/.agent/docs/hooks.md index e6fb638..aac38ec 100644 --- a/.agent/docs/hooks.md +++ b/.agent/docs/hooks.md @@ -1,10 +1,11 @@ # Hooks & plugins — extension points -Source: `hooks.go`, `component/hooks.go`, `port/hooks.go`, `hook/hook_group.go`, `component/plugin.go`. +Source: `hooks.go`, `component/hooks.go`, `port/hooks.go`, `internal/hook/hook_group.go`, `component/plugin.go`. ## The hook primitive -`hook.Group[T]` (package `hook`) — the **one approved generic type** in the codebase. An ordered +`hook.Group[T]` (package `internal/hook` — not part of the public API) — the **one approved +generic type** in the codebase. An ordered slice of `func(T) error`; `Trigger(arg)` runs all in insertion order, **fail-fast** on first error. Registration is chainable and happens through `SetupHooks(func(*Hooks))` closures (or the `WithHooks` constructor option on components) — the `Hooks` structs' fields are unexported, so diff --git a/.agent/docs/testing.md b/.agent/docs/testing.md index 37d14d8..f564219 100644 --- a/.agent/docs/testing.md +++ b/.agent/docs/testing.md @@ -6,7 +6,9 @@ - Table-driven by default; `t.Run` subtests for grouped inline assertions - `require` for preconditions and error checks (stops on failure); `assert` for value checks (continues) - No assertion helpers — use plain `assert`/`require` directly -- Only allowed helper: `mustXxx()` panic-on-error for fixture setup, never for assertions +- Only allowed helper: `mustXxx()` panic-on-error for fixture setup, never for assertions. + Shared implementations live in `internal/testutil` (usable from `integration_tests/` and other + external test packages); in-package tests of `fmesh` and `port` keep local copies (import cycle) - Use `assert.InDelta` for float64 comparisons (tolerance `1e-9` for exact values, larger for computed averages) ## What to cover diff --git a/component/activation.go b/component/activation.go index de4c90d..c492b5e 100644 --- a/component/activation.go +++ b/component/activation.go @@ -5,7 +5,7 @@ import ( "fmt" "runtime/debug" - "github.com/hovsep/fmesh/hook" + "github.com/hovsep/fmesh/internal/hook" ) // WithActivationFunc is a component option that sets the activation function. diff --git a/component/component.go b/component/component.go index 5ce2479..cb142be 100644 --- a/component/component.go +++ b/component/component.go @@ -23,7 +23,7 @@ type Component struct { state State parentMesh ParentMesh hooks *Hooks - plugins Plugins + plugins plugins } // New creates a new component with the given name and options. diff --git a/component/doc.go b/component/doc.go new file mode 100644 index 0000000..386e45c --- /dev/null +++ b/component/doc.go @@ -0,0 +1,8 @@ +// Package component provides [Component], the building block of a mesh. +// +// A component owns named input and output ports and an activation function +// that runs once per cycle when inputs are ready. Components are built with +// [New] and functional options (WithInputs, WithActivationFunc, ...); after a +// run, each cycle exposes an [ActivationResult] per component describing +// whether and how it activated. Component types mutate in place. +package component diff --git a/component/hooks.go b/component/hooks.go index d8dec7d..209e1d8 100644 --- a/component/hooks.go +++ b/component/hooks.go @@ -1,7 +1,7 @@ package component import ( - "github.com/hovsep/fmesh/hook" + "github.com/hovsep/fmesh/internal/hook" ) // ActivationContext provides context for activation hooks. diff --git a/component/plugin.go b/component/plugin.go index 1dd08b2..974ad3e 100644 --- a/component/plugin.go +++ b/component/plugin.go @@ -8,12 +8,12 @@ type Plugin interface { Init(*Component) error } -// Plugins defines a container of component plugins. -type Plugins map[string]Plugin +// plugins defines a container of component plugins. +type plugins map[string]Plugin -// newPlugins is a constructor for Plugins. -func newPlugins() Plugins { - return make(Plugins) +// newPlugins is a constructor for plugins. +func newPlugins() plugins { + return make(plugins) } // WithPlugins is a component constructor option that adds plugins. diff --git a/cycle/doc.go b/cycle/doc.go new file mode 100644 index 0000000..acef74e --- /dev/null +++ b/cycle/doc.go @@ -0,0 +1,7 @@ +// Package cycle provides [Cycle], one synchronized execution tick of a mesh, +// and [Group], the bounded history of cycles kept during a run. +// +// A cycle collects the activation results of every component that was +// considered in that tick; inspect them after a run through the mesh's +// runtime info. Cycle types mutate in place. +package cycle diff --git a/doc.go b/doc.go new file mode 100644 index 0000000..d4f7f2a --- /dev/null +++ b/doc.go @@ -0,0 +1,10 @@ +// Package fmesh orchestrates a mesh of components exchanging signals through +// ports, following the Flow-Based Programming model. +// +// Build a mesh with [New], add components, and start it with [FMesh.Run]. +// Execution proceeds in discrete synchronized cycles: every cycle all ready +// components activate concurrently, then outputs are flushed through pipes to +// downstream inputs. The mesh stops naturally when no component activates in a +// cycle, or on the cycle limit, time limit, or error-handling strategy +// configured via [Config] and the With* options. +package fmesh diff --git a/docs/wiki/999.-Internals.md b/docs/wiki/999.-Internals.md index c719256..3bd16cc 100644 --- a/docs/wiki/999.-Internals.md +++ b/docs/wiki/999.-Internals.md @@ -126,7 +126,7 @@ Pre-run checks (return `error` on failure): ## hook.Group -The `hook` package provides `hook.Group[T]` — a generic ordered hook collection used by `fmesh.Hooks`, `component.Hooks`, and `port.Hooks`: +The `internal/hook` package provides `hook.Group[T]` — a generic ordered hook collection used by `fmesh.Hooks`, `component.Hooks`, and `port.Hooks`. It is internal machinery, not importable by user code — hooks are registered through the closure-based API shown in [Hooks](https://github.com/hovsep/fmesh/wiki/501.-Hooks): ```go g := hook.NewGroup[*MyContext]() diff --git a/example_test.go b/example_test.go new file mode 100644 index 0000000..d5a0fd0 --- /dev/null +++ b/example_test.go @@ -0,0 +1,56 @@ +package fmesh_test + +import ( + "fmt" + "strings" + + "github.com/hovsep/fmesh" + "github.com/hovsep/fmesh/component" + "github.com/hovsep/fmesh/signal" +) + +// Example mirrors the README quick-start: two components connected by a pipe, +// run to completion in discrete cycles. +func Example() { + must := func(err error) { + if err != nil { + panic(err) + } + } + + concat, err := component.New("concat", + component.WithInputs("i1", "i2"), + component.WithOutputs("res"), + component.WithActivationFunc(func(this *component.Component) error { + word1 := this.InputByName("i1").Signals().FirstPayloadOrDefault("").(string) + word2 := this.InputByName("i2").Signals().FirstPayloadOrDefault("").(string) + return this.OutputByName("res").PutSignals(signal.New(word1 + word2)) + })) + must(err) + + uppercase, err := component.New("uppercase", + component.WithInputs("i1"), + component.WithOutputs("res"), + component.WithActivationFunc(func(this *component.Component) error { + str := this.InputByName("i1").Signals().FirstPayloadOrDefault("").(string) + return this.OutputByName("res").PutSignals(signal.New(strings.ToUpper(str))) + })) + must(err) + + fm, err := fmesh.New("hello world") + must(err) + must(fm.AddComponents(concat, uppercase)) + + must(concat.OutputByName("res").PipeTo(uppercase.InputByName("i1"))) + + must(concat.InputByName("i1").PutSignals(signal.New("hello "))) + must(concat.InputByName("i2").PutSignals(signal.New("world!"))) + + _, err = fm.Run() + must(err) + + result, err := uppercase.OutputByName("res").Signals().FirstPayload() + must(err) + fmt.Printf("Result: %v\n", result) + // Output: Result: HELLO WORLD! +} diff --git a/fmesh.go b/fmesh.go index a57d360..dfa9a31 100644 --- a/fmesh.go +++ b/fmesh.go @@ -322,7 +322,7 @@ func (fm *FMesh) cleanUpPreviousRun() error { // Init runtime info fm.runtimeInfo = newRuntimeInfo(fm.config.CyclesHistoryLimit) - fm.runtimeInfo.MarkStarted() + fm.runtimeInfo.markStarted() return nil } @@ -335,7 +335,7 @@ func (fm *FMesh) Run() (ri *RuntimeInfo, runErr error) { ri = fm.runtimeInfo defer func() { - fm.runtimeInfo.MarkStopped() + fm.runtimeInfo.markStopped() if err := fm.hooks.afterRun.Trigger(fm); err != nil { if runErr == nil { runErr = fmt.Errorf("afterRun hook failed: %w", err) diff --git a/fmesh_test.go b/fmesh_test.go index 6a98ebd..c18dd21 100644 --- a/fmesh_test.go +++ b/fmesh_test.go @@ -18,6 +18,8 @@ import ( var noOpActivationFunc = func(this *component.Component) error { return nil } +// Local helpers: this in-package test cannot import internal/testutil +// (testutil imports fmesh). func mustNewFMesh(name string, opts ...Option) *FMesh { fm, err := New(name, opts...) if err != nil { @@ -34,20 +36,6 @@ func mustNewComponent(name string, opts ...component.Option) *component.Componen return c } -func mustAddInputs(c *component.Component, names ...string) *component.Component { - if err := c.AddInputs(names...); err != nil { - panic(err) - } - return c -} - -func mustAddOutputs(c *component.Component, names ...string) *component.Component { - if err := c.AddOutputs(names...); err != nil { - panic(err) - } - return c -} - func mustPipeTo(src *port.Port, dsts ...*port.Port) { if err := src.PipeTo(dsts...); err != nil { panic(err) @@ -952,7 +940,7 @@ func TestFMesh_runCycle(t *testing.T) { if tt.initFM != nil { tt.initFM(fm) } - fm.runtimeInfo.MarkStarted() + fm.runtimeInfo.markStarted() cycleErr := fm.runCycle() gotCycleResult := fm.runtimeInfo.Cycles.Last() if tt.wantError { diff --git a/hooks.go b/hooks.go index 0495b1a..967ed11 100644 --- a/hooks.go +++ b/hooks.go @@ -5,7 +5,7 @@ import ( "github.com/hovsep/fmesh/component" "github.com/hovsep/fmesh/cycle" - "github.com/hovsep/fmesh/hook" + "github.com/hovsep/fmesh/internal/hook" "github.com/hovsep/fmesh/port" ) diff --git a/integration_tests/chainability/chainability_test.go b/integration_tests/chainability/chainability_test.go index b301d0b..54f3432 100644 --- a/integration_tests/chainability/chainability_test.go +++ b/integration_tests/chainability/chainability_test.go @@ -3,33 +3,20 @@ package chainability import ( "testing" + "github.com/hovsep/fmesh/internal/testutil" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/hovsep/fmesh/component" "github.com/hovsep/fmesh/port" "github.com/hovsep/fmesh/signal" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" ) -func mustComponent(name string, opts ...component.Option) *component.Component { - c, err := component.New(name, opts...) - if err != nil { - panic(err) - } - return c -} - -func mustInputPort(name string, opts ...port.Option) *port.Port { - p, err := port.NewInput(name, opts...) - if err != nil { - panic(err) - } - return p -} - // TestChainability_CrossPackage verifies realistic cross-package chaining scenarios. func TestChainability_CrossPackage(t *testing.T) { t.Run("full component setup", func(t *testing.T) { - c := mustComponent("processor", + c := testutil.MustComponent("processor", component.WithInputs("in1", "in2"), component.WithOutputs("out1", "out2"), component.WithDescription("main processor"), @@ -50,7 +37,7 @@ func TestChainability_CrossPackage(t *testing.T) { }) t.Run("port with signals and labels", func(t *testing.T) { - p := mustInputPort("data", port.WithDescription("data input")). + p := testutil.MustInputPort("data", port.WithDescription("data input")). AddLabel("type", "data") require.NoError(t, p.PutSignals(signal.New(1), signal.New(2))) p.AddLabel("count", "2") @@ -75,7 +62,7 @@ func TestChainability_CrossPackage(t *testing.T) { t.Run("component with label cleanup", func(t *testing.T) { // Simulate component lifecycle: setup with debug labels, then clean them up - c := mustComponent("worker", + c := testutil.MustComponent("worker", component.WithInputs("tasks"), component.WithOutputs("results"), component.WithDescription("background worker"), @@ -97,7 +84,7 @@ func TestChainability_CrossPackage(t *testing.T) { t.Run("port with label reset workflow", func(t *testing.T) { // Port initially configured with temporary setup labels, then cleared for production - p := mustInputPort("input"). + p := testutil.MustInputPort("input"). AddLabels(map[string]string{ "setup": "true", "test": "mode", @@ -144,7 +131,7 @@ func TestChainability_CrossPackage(t *testing.T) { t.Run("complex label lifecycle", func(t *testing.T) { // Realistic scenario: component setup -> debug -> cleanup -> finalize - c := mustComponent("api-handler", + c := testutil.MustComponent("api-handler", component.WithInputs("request"), component.WithOutputs("response", "errors"), component.WithDescription("HTTP API handler"), @@ -183,7 +170,7 @@ func TestChainability_CrossPackage(t *testing.T) { WithNoLabels(). WithLabels(map[string]string{"priority": "high", "source": "validated"}) - p := mustInputPort("validated-input"). + p := testutil.MustInputPort("validated-input"). AddLabel("type", "input") require.NoError(t, p.PutSignals(s1, s2)) p.AddLabel("count", "2"). diff --git a/integration_tests/component_hooks/basic_test.go b/integration_tests/component_hooks/basic_test.go index de9cce4..4dca9dc 100644 --- a/integration_tests/component_hooks/basic_test.go +++ b/integration_tests/component_hooks/basic_test.go @@ -5,33 +5,19 @@ import ( "sync" "testing" - "github.com/hovsep/fmesh" - "github.com/hovsep/fmesh/component" - "github.com/hovsep/fmesh/signal" + "github.com/hovsep/fmesh/internal/testutil" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" -) - -func mustComponent(name string, opts ...component.Option) *component.Component { - c, err := component.New(name, opts...) - if err != nil { - panic(err) - } - return c -} -func mustFMesh(name string, opts ...fmesh.Option) *fmesh.FMesh { - fm, err := fmesh.New(name, opts...) - if err != nil { - panic(err) - } - return fm -} + "github.com/hovsep/fmesh/component" + "github.com/hovsep/fmesh/signal" +) func TestComponentHooks_AllTypes(t *testing.T) { var executionLog []string - c := mustComponent("processor", + c := testutil.MustComponent("processor", component.WithInputs("in"), component.WithOutputs("out"), component.WithActivationFunc(func(c *component.Component) error { @@ -67,7 +53,7 @@ func TestComponentHooks_OnError(t *testing.T) { var afterFired bool testErr := errors.New("test error") - c := mustComponent("processor", + c := testutil.MustComponent("processor", component.WithInputs("in"), component.WithActivationFunc(func(c *component.Component) error { return testErr @@ -99,7 +85,7 @@ func TestComponentHooks_OnPanic(t *testing.T) { var panicCaught bool var afterFired bool - c := mustComponent("processor", + c := testutil.MustComponent("processor", component.WithInputs("in"), component.WithActivationFunc(func(c *component.Component) error { panic("oh no!") @@ -130,7 +116,7 @@ func TestComponentHooks_OnPanic(t *testing.T) { func TestComponentHooks_OnWaitingForInputs(t *testing.T) { var waitingCaught bool - c := mustComponent("processor", + c := testutil.MustComponent("processor", component.WithInputs("data", "config"), component.WithActivationFunc(func(c *component.Component) error { // Wait for config input @@ -158,7 +144,7 @@ func TestComponentHooks_OnWaitingForInputs(t *testing.T) { func TestComponentHooks_MultipleHooksPerType(t *testing.T) { var log []string - c := mustComponent("processor", + c := testutil.MustComponent("processor", component.WithInputs("in"), component.WithActivationFunc(func(c *component.Component) error { return nil @@ -191,7 +177,7 @@ func TestComponentHooks_MultipleHooksPerType(t *testing.T) { func TestComponentHooks_NoHooksOnNoInput(t *testing.T) { var beforeFired bool - c := mustComponent("processor", + c := testutil.MustComponent("processor", component.WithInputs("in"), component.WithActivationFunc(func(c *component.Component) error { return nil @@ -215,7 +201,7 @@ func TestComponentHooks_ContextAccess(t *testing.T) { var componentName string var activationCode component.ActivationResultCode - c := mustComponent("test-component", + c := testutil.MustComponent("test-component", component.WithInputs("in"), component.WithOutputs("out"), component.WithActivationFunc(func(c *component.Component) error { @@ -240,7 +226,7 @@ func TestComponentHooks_ContextAccess(t *testing.T) { func TestComponentHooks_IntegrationWithFMesh(t *testing.T) { var log hookLog - c1 := mustComponent("c1", + c1 := testutil.MustComponent("c1", component.WithInputs("in"), component.WithOutputs("out"), component.WithActivationFunc(func(c *component.Component) error { @@ -253,7 +239,7 @@ func TestComponentHooks_IntegrationWithFMesh(t *testing.T) { }) }) - c2 := mustComponent("c2", + c2 := testutil.MustComponent("c2", component.WithInputs("in"), component.WithActivationFunc(func(c *component.Component) error { return nil @@ -267,7 +253,7 @@ func TestComponentHooks_IntegrationWithFMesh(t *testing.T) { require.NoError(t, c1.OutputByName("out").PipeTo(c2.InputByName("in"))) - fm := mustFMesh("test") + fm := testutil.MustFMesh("test") require.NoError(t, fm.AddComponents(c1, c2)) require.NoError(t, c1.InputByName("in").PutSignals(signal.New(0))) @@ -284,7 +270,7 @@ func TestComponentHooks_ExecutionOrderAcrossComponents(t *testing.T) { var log hookLog // Create three components with hooks - c1 := mustComponent("c1", + c1 := testutil.MustComponent("c1", component.WithInputs("in"), component.WithOutputs("out"), component.WithActivationFunc(func(c *component.Component) error { @@ -305,7 +291,7 @@ func TestComponentHooks_ExecutionOrderAcrossComponents(t *testing.T) { }) }) - c2 := mustComponent("c2", + c2 := testutil.MustComponent("c2", component.WithInputs("in"), component.WithOutputs("out"), component.WithActivationFunc(func(c *component.Component) error { @@ -326,7 +312,7 @@ func TestComponentHooks_ExecutionOrderAcrossComponents(t *testing.T) { }) }) - c3 := mustComponent("c3", + c3 := testutil.MustComponent("c3", component.WithInputs("in"), component.WithActivationFunc(func(c *component.Component) error { return nil @@ -350,7 +336,7 @@ func TestComponentHooks_ExecutionOrderAcrossComponents(t *testing.T) { require.NoError(t, c1.OutputByName("out").PipeTo(c3.InputByName("in"))) require.NoError(t, c2.OutputByName("out").PipeTo(c3.InputByName("in"))) - fm := mustFMesh("test") + fm := testutil.MustFMesh("test") require.NoError(t, fm.AddComponents(c1, c2, c3)) require.NoError(t, c1.InputByName("in").PutSignals(signal.New(0))) require.NoError(t, c2.InputByName("in").PutSignals(signal.New(0))) @@ -387,7 +373,7 @@ func TestComponentHooks_MultipleSetupCalls(t *testing.T) { var log []string // Multiple SetupHooks calls should accumulate hooks - c := mustComponent("processor", + c := testutil.MustComponent("processor", component.WithInputs("in"), component.WithActivationFunc(func(c *component.Component) error { return nil @@ -419,7 +405,7 @@ func TestComponentHooks_MultipleSetupCalls(t *testing.T) { func BenchmarkComponentHooks_Overhead(b *testing.B) { // Measure overhead of hooks vs no hooks b.Run("WithoutHooks", func(b *testing.B) { - c := mustComponent("processor", + c := testutil.MustComponent("processor", component.WithInputs("in"), component.WithActivationFunc(func(c *component.Component) error { return nil @@ -439,7 +425,7 @@ func BenchmarkComponentHooks_Overhead(b *testing.B) { }) b.Run("WithHooks", func(b *testing.B) { - c := mustComponent("processor", + c := testutil.MustComponent("processor", component.WithInputs("in"), component.WithActivationFunc(func(c *component.Component) error { return nil diff --git a/integration_tests/computation/basic_test.go b/integration_tests/computation/basic_test.go index d0450cb..97997de 100644 --- a/integration_tests/computation/basic_test.go +++ b/integration_tests/computation/basic_test.go @@ -6,59 +6,18 @@ import ( "strings" "testing" + "github.com/hovsep/fmesh/internal/testutil" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/hovsep/fmesh" "github.com/hovsep/fmesh/component" "github.com/hovsep/fmesh/cycle" "github.com/hovsep/fmesh/port" "github.com/hovsep/fmesh/signal" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" ) -func mustComponent(name string, opts ...component.Option) *component.Component { - c, err := component.New(name, opts...) - if err != nil { - panic(err) - } - return c -} - -func mustInputPort(name string, opts ...port.Option) *port.Port { - p, err := port.NewInput(name, opts...) - if err != nil { - panic(err) - } - return p -} - -func mustOutputPort(name string, opts ...port.Option) *port.Port { - p, err := port.NewOutput(name, opts...) - if err != nil { - panic(err) - } - return p -} - -func mustFMesh(name string, opts ...fmesh.Option) *fmesh.FMesh { - fm, err := fmesh.New(name, opts...) - if err != nil { - panic(err) - } - return fm -} - -func mustPutSignals(p *port.Port, signals ...*signal.Signal) { - if err := p.PutSignals(signals...); err != nil { - panic(err) - } -} - -func mustPipeTo(src *port.Port, dsts ...*port.Port) { - if err := src.PipeTo(dsts...); err != nil { - panic(err) - } -} - func Test_Math(t *testing.T) { tests := []struct { name string @@ -69,7 +28,7 @@ func Test_Math(t *testing.T) { { name: "add and multiply", setupFM: func() *fmesh.FMesh { - c1 := mustComponent("c1", + c1 := testutil.MustComponent("c1", component.WithInputs("num"), component.WithOutputs("res"), component.WithDescription("adds 2 to the input"), @@ -79,7 +38,7 @@ func Test_Math(t *testing.T) { }), ) - c2 := mustComponent("c2", + c2 := testutil.MustComponent("c2", component.WithInputs("num"), component.WithOutputs("res"), component.WithDescription("multiplies by 3"), @@ -92,7 +51,7 @@ func Test_Math(t *testing.T) { if err := c1.OutputByName("res").PipeTo(c2.InputByName("num")); err != nil { panic(err) } - fm := mustFMesh("fm", fmesh.WithConfig(fmesh.Config{ + fm := testutil.MustFMesh("fm", fmesh.WithConfig(fmesh.Config{ ErrorHandlingStrategy: fmesh.StopOnFirstErrorOrPanic, CyclesLimit: 10, })) @@ -121,7 +80,7 @@ func Test_Math(t *testing.T) { name: "mixed port creation - simple and advanced", setupFM: func() *fmesh.FMesh { // Component with mixed port creation - processor := mustComponent("processor", + processor := testutil.MustComponent("processor", component.WithInputs("raw_data", "metadata"), component.WithOutputs("logs"), component.WithDescription("processes data using mixed ports"), @@ -150,7 +109,7 @@ func Test_Math(t *testing.T) { // Add advanced ports if err := processor.AttachInputPorts( - mustInputPort("config", + testutil.MustInputPort("config", port.WithDescription("Configuration parameters"), port.WithLabel("required", "true"), port.WithLabel("type", "config"), @@ -159,11 +118,11 @@ func Test_Math(t *testing.T) { panic(err) } if err := processor.AttachOutputPorts( - mustOutputPort("result", + testutil.MustOutputPort("result", port.WithDescription("Processed result"), port.WithLabel("format", "json"), ), - mustOutputPort("error", + testutil.MustOutputPort("error", port.WithDescription("Error details if any"), port.WithLabel("status", "error"), ), @@ -172,7 +131,7 @@ func Test_Math(t *testing.T) { } // Verifier component with simple ports - verifier := mustComponent("verifier", + verifier := testutil.MustComponent("verifier", component.WithInputs("value", "log"), component.WithOutputs("verified"), component.WithActivationFunc(func(this *component.Component) error { @@ -198,7 +157,7 @@ func Test_Math(t *testing.T) { panic(err) } - fm := mustFMesh("mixed_ports_fm", fmesh.WithConfig(fmesh.Config{ + fm := testutil.MustFMesh("mixed_ports_fm", fmesh.WithConfig(fmesh.Config{ ErrorHandlingStrategy: fmesh.StopOnFirstErrorOrPanic, CyclesLimit: 10, })) @@ -210,10 +169,10 @@ func Test_Math(t *testing.T) { setInputs: func(fm *fmesh.FMesh) { proc := fm.Components().ByName("processor") // Send data to simple ports - mustPutSignals(proc.InputByName("raw_data"), signal.New(10)) - mustPutSignals(proc.InputByName("metadata"), signal.New("test")) + testutil.MustPutSignals(proc.InputByName("raw_data"), signal.New(10)) + testutil.MustPutSignals(proc.InputByName("metadata"), signal.New("test")) // Send data to advanced port - mustPutSignals(proc.InputByName("config"), signal.New(5)) + testutil.MustPutSignals(proc.InputByName("config"), signal.New(5)) }, assertions: func(t *testing.T, fm *fmesh.FMesh, cycles []*cycle.Cycle, err error) { require.NoError(t, err) @@ -257,12 +216,12 @@ func Test_Math(t *testing.T) { func Test_Readme(t *testing.T) { t.Run("readme test", func(t *testing.T) { - fm := mustFMesh("hello world", fmesh.WithConfig(fmesh.Config{ + fm := testutil.MustFMesh("hello world", fmesh.WithConfig(fmesh.Config{ ErrorHandlingStrategy: fmesh.StopOnFirstErrorOrPanic, CyclesLimit: 10, })) - concat := mustComponent("concat", + concat := testutil.MustComponent("concat", component.WithInputs("i1", "i2"), component.WithOutputs("res"), component.WithActivationFunc(func(this *component.Component) error { @@ -272,7 +231,7 @@ func Test_Readme(t *testing.T) { }), ) - caseC := mustComponent("case", + caseC := testutil.MustComponent("case", component.WithInputs("i1"), component.WithOutputs("res"), component.WithActivationFunc(func(this *component.Component) error { @@ -292,8 +251,8 @@ func Test_Readme(t *testing.T) { } // Init inputs - mustPutSignals(fm.Components().ByName("concat").InputByName("i1"), signal.New("hello ")) - mustPutSignals(fm.Components().ByName("concat").InputByName("i2"), signal.New("world !")) + testutil.MustPutSignals(fm.Components().ByName("concat").InputByName("i1"), signal.New("hello ")) + testutil.MustPutSignals(fm.Components().ByName("concat").InputByName("i2"), signal.New("world !")) // Run the mesh _, err := fm.Run() diff --git a/integration_tests/constraints/time_test.go b/integration_tests/constraints/time_test.go index 8c04e25..f5f2562 100644 --- a/integration_tests/constraints/time_test.go +++ b/integration_tests/constraints/time_test.go @@ -4,29 +4,16 @@ import ( "testing" "time" + "github.com/hovsep/fmesh/internal/testutil" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/hovsep/fmesh" "github.com/hovsep/fmesh/component" "github.com/hovsep/fmesh/signal" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" ) -func mustComponent(name string, opts ...component.Option) *component.Component { - c, err := component.New(name, opts...) - if err != nil { - panic(err) - } - return c -} - -func mustFMesh(name string, opts ...fmesh.Option) *fmesh.FMesh { - fm, err := fmesh.New(name, opts...) - if err != nil { - panic(err) - } - return fm -} - func Test_TimeConstraint(t *testing.T) { tests := []struct { name string @@ -37,7 +24,7 @@ func Test_TimeConstraint(t *testing.T) { { name: "mesh stops by time constraint", setupFM: func() *fmesh.FMesh { - ticker := mustComponent("ticker", + ticker := testutil.MustComponent("ticker", component.WithInputs("tick_in", "start"), component.WithOutputs("tick_out"), component.WithDescription("simple clock ticking for 10 seconds"), @@ -60,7 +47,7 @@ func Test_TimeConstraint(t *testing.T) { panic(err) } - fm := mustFMesh("fm", fmesh.WithConfig(fmesh.Config{ + fm := testutil.MustFMesh("fm", fmesh.WithConfig(fmesh.Config{ Debug: true, TimeLimit: 2 * time.Second, }), fmesh.WithDescription("this mesh ticks every second for 10 seconds")) @@ -83,7 +70,7 @@ func Test_TimeConstraint(t *testing.T) { { name: "mesh stops naturally", setupFM: func() *fmesh.FMesh { - ticker := mustComponent("ticker", + ticker := testutil.MustComponent("ticker", component.WithInputs("tick_in", "start"), component.WithOutputs("tick_out"), component.WithDescription("simple clock ticking for 3 seconds"), @@ -106,7 +93,7 @@ func Test_TimeConstraint(t *testing.T) { panic(err) } - fm := mustFMesh("fm", fmesh.WithConfig(fmesh.Config{ + fm := testutil.MustFMesh("fm", fmesh.WithConfig(fmesh.Config{ Debug: true, TimeLimit: 0, }), fmesh.WithDescription("this mesh ticks every second for 10 seconds")) diff --git a/integration_tests/errorhandling/error_returns_test.go b/integration_tests/errorhandling/error_returns_test.go index 537675b..7c31278 100644 --- a/integration_tests/errorhandling/error_returns_test.go +++ b/integration_tests/errorhandling/error_returns_test.go @@ -3,12 +3,13 @@ package errorhandling_test import ( "testing" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/hovsep/fmesh" "github.com/hovsep/fmesh/component" "github.com/hovsep/fmesh/port" "github.com/hovsep/fmesh/signal" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" ) func TestErrorReturns_DuplicateComponent(t *testing.T) { diff --git a/integration_tests/errorhandling/orphaned_component_test.go b/integration_tests/errorhandling/orphaned_component_test.go index b15633e..61d4dbd 100644 --- a/integration_tests/errorhandling/orphaned_component_test.go +++ b/integration_tests/errorhandling/orphaned_component_test.go @@ -3,10 +3,11 @@ package errorhandling import ( "testing" + "github.com/stretchr/testify/require" + "github.com/hovsep/fmesh" "github.com/hovsep/fmesh/component" "github.com/hovsep/fmesh/signal" - "github.com/stretchr/testify/require" ) func Test_AllComponentsMustBeRegistered(t *testing.T) { diff --git a/integration_tests/hooks/component_test.go b/integration_tests/hooks/component_test.go index cb96218..f2f3e15 100644 --- a/integration_tests/hooks/component_test.go +++ b/integration_tests/hooks/component_test.go @@ -4,11 +4,14 @@ import ( "errors" "testing" + "github.com/hovsep/fmesh/internal/testutil" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/hovsep/fmesh/component" "github.com/hovsep/fmesh/port" "github.com/hovsep/fmesh/signal" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" ) func TestComponentHooks_PracticalErrorLogging(t *testing.T) { @@ -22,7 +25,7 @@ func TestComponentHooks_PracticalErrorLogging(t *testing.T) { validationErr := errors.New("validation failed: negative value") - c := mustComponent("validator", + c := testutil.MustComponent("validator", component.WithInputs("data"), component.WithActivationFunc(func(c *component.Component) error { // Simulate validation logic @@ -58,7 +61,7 @@ func TestComponentHooks_PracticalOutputValidation(t *testing.T) { var outputIsValid bool var outputValue int - c := mustComponent("calculator", + c := testutil.MustComponent("calculator", component.WithInputs("x", "y"), component.WithOutputs("result"), component.WithActivationFunc(func(c *component.Component) error { @@ -102,7 +105,7 @@ func TestComponentHooks_PracticalMetricsCollection(t *testing.T) { OutputSignalCounts: make(map[string]int), } - c := mustComponent("processor", + c := testutil.MustComponent("processor", component.WithInputs("in"), component.WithOutputs("success", "failure"), component.WithActivationFunc(func(c *component.Component) error { @@ -163,7 +166,7 @@ func TestComponentHooks_PracticalDataTransformation(t *testing.T) { // Practical example: Transform or enrich output data in hooks var enrichedOutput []map[string]any - c := mustComponent("enricher", + c := testutil.MustComponent("enricher", component.WithInputs("raw"), component.WithOutputs("enriched"), component.WithActivationFunc(func(c *component.Component) error { @@ -206,7 +209,7 @@ func TestComponentHooks_PracticalErrorRecovery(t *testing.T) { var recoveryAttempted bool var fallbackValueProvided bool - c := mustComponent("resilient", + c := testutil.MustComponent("resilient", component.WithInputs("in"), component.WithOutputs("out"), component.WithActivationFunc(func(c *component.Component) error { @@ -250,7 +253,7 @@ func TestComponentHooks_PracticalInputOutputInspection(t *testing.T) { } var trace ActivationTrace - c := mustComponent("aggregator", + c := testutil.MustComponent("aggregator", component.WithInputs("numbers"), component.WithOutputs("sum"), component.WithActivationFunc(func(c *component.Component) error { diff --git a/integration_tests/hooks/fmesh_test.go b/integration_tests/hooks/fmesh_test.go index 7d02262..e8356bb 100644 --- a/integration_tests/hooks/fmesh_test.go +++ b/integration_tests/hooks/fmesh_test.go @@ -3,35 +3,22 @@ package hooks import ( "testing" + "github.com/hovsep/fmesh/internal/testutil" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/hovsep/fmesh" "github.com/hovsep/fmesh/component" "github.com/hovsep/fmesh/signal" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" ) -func mustFMesh(name string, opts ...fmesh.Option) *fmesh.FMesh { - fm, err := fmesh.New(name, opts...) - if err != nil { - panic(err) - } - return fm -} - -func mustComponent(name string, opts ...component.Option) *component.Component { - c, err := component.New(name, opts...) - if err != nil { - panic(err) - } - return c -} - func TestHooks_AllTypes(t *testing.T) { // Track hook execution var executionLog []string // Create a simple component - c := mustComponent("processor", + c := testutil.MustComponent("processor", component.WithInputs("in"), component.WithActivationFunc(func(c *component.Component) error { return nil @@ -40,7 +27,7 @@ func TestHooks_AllTypes(t *testing.T) { // Create a mesh with all hook types. // Hooks are registered before AddComponents so OnComponentAdded fires. - fm := mustFMesh("test-mesh") + fm := testutil.MustFMesh("test-mesh") fm.SetupHooks(func(h *fmesh.Hooks) { h.OnComponentAdded(func(ctx *fmesh.ComponentAddedContext) error { executionLog = append(executionLog, "componentAdded") @@ -94,14 +81,14 @@ func TestHooks_CycleContext(t *testing.T) { cycleNumbers := []int{} // Create a simple component - c := mustComponent("processor", + c := testutil.MustComponent("processor", component.WithInputs("in"), component.WithActivationFunc(func(c *component.Component) error { return nil }), ) - fm := mustFMesh("test-mesh") + fm := testutil.MustFMesh("test-mesh") require.NoError(t, fm.AddComponents(c)) fm.SetupHooks(func(h *fmesh.Hooks) { h.BeforeCycle(func(ctx *fmesh.CycleContext) error { @@ -127,14 +114,14 @@ func TestHooks_MultipleHooksPerType(t *testing.T) { // Test that multiple hooks of the same type execute in order var log []string - c := mustComponent("test", + c := testutil.MustComponent("test", component.WithInputs("in"), component.WithActivationFunc(func(c *component.Component) error { return nil }), ) - fm := mustFMesh("test-mesh") + fm := testutil.MustFMesh("test-mesh") require.NoError(t, fm.AddComponents(c)) fm.SetupHooks(func(h *fmesh.Hooks) { h.BeforeRun(func(fm *fmesh.FMesh) error { @@ -167,14 +154,14 @@ func TestHooks_ContextAccess(t *testing.T) { var cycleNumber int var activationCount int - c := mustComponent("test", + c := testutil.MustComponent("test", component.WithInputs("in"), component.WithActivationFunc(func(c *component.Component) error { return nil }), ) - fm := mustFMesh("my-mesh") + fm := testutil.MustFMesh("my-mesh") require.NoError(t, fm.AddComponents(c)) fm.SetupHooks(func(h *fmesh.Hooks) { h.AfterCycle(func(ctx *fmesh.CycleContext) error { @@ -203,14 +190,14 @@ func TestHooks_FireOncePerCycle(t *testing.T) { var beginCount int var endCount int - c := mustComponent("processor", + c := testutil.MustComponent("processor", component.WithInputs("in"), component.WithActivationFunc(func(c *component.Component) error { return nil }), ) - fm := mustFMesh("test-mesh") + fm := testutil.MustFMesh("test-mesh") require.NoError(t, fm.AddComponents(c)) fm.SetupHooks(func(h *fmesh.Hooks) { h.BeforeCycle(func(ctx *fmesh.CycleContext) error { @@ -242,7 +229,7 @@ func TestHooks_RunWithError(t *testing.T) { var afterRunFired bool // Create a mesh that will have no components (simulating a broken mesh) - fm := mustFMesh("test-mesh") + fm := testutil.MustFMesh("test-mesh") fm.SetupHooks(func(h *fmesh.Hooks) { h.BeforeRun(func(fm *fmesh.FMesh) error { beforeRunFired = true @@ -267,7 +254,7 @@ func TestHooks_EmptyMesh(t *testing.T) { var afterRunFired bool // Create mesh with no components - this will error - fm := mustFMesh("empty-mesh") + fm := testutil.MustFMesh("empty-mesh") fm.SetupHooks(func(h *fmesh.Hooks) { h.BeforeRun(func(fm *fmesh.FMesh) error { beforeRunFired = true @@ -290,7 +277,7 @@ func TestHooks_EmptyMesh(t *testing.T) { func TestHooks_MultipleSetupCalls(t *testing.T) { var log []string - c := mustComponent("test", + c := testutil.MustComponent("test", component.WithInputs("in"), component.WithActivationFunc(func(c *component.Component) error { return nil @@ -298,7 +285,7 @@ func TestHooks_MultipleSetupCalls(t *testing.T) { ) // Multiple SetupHooks calls should accumulate hooks - fm := mustFMesh("test-mesh") + fm := testutil.MustFMesh("test-mesh") require.NoError(t, fm.AddComponents(c)) fm.SetupHooks(func(h *fmesh.Hooks) { h.BeforeRun(func(fm *fmesh.FMesh) error { diff --git a/integration_tests/hooks/port_test.go b/integration_tests/hooks/port_test.go index e1780d5..6fd9e52 100644 --- a/integration_tests/hooks/port_test.go +++ b/integration_tests/hooks/port_test.go @@ -4,34 +4,21 @@ import ( "errors" "testing" - "github.com/hovsep/fmesh/port" - "github.com/hovsep/fmesh/signal" + "github.com/hovsep/fmesh/internal/testutil" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" -) - -func mustInputPort(name string, opts ...port.Option) *port.Port { - p, err := port.NewInput(name, opts...) - if err != nil { - panic(err) - } - return p -} -func mustOutputPort(name string, opts ...port.Option) *port.Port { - p, err := port.NewOutput(name, opts...) - if err != nil { - panic(err) - } - return p -} + "github.com/hovsep/fmesh/port" + "github.com/hovsep/fmesh/signal" +) func TestPortHooks_OnSignalsAdded(t *testing.T) { var hookFired bool var portName string var signalsAdded int - p := mustInputPort("data"). + p := testutil.MustInputPort("data"). SetupHooks(func(h *port.Hooks) { h.OnSignalsAdded(func(ctx *port.SignalsAddedContext) error { hookFired = true @@ -53,7 +40,7 @@ func TestPortHooks_OnSignalsAdded_MultipleCalls(t *testing.T) { var callCount int var totalSignalsHistory []int - p := mustOutputPort("result"). + p := testutil.MustOutputPort("result"). SetupHooks(func(h *port.Hooks) { h.OnSignalsAdded(func(ctx *port.SignalsAddedContext) error { callCount++ @@ -74,7 +61,7 @@ func TestPortHooks_OnClear(t *testing.T) { var clearFired bool var signalsCleared int - p := mustInputPort("data"). + p := testutil.MustInputPort("data"). SetupHooks(func(h *port.Hooks) { h.OnClear(func(ctx *port.ClearContext) error { clearFired = true @@ -95,7 +82,7 @@ func TestPortHooks_OnClear_EmptyPort(t *testing.T) { var clearFired bool var signalsCleared int - p := mustInputPort("data"). + p := testutil.MustInputPort("data"). SetupHooks(func(h *port.Hooks) { h.OnClear(func(ctx *port.ClearContext) error { clearFired = true @@ -115,7 +102,7 @@ func TestPortHooks_OnOutboundPipe(t *testing.T) { var sourceName string var destName string - outPort := mustOutputPort("out"). + outPort := testutil.MustOutputPort("out"). SetupHooks(func(h *port.Hooks) { h.OnOutboundPipe(func(ctx *port.OutboundPipeContext) error { outboundFired = true @@ -125,7 +112,7 @@ func TestPortHooks_OnOutboundPipe(t *testing.T) { }) }) - inPort := mustInputPort("in") + inPort := testutil.MustInputPort("in") require.NoError(t, outPort.PipeTo(inPort)) @@ -139,9 +126,9 @@ func TestPortHooks_OnInboundPipe(t *testing.T) { var sourceName string var destName string - outPort := mustOutputPort("out") + outPort := testutil.MustOutputPort("out") - inPort := mustInputPort("in"). + inPort := testutil.MustInputPort("in"). SetupHooks(func(h *port.Hooks) { h.OnInboundPipe(func(ctx *port.InboundPipeContext) error { inboundFired = true @@ -162,7 +149,7 @@ func TestPortHooks_OnOutboundAndInbound_BothFire(t *testing.T) { var outboundFired bool var inboundFired bool - outPort := mustOutputPort("out"). + outPort := testutil.MustOutputPort("out"). SetupHooks(func(h *port.Hooks) { h.OnOutboundPipe(func(ctx *port.OutboundPipeContext) error { outboundFired = true @@ -170,7 +157,7 @@ func TestPortHooks_OnOutboundAndInbound_BothFire(t *testing.T) { }) }) - inPort := mustInputPort("in"). + inPort := testutil.MustInputPort("in"). SetupHooks(func(h *port.Hooks) { h.OnInboundPipe(func(ctx *port.InboundPipeContext) error { inboundFired = true @@ -188,7 +175,7 @@ func TestPortHooks_OnOutboundPipe_MultipleDest(t *testing.T) { var outboundCount int var destNames []string - outPort := mustOutputPort("out"). + outPort := testutil.MustOutputPort("out"). SetupHooks(func(h *port.Hooks) { h.OnOutboundPipe(func(ctx *port.OutboundPipeContext) error { outboundCount++ @@ -197,9 +184,9 @@ func TestPortHooks_OnOutboundPipe_MultipleDest(t *testing.T) { }) }) - in1 := mustInputPort("in1") - in2 := mustInputPort("in2") - in3 := mustInputPort("in3") + in1 := testutil.MustInputPort("in1") + in2 := testutil.MustInputPort("in2") + in3 := testutil.MustInputPort("in3") require.NoError(t, outPort.PipeTo(in1, in2, in3)) @@ -210,7 +197,7 @@ func TestPortHooks_OnOutboundPipe_MultipleDest(t *testing.T) { func TestPortHooks_MultipleHooksPerType(t *testing.T) { var log []string - p := mustInputPort("data"). + p := testutil.MustInputPort("data"). SetupHooks(func(h *port.Hooks) { h.OnSignalsAdded(func(ctx *port.SignalsAddedContext) error { log = append(log, "put1") @@ -240,7 +227,7 @@ func TestPortHooks_ContextAccess(t *testing.T) { var portName string var signalPayloads []int - p := mustInputPort("sensor"). + p := testutil.MustInputPort("sensor"). SetupHooks(func(h *port.Hooks) { h.OnSignalsAdded(func(ctx *port.SignalsAddedContext) error { portName = ctx.Port.Name() @@ -275,7 +262,7 @@ func TestPortHooks_PracticalVolumeMonitoring(t *testing.T) { } metrics := VolumeMetrics{} - p := mustOutputPort("stream"). + p := testutil.MustOutputPort("stream"). SetupHooks(func(h *port.Hooks) { h.OnSignalsAdded(func(ctx *port.SignalsAddedContext) error { metrics.TotalPuts++ @@ -313,7 +300,7 @@ func TestPortHooks_PracticalTopologyTracking(t *testing.T) { Connections: make(map[string][]string), } - out1 := mustOutputPort("out1"). + out1 := testutil.MustOutputPort("out1"). SetupHooks(func(h *port.Hooks) { h.OnOutboundPipe(func(ctx *port.OutboundPipeContext) error { srcName := ctx.SourcePort.Name() @@ -323,7 +310,7 @@ func TestPortHooks_PracticalTopologyTracking(t *testing.T) { }) }) - out2 := mustOutputPort("out2"). + out2 := testutil.MustOutputPort("out2"). SetupHooks(func(h *port.Hooks) { h.OnOutboundPipe(func(ctx *port.OutboundPipeContext) error { srcName := ctx.SourcePort.Name() @@ -333,9 +320,9 @@ func TestPortHooks_PracticalTopologyTracking(t *testing.T) { }) }) - in1 := mustInputPort("in1") - in2 := mustInputPort("in2") - in3 := mustInputPort("in3") + in1 := testutil.MustInputPort("in1") + in2 := testutil.MustInputPort("in2") + in3 := testutil.MustInputPort("in3") // Create topology: out1 -> in1, in2; out2 -> in2, in3 require.NoError(t, out1.PipeTo(in1, in2)) @@ -347,7 +334,7 @@ func TestPortHooks_PracticalTopologyTracking(t *testing.T) { func TestPortHooks_PracticalDataValidation(t *testing.T) { // Practical example: Validate incoming data - p := mustInputPort("validated"). + p := testutil.MustInputPort("validated"). SetupHooks(func(h *port.Hooks) { h.OnSignalsAdded(func(ctx *port.SignalsAddedContext) error { // Validate: must receive exactly 3 signals diff --git a/integration_tests/labels/transform_test.go b/integration_tests/labels/transform_test.go index f5676ce..ec2b3ee 100644 --- a/integration_tests/labels/transform_test.go +++ b/integration_tests/labels/transform_test.go @@ -4,28 +4,14 @@ import ( "strings" "testing" - "github.com/hovsep/fmesh" - "github.com/hovsep/fmesh/component" - "github.com/hovsep/fmesh/signal" + "github.com/hovsep/fmesh/internal/testutil" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" -) - -func mustComponent(name string, opts ...component.Option) *component.Component { - c, err := component.New(name, opts...) - if err != nil { - panic(err) - } - return c -} -func mustFMesh(name string, opts ...fmesh.Option) *fmesh.FMesh { - fm, err := fmesh.New(name, opts...) - if err != nil { - panic(err) - } - return fm -} + "github.com/hovsep/fmesh/component" + "github.com/hovsep/fmesh/signal" +) // Test_LabelTransformation demonstrates real-world use cases for labels.Labels.Map(). func Test_LabelTransformation(t *testing.T) { @@ -51,7 +37,7 @@ func Test_LabelTransformation(t *testing.T) { t.Run("add namespace prefix to labels", func(t *testing.T) { // Scenario: You want to namespace all labels to avoid conflicts - c := mustComponent("processor", + c := testutil.MustComponent("processor", component.WithInputs("in"), component.WithOutputs("out"), ).SetLabels(map[string]string{ @@ -104,7 +90,7 @@ func Test_LabelTransformation(t *testing.T) { t.Run("transform labels in mesh processing", func(t *testing.T) { // Scenario: Process signals and normalize their labels during mesh execution - normalizer := mustComponent("normalizer", + normalizer := testutil.MustComponent("normalizer", component.WithInputs("in"), component.WithOutputs("out"), component.WithActivationFunc(func(this *component.Component) error { @@ -129,7 +115,7 @@ func Test_LabelTransformation(t *testing.T) { }), ) - fm := mustFMesh("label-transform-mesh") + fm := testutil.MustFMesh("label-transform-mesh") require.NoError(t, fm.AddComponents(normalizer)) // Input signal with mixed-case labels diff --git a/integration_tests/meta/meta_test.go b/integration_tests/meta/meta_test.go index 15511cb..15a953b 100644 --- a/integration_tests/meta/meta_test.go +++ b/integration_tests/meta/meta_test.go @@ -3,46 +3,16 @@ package meta import ( "testing" - "github.com/hovsep/fmesh" + "github.com/hovsep/fmesh/internal/testutil" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/hovsep/fmesh/component" "github.com/hovsep/fmesh/port" "github.com/hovsep/fmesh/signal" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" ) -func mustComponent(name string, opts ...component.Option) *component.Component { - c, err := component.New(name, opts...) - if err != nil { - panic(err) - } - return c -} - -func mustFMesh(name string, opts ...fmesh.Option) *fmesh.FMesh { - fm, err := fmesh.New(name, opts...) - if err != nil { - panic(err) - } - return fm -} - -func mustInput(name string, opts ...port.Option) *port.Port { - p, err := port.NewInput(name, opts...) - if err != nil { - panic(err) - } - return p -} - -func mustOutput(name string, opts ...port.Option) *port.Port { - p, err := port.NewOutput(name, opts...) - if err != nil { - panic(err) - } - return p -} - // Test_ScalarsOnSignals verifies that scalars can be attached to signals and // aggregate methods on signal.Group work as expected. func Test_ScalarsOnSignals(t *testing.T) { @@ -52,7 +22,7 @@ func Test_ScalarsOnSignals(t *testing.T) { var collectedGroup *signal.Group - sensor := mustComponent("sensor", + sensor := testutil.MustComponent("sensor", component.WithInputs("trigger"), component.WithOutputs("out"), component.WithActivationFunc(func(this *component.Component) error { @@ -67,7 +37,7 @@ func Test_ScalarsOnSignals(t *testing.T) { }), ) - monitor := mustComponent("monitor", + monitor := testutil.MustComponent("monitor", component.WithInputs("in"), component.WithActivationFunc(func(this *component.Component) error { grp := this.Inputs().ByName("in").Signals() @@ -76,7 +46,7 @@ func Test_ScalarsOnSignals(t *testing.T) { }), ) - fm := mustFMesh("temp-mesh") + fm := testutil.MustFMesh("temp-mesh") require.NoError(t, fm.AddComponents(sensor, monitor)) require.NoError(t, sensor.Outputs().ByName("out").PipeTo(monitor.Inputs().ByName("in"))) @@ -140,7 +110,7 @@ func Test_ScalarsOnSignals(t *testing.T) { // Test_ScalarsOnComponents verifies scalar metadata on components. func Test_ScalarsOnComponents(t *testing.T) { t.Run("component scalars are independent of signal scalars", func(t *testing.T) { - c := mustComponent("proc", + c := testutil.MustComponent("proc", component.WithInputs("in"), component.WithOutputs("out"), component.WithLabel("tier", "premium"), @@ -186,14 +156,14 @@ func Test_ScalarGroupMetadata(t *testing.T) { // Test_PortScalarsWithOptions verifies the WithScalar port constructor option. func Test_PortScalarsWithOptions(t *testing.T) { t.Run("port scalars set via constructor option", func(t *testing.T) { - p := mustInput("sensor-in", port.WithScalar("sample_rate", 100.0)) + p := testutil.MustInputPort("sensor-in", port.WithScalar("sample_rate", 100.0)) v, err := p.Scalars().Value("sample_rate") require.NoError(t, err) assert.InDelta(t, 100.0, v, 1e-9) }) t.Run("port WithScalar mutating method", func(t *testing.T) { - p := mustOutput("data-out").AddScalar("bandwidth", 1e6) + p := testutil.MustOutputPort("data-out").AddScalar("bandwidth", 1e6) v, err := p.Scalars().Value("bandwidth") require.NoError(t, err) assert.InDelta(t, 1e6, v, 1e-9) diff --git a/integration_tests/piping/fan_out_concurrency_integration_test.go b/integration_tests/piping/fan_out_concurrency_integration_test.go index 465adcd..7d830c3 100644 --- a/integration_tests/piping/fan_out_concurrency_integration_test.go +++ b/integration_tests/piping/fan_out_concurrency_integration_test.go @@ -6,12 +6,15 @@ import ( "testing" "unsafe" + "github.com/hovsep/fmesh/internal/testutil" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/hovsep/fmesh" "github.com/hovsep/fmesh/component" "github.com/hovsep/fmesh/port" "github.com/hovsep/fmesh/signal" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" ) // TestFanOut_threeConsumers_seeSameSignalPointer asserts fan-out wiring: one @@ -20,7 +23,7 @@ import ( func TestFanOut_threeConsumers_seeSameSignalPointer(t *testing.T) { var ptrs sync.Map - producer := mustComponent("producer", + producer := testutil.MustComponent("producer", component.WithInputs("start"), component.WithOutputs("o1"), component.WithActivationFunc(func(this *component.Component) error { @@ -28,7 +31,7 @@ func TestFanOut_threeConsumers_seeSameSignalPointer(t *testing.T) { })) makeConsumer := func(name, slot string) *component.Component { - return mustComponent(name, + return testutil.MustComponent(name, component.WithInputs("i1"), component.WithOutputs("o1"), component.WithActivationFunc(func(this *component.Component) error { @@ -44,7 +47,7 @@ func TestFanOut_threeConsumers_seeSameSignalPointer(t *testing.T) { c2 := makeConsumer("consumer2", "2") c3 := makeConsumer("consumer3", "3") - fm := mustFMesh("fan-out-same-pointer") + fm := testutil.MustFMesh("fan-out-same-pointer") require.NoError(t, fm.AddComponents(producer, c1, c2, c3)) require.NoError(t, fm.Components().ByName("producer").OutputByName("o1").PipeTo( fm.Components().ByName("consumer1").InputByName("i1"), @@ -74,7 +77,7 @@ func TestFanOut_sharedSignal_parallelStress_completes(t *testing.T) { var ptrs sync.Map - producer := mustComponent("producer", + producer := testutil.MustComponent("producer", component.WithInputs("start"), component.WithOutputs("o1"), component.WithActivationFunc(func(this *component.Component) error { @@ -82,7 +85,7 @@ func TestFanOut_sharedSignal_parallelStress_completes(t *testing.T) { })) makeConsumer := func(name string, mode int) *component.Component { - return mustComponent(name, + return testutil.MustComponent(name, component.WithInputs("i1"), component.WithOutputs("o1"), component.WithActivationFunc(func(this *component.Component) error { diff --git a/integration_tests/piping/fan_test.go b/integration_tests/piping/fan_test.go index c2303f8..6f260de 100644 --- a/integration_tests/piping/fan_test.go +++ b/integration_tests/piping/fan_test.go @@ -5,31 +5,18 @@ import ( "testing" "time" + "github.com/hovsep/fmesh/internal/testutil" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/hovsep/fmesh" "github.com/hovsep/fmesh/component" "github.com/hovsep/fmesh/cycle" "github.com/hovsep/fmesh/port" "github.com/hovsep/fmesh/signal" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" ) -func mustComponent(name string, opts ...component.Option) *component.Component { - c, err := component.New(name, opts...) - if err != nil { - panic(err) - } - return c -} - -func mustFMesh(name string, opts ...fmesh.Option) *fmesh.FMesh { - fm, err := fmesh.New(name, opts...) - if err != nil { - panic(err) - } - return fm -} - func Test_Fan(t *testing.T) { tests := []struct { name string @@ -40,28 +27,28 @@ func Test_Fan(t *testing.T) { { name: "fan-out (3 pipes from 1 source port)", setupFM: func() *fmesh.FMesh { - producer := mustComponent("producer", + producer := testutil.MustComponent("producer", component.WithInputs("start"), component.WithOutputs("o1"), component.WithActivationFunc(func(this *component.Component) error { return this.OutputByName("o1").PutSignals(signal.New(time.Now())) })) - consumer1 := mustComponent("consumer1", + consumer1 := testutil.MustComponent("consumer1", component.WithInputs("i1"), component.WithOutputs("o1"), component.WithActivationFunc(func(this *component.Component) error { return port.ForwardSignals(this.InputByName("i1"), this.OutputByName("o1")) })) - consumer2 := mustComponent("consumer2", + consumer2 := testutil.MustComponent("consumer2", component.WithInputs("i1"), component.WithOutputs("o1"), component.WithActivationFunc(func(this *component.Component) error { return port.ForwardSignals(this.InputByName("i1"), this.OutputByName("o1")) })) - consumer3 := mustComponent("consumer3", + consumer3 := testutil.MustComponent("consumer3", component.WithInputs("i1"), component.WithOutputs("o1"), component.WithActivationFunc(func(this *component.Component) error { @@ -75,7 +62,7 @@ func Test_Fan(t *testing.T) { panic(err) } - fm := mustFMesh("fan-out") + fm := testutil.MustFMesh("fan-out") if err := fm.AddComponents(producer, consumer1, consumer2, consumer3); err != nil { panic(err) } @@ -109,28 +96,28 @@ func Test_Fan(t *testing.T) { { name: "fan-in (3 pipes coming into 1 destination port)", setupFM: func() *fmesh.FMesh { - producer1 := mustComponent("producer1", + producer1 := testutil.MustComponent("producer1", component.WithInputs("start"), component.WithOutputs("o1"), component.WithActivationFunc(func(this *component.Component) error { return this.OutputByName("o1").PutSignals(signal.New(rand.Int())) })) - producer2 := mustComponent("producer2", + producer2 := testutil.MustComponent("producer2", component.WithInputs("start"), component.WithOutputs("o1"), component.WithActivationFunc(func(this *component.Component) error { return this.OutputByName("o1").PutSignals(signal.New(rand.Int())) })) - producer3 := mustComponent("producer3", + producer3 := testutil.MustComponent("producer3", component.WithInputs("start"), component.WithOutputs("o1"), component.WithActivationFunc(func(this *component.Component) error { return this.OutputByName("o1").PutSignals(signal.New(rand.Int())) })) - consumer := mustComponent("consumer", + consumer := testutil.MustComponent("consumer", component.WithInputs("i1"), component.WithOutputs("o1"), component.WithActivationFunc(func(this *component.Component) error { @@ -147,7 +134,7 @@ func Test_Fan(t *testing.T) { panic(err) } - fm := mustFMesh("multiplexer") + fm := testutil.MustFMesh("multiplexer") if err := fm.AddComponents(producer1, producer2, producer3, consumer); err != nil { panic(err) } diff --git a/integration_tests/ports/port_creation_test.go b/integration_tests/ports/port_creation_test.go index 4da5e77..6a1b07f 100644 --- a/integration_tests/ports/port_creation_test.go +++ b/integration_tests/ports/port_creation_test.go @@ -4,50 +4,20 @@ import ( "fmt" "testing" - "github.com/hovsep/fmesh" + "github.com/hovsep/fmesh/internal/testutil" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/hovsep/fmesh/component" "github.com/hovsep/fmesh/port" "github.com/hovsep/fmesh/signal" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" ) -func mustComponent(name string, opts ...component.Option) *component.Component { - c, err := component.New(name, opts...) - if err != nil { - panic(err) - } - return c -} - -func mustInputPort(name string, opts ...port.Option) *port.Port { - p, err := port.NewInput(name, opts...) - if err != nil { - panic(err) - } - return p -} - -func mustOutputPort(name string, opts ...port.Option) *port.Port { - p, err := port.NewOutput(name, opts...) - if err != nil { - panic(err) - } - return p -} - -func mustFMesh(name string, opts ...fmesh.Option) *fmesh.FMesh { - fm, err := fmesh.New(name, opts...) - if err != nil { - panic(err) - } - return fm -} - func Test_PortCreationAndManipulation(t *testing.T) { t.Run("mixed port creation with all features", func(t *testing.T) { // Create a component using both simple and advanced port creation APIs - processor := mustComponent("data-processor", + processor := testutil.MustComponent("data-processor", component.WithInputs("raw_data", "filter"), component.WithOutputs("processed", "metrics"), component.WithDescription("Demonstrates all port creation and manipulation features"), @@ -79,14 +49,14 @@ func Test_PortCreationAndManipulation(t *testing.T) { // Advanced API: attach ports with descriptions and labels require.NoError(t, processor.AttachInputPorts( - mustInputPort("config", port.WithDescription("Configuration parameters")). + testutil.MustInputPort("config", port.WithDescription("Configuration parameters")). AddLabel("required", "true"). AddLabel("type", "json"), - mustInputPort("metadata", port.WithDescription("Request metadata")). + testutil.MustInputPort("metadata", port.WithDescription("Request metadata")). AddLabel("required", "false"), )) require.NoError(t, processor.AttachOutputPorts( - mustOutputPort("errors", port.WithDescription("Error details if processing fails")). + testutil.MustOutputPort("errors", port.WithDescription("Error details if processing fails")). AddLabel("severity", "high"). AddLabel("format", "structured"), )) @@ -98,7 +68,7 @@ func Test_PortCreationAndManipulation(t *testing.T) { require.NoError(t, processor.InputByName("metadata").PutSignals(signal.New("user123"))) // Create and run mesh - fm := mustFMesh("test-mesh") + fm := testutil.MustFMesh("test-mesh") require.NoError(t, fm.AddComponents(processor)) _, err := fm.Run() require.NoError(t, err) @@ -156,7 +126,7 @@ func Test_PortCreationAndManipulation(t *testing.T) { t.Run("port label manipulation", func(t *testing.T) { // Create a component and manipulate port labels - c := mustComponent("label-demo", + c := testutil.MustComponent("label-demo", component.WithOutputs("output"), component.WithActivationFunc(func(this *component.Component) error { if !this.InputByName("input").HasSignals() { @@ -184,7 +154,7 @@ func Test_PortCreationAndManipulation(t *testing.T) { }), ) require.NoError(t, c.AttachInputPorts( - mustInputPort("input"). + testutil.MustInputPort("input"). AddLabel("env", "dev"). AddLabel("version", "1.0"). AddLabel("owner", "team-a"), @@ -192,7 +162,7 @@ func Test_PortCreationAndManipulation(t *testing.T) { // Set up and run require.NoError(t, c.InputByName("input").PutSignals(signal.New("data"))) - fm := mustFMesh("label-mesh") + fm := testutil.MustFMesh("label-mesh") require.NoError(t, fm.AddComponents(c)) _, err := fm.Run() require.NoError(t, err) @@ -211,7 +181,7 @@ func Test_PortCreationAndManipulation(t *testing.T) { t.Run("incremental port addition", func(t *testing.T) { // Demonstrate adding ports one by one - c := mustComponent("incremental", + c := testutil.MustComponent("incremental", component.WithOutputs("result"), component.WithActivationFunc(func(this *component.Component) error { if !this.Inputs().AllHaveSignals() { @@ -228,7 +198,7 @@ func Test_PortCreationAndManipulation(t *testing.T) { require.NoError(t, c.AddInputs("a")) // Add first input require.NoError(t, c.AddInputs("b")) // Add second input require.NoError(t, c.AttachInputPorts( // Add with details - mustInputPort("c", port.WithDescription("Third input")), + testutil.MustInputPort("c", port.WithDescription("Third input")), )) // Verify all ports exist and work @@ -236,7 +206,7 @@ func Test_PortCreationAndManipulation(t *testing.T) { require.NoError(t, c.InputByName("b").PutSignals(signal.New(2))) require.NoError(t, c.InputByName("c").PutSignals(signal.New(3))) - fm := mustFMesh("incremental-mesh") + fm := testutil.MustFMesh("incremental-mesh") require.NoError(t, fm.AddComponents(c)) _, err := fm.Run() require.NoError(t, err) @@ -251,7 +221,7 @@ func Test_PortCreationAndManipulation(t *testing.T) { t.Run("port collection operations", func(t *testing.T) { // Demonstrate port collection methods - c := mustComponent("collection-demo", + c := testutil.MustComponent("collection-demo", component.WithInputs("i1", "i2", "i3"), component.WithOutputs("summary"), component.WithActivationFunc(func(this *component.Component) error { @@ -284,8 +254,8 @@ func Test_PortCreationAndManipulation(t *testing.T) { }), ) require.NoError(t, c.AttachInputPorts( - mustInputPort("i4").AddLabel("priority", "high"), - mustInputPort("i5").AddLabel("priority", "low"), + testutil.MustInputPort("i4").AddLabel("priority", "high"), + testutil.MustInputPort("i5").AddLabel("priority", "low"), )) // Put signals on some ports @@ -293,7 +263,7 @@ func Test_PortCreationAndManipulation(t *testing.T) { require.NoError(t, c.InputByName("i2").PutSignals(signal.New(2))) require.NoError(t, c.InputByName("i4").PutSignals(signal.New(4))) - fm := mustFMesh("collection-mesh") + fm := testutil.MustFMesh("collection-mesh") require.NoError(t, fm.AddComponents(c)) _, err := fm.Run() require.NoError(t, err) diff --git a/integration_tests/ports/waiting_for_inputs_test.go b/integration_tests/ports/waiting_for_inputs_test.go index df60a30..0a3021c 100644 --- a/integration_tests/ports/waiting_for_inputs_test.go +++ b/integration_tests/ports/waiting_for_inputs_test.go @@ -3,12 +3,15 @@ package ports import ( "testing" + "github.com/hovsep/fmesh/internal/testutil" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/hovsep/fmesh" "github.com/hovsep/fmesh/component" "github.com/hovsep/fmesh/cycle" "github.com/hovsep/fmesh/signal" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" ) func Test_WaitingForInputs(t *testing.T) { @@ -22,7 +25,7 @@ func Test_WaitingForInputs(t *testing.T) { name: "waiting for longer chain", setupFM: func() *fmesh.FMesh { getDoubler := func(name string) *component.Component { - return mustComponent(name, + return testutil.MustComponent(name, component.WithInputs("i1"), component.WithOutputs("o1"), component.WithDescription("This component just doubles the input"), @@ -39,7 +42,7 @@ func Test_WaitingForInputs(t *testing.T) { d4 := getDoubler("d4") d5 := getDoubler("d5") - s := mustComponent("sum", + s := testutil.MustComponent("sum", component.WithInputs("i1", "i2"), component.WithOutputs("o1"), component.WithDescription("This component just sums 2 inputs"), @@ -76,7 +79,7 @@ func Test_WaitingForInputs(t *testing.T) { panic(err) } - fm := mustFMesh("fm", fmesh.WithConfig(fmesh.Config{ + fm := testutil.MustFMesh("fm", fmesh.WithConfig(fmesh.Config{ ErrorHandlingStrategy: fmesh.StopOnFirstErrorOrPanic, CyclesLimit: 5, })) diff --git a/integration_tests/state/basic_test.go b/integration_tests/state/basic_test.go index b20f442..5f81dbe 100644 --- a/integration_tests/state/basic_test.go +++ b/integration_tests/state/basic_test.go @@ -4,31 +4,18 @@ import ( "math/rand" "testing" + "github.com/hovsep/fmesh/internal/testutil" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/hovsep/fmesh" "github.com/hovsep/fmesh/component" "github.com/hovsep/fmesh/cycle" "github.com/hovsep/fmesh/port" "github.com/hovsep/fmesh/signal" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" ) -func mustComponent(name string, opts ...component.Option) *component.Component { - c, err := component.New(name, opts...) - if err != nil { - panic(err) - } - return c -} - -func mustFMesh(name string, opts ...fmesh.Option) *fmesh.FMesh { - fm, err := fmesh.New(name, opts...) - if err != nil { - panic(err) - } - return fm -} - func Test_State(t *testing.T) { tests := []struct { name string @@ -39,7 +26,7 @@ func Test_State(t *testing.T) { { name: "stateful counter", setupFM: func() *fmesh.FMesh { - producer := mustComponent("producer", + producer := testutil.MustComponent("producer", component.WithInputs("demand_rate"), component.WithOutputs("signal_out"), component.WithDescription("produces some signals"), @@ -56,7 +43,7 @@ func Test_State(t *testing.T) { }), ) - counter := mustComponent("stateful_counter", + counter := testutil.MustComponent("stateful_counter", component.WithInputs("bypass_in"), component.WithOutputs("bypass_out"), component.WithDescription("counts all observed signals and bypasses them down the stream"), @@ -79,7 +66,7 @@ func Test_State(t *testing.T) { }), ) - consumer := mustComponent("consumer", + consumer := testutil.MustComponent("consumer", component.WithInputs("signal_in", "start"), component.WithOutputs("consumed_signals", "demand_rate"), component.WithDescription("consumes signals"), @@ -118,7 +105,7 @@ func Test_State(t *testing.T) { panic(err) } - fm := mustFMesh("fm", fmesh.WithConfig(fmesh.Config{ + fm := testutil.MustFMesh("fm", fmesh.WithConfig(fmesh.Config{ ErrorHandlingStrategy: fmesh.StopOnFirstErrorOrPanic, CyclesLimit: 10000, })) diff --git a/hook/hook_group.go b/internal/hook/hook_group.go similarity index 100% rename from hook/hook_group.go rename to internal/hook/hook_group.go diff --git a/hook/hook_group_test.go b/internal/hook/hook_group_test.go similarity index 100% rename from hook/hook_group_test.go rename to internal/hook/hook_group_test.go diff --git a/internal/testutil/testutil.go b/internal/testutil/testutil.go new file mode 100644 index 0000000..18d7318 --- /dev/null +++ b/internal/testutil/testutil.go @@ -0,0 +1,61 @@ +// Package testutil provides panic-on-error constructors shared by the +// integration test suites. Not usable from in-package unit tests of fmesh or +// port (import cycle); those keep small local equivalents. +package testutil + +import ( + "github.com/hovsep/fmesh" + "github.com/hovsep/fmesh/component" + "github.com/hovsep/fmesh/port" + "github.com/hovsep/fmesh/signal" +) + +// MustComponent builds a component or panics. +func MustComponent(name string, opts ...component.Option) *component.Component { + c, err := component.New(name, opts...) + if err != nil { + panic(err) + } + return c +} + +// MustFMesh builds a mesh or panics. +func MustFMesh(name string, opts ...fmesh.Option) *fmesh.FMesh { + fm, err := fmesh.New(name, opts...) + if err != nil { + panic(err) + } + return fm +} + +// MustInputPort builds an input port or panics. +func MustInputPort(name string, opts ...port.Option) *port.Port { + p, err := port.NewInput(name, opts...) + if err != nil { + panic(err) + } + return p +} + +// MustOutputPort builds an output port or panics. +func MustOutputPort(name string, opts ...port.Option) *port.Port { + p, err := port.NewOutput(name, opts...) + if err != nil { + panic(err) + } + return p +} + +// MustPutSignals puts signals on a port or panics. +func MustPutSignals(p *port.Port, signals ...*signal.Signal) { + if err := p.PutSignals(signals...); err != nil { + panic(err) + } +} + +// MustPipeTo pipes src to dsts or panics. +func MustPipeTo(src *port.Port, dsts ...*port.Port) { + if err := src.PipeTo(dsts...); err != nil { + panic(err) + } +} diff --git a/meta/doc.go b/meta/doc.go new file mode 100644 index 0000000..4dd47f3 --- /dev/null +++ b/meta/doc.go @@ -0,0 +1,7 @@ +// Package meta provides [Labels] (string key/value) and [Scalars] +// (string→float64) metadata carried by signals, ports, components, cycles +// and the mesh itself. +// +// Both types mutate in place; Keys and Values return sorted slices for +// determinism, and Merge is the one non-mutating method on each type. +package meta diff --git a/port/doc.go b/port/doc.go new file mode 100644 index 0000000..fd91d4b --- /dev/null +++ b/port/doc.go @@ -0,0 +1,8 @@ +// Package port provides [Port], the data endpoint of a component, along with +// [Group] (ordered port list) and [Collection] (name-keyed port map). +// +// Ports are connected output→input with [Port.PipeTo]; [Port.Flush] fans a +// port's signal buffer out through its pipes and clears the source. Unlike +// the copy-on-write signal package, port types mutate in place: Set*, Add* +// and Remove* methods modify the receiver. +package port diff --git a/port/hooks.go b/port/hooks.go index 1d73709..56a8420 100644 --- a/port/hooks.go +++ b/port/hooks.go @@ -1,7 +1,7 @@ package port import ( - "github.com/hovsep/fmesh/hook" + "github.com/hovsep/fmesh/internal/hook" "github.com/hovsep/fmesh/signal" ) @@ -40,8 +40,8 @@ type Hooks struct { onOutboundPipe *hook.Group[*OutboundPipeContext] } -// NewHooks creates a new hooks registry. -func NewHooks() *Hooks { +// newHooks creates a new hooks registry. +func newHooks() *Hooks { return &Hooks{ onSignalsAdded: hook.NewGroup[*SignalsAddedContext](), onClear: hook.NewGroup[*ClearContext](), diff --git a/port/port.go b/port/port.go index 5a29563..80d6a38 100644 --- a/port/port.go +++ b/port/port.go @@ -45,7 +45,7 @@ func NewInput(name string, opts ...Option) (*Port, error) { scalars: meta.NewScalars(), pipes: NewGroup(), signals: signal.NewGroup(), - hooks: NewHooks(), + hooks: newHooks(), } for _, opt := range opts { if err := opt(p); err != nil { @@ -64,7 +64,7 @@ func NewOutput(name string, opts ...Option) (*Port, error) { scalars: meta.NewScalars(), pipes: NewGroup(), signals: signal.NewGroup(), - hooks: NewHooks(), + hooks: newHooks(), } for _, opt := range opts { if err := opt(p); err != nil { diff --git a/port/port_test.go b/port/port_test.go index b55ecd9..5817fd9 100644 --- a/port/port_test.go +++ b/port/port_test.go @@ -8,6 +8,9 @@ import ( "github.com/stretchr/testify/require" ) +// Local helpers: this in-package test cannot import internal/testutil +// (testutil imports port). + // mustInput is a test helper that panics if NewInput returns an error. func mustInput(name string) *Port { p, err := NewInput(name) diff --git a/runtime.go b/runtime.go index 87c0bff..50c13fb 100644 --- a/runtime.go +++ b/runtime.go @@ -21,15 +21,15 @@ func newRuntimeInfo(historyLimit int) *RuntimeInfo { } } -// MarkStarted sets when fmesh is started running. -func (r *RuntimeInfo) MarkStarted() { +// markStarted sets when fmesh is started running. +func (r *RuntimeInfo) markStarted() { if r.StartedAt.IsZero() { r.StartedAt = time.Now() } } -// MarkStopped sets when the fmesh is stopped running. -func (r *RuntimeInfo) MarkStopped() { +// markStopped sets when the fmesh is stopped running. +func (r *RuntimeInfo) markStopped() { if r.StoppedAt.IsZero() { r.StoppedAt = time.Now() }