Merge remote-tracking branch 'pokt/main' into feat/observable-map

* pokt/main:
  chore: enforce go standard interface implementation registration (#87)
  [E2E] Add Regression Testing for Send E2E Feature Test (#84)
  feat: seperate tests from go_develop (#89)
  [Observable] chore: observable touchup (#83)

.github/workflows/reviewdog.yml

@@ -20,6 +20,20 @@ jobs:
           fail_on_error: true
           pattern: TODO_IN_THIS_
 
+  check_non_standard_interface_implementations:
+    name: Check for non-standard interface implementation statements
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: pokt-network/action-fail-on-found@v1
+        with:
+          github_token: ${{ secrets.github_token }}
+          reporter: github-pr-review
+          level: error
+          fail_on_error: true
+          pattern: var _ .* = &.*{}
+          ignore: .github,.git
+
   # More info: https://github.com/reviewdog/action-misspell
   check_misspell:
     name: Check misspelling
@@ -31,4 +45,4 @@ jobs:
           github_token: ${{ secrets.github_token }}
           reporter: github-check
           level: warning
-          locale: "US"
+          locale: "US"

Makefile

@@ -123,7 +123,10 @@ go_mockgen: ## Use `mockgen` to generate mocks used for testing purposes of all
 	go generate ./x/supplier/types/
 
 .PHONY: go_develop
-go_develop: proto_regen go_mockgen go_test ## Generate protos, mocks and run all tests
+go_develop: proto_regen go_mockgen ## Generate protos and mocks
+
+.PHONY: go_develop_and_test
+go_develop_and_test: go_develop go_test ## Generate protos, mocks and run all tests
 
 #############
 ### TODOS ###

docs/pkg/observable/README.md

@@ -1,6 +1,6 @@
-## `pkg/observable` Package
+## `pocket/pkg/observable` Package
 
-The `pkg/observable` package provides a lightweight and straightforward mechanism to handle asynchronous notifications using the Observer pattern. This is achieved through two primary interfaces: `Observable` and `Observer`.
+The `pocket/pkg/observable` package provides a lightweight and straightforward mechanism to handle asynchronous notifications using the Observer pattern. This is achieved through two primary interfaces: `Observable` and `Observer`.
 
 ## Overview
 

docs/template/pkg/README.md

@@ -1,6 +1,3 @@
-Certainly! I've added a section named "Architecture Diagrams" in the documentation template below:
-
-```markdown
 # Package [PackageName]
 
 > Brief one-liner or quote about what this package does.
@@ -16,11 +13,21 @@ Provide a few sentences about the purpose and functionality of this package. Con
 
 Visual representations often make it easier to understand the design and flow of a package. Below are the architecture diagrams that explain the high-level structure and interactions in this package:
 
-![Architecture Overview](./path-to-diagram1.png)
+```mermaid
+---
+title: Architecture Overview
+---
+flowchart
+```
 
 > **Figure 1**: Brief description about what this diagram represents.
 
-![Another Diagram](./path-to-diagram2.png)
+```mermaid
+---
+title: Another Diagram
+---
+flowchart
+```
 
 > **Figure 2**: Brief description about what this other diagram represents.
 

e2e/tests/init_test.go

@@ -5,26 +5,37 @@ package e2e
 import (
 	"fmt"
 	"regexp"
+	"strconv"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/regen-network/gocuke"
 	"github.com/stretchr/testify/require"
 )
 
-var addrRe *regexp.Regexp
+var (
+	addrRe           *regexp.Regexp
+	amountRe         *regexp.Regexp
+	accNameToAddrMap = make(map[string]string)
+	keyRingFlag      = "--keyring-backend=test"
+)
 
 func init() {
-	addrRe = regexp.MustCompile(`address:\s+(pokt1\w+)`)
+	addrRe = regexp.MustCompile(`address: (\S+)\s+name: (\S+)`)
+	amountRe = regexp.MustCompile(`amount: "(.+?)"\s+denom: upokt`)
 }
 
 type suite struct {
 	gocuke.TestingT
-	pocketd *pocketdBin
+	pocketd       *pocketdBin
+	scenarioState map[string]any // temporary state for each scenario
 }
 
 func (s *suite) Before() {
 	s.pocketd = new(pocketdBin)
+	s.scenarioState = make(map[string]any)
+	s.buildAddrMap()
 }
 
 // TestFeatures runs the e2e tests specified in any .features files in this directory
@@ -56,17 +67,15 @@ func (s *suite) TheUserShouldBeAbleToSeeStandardOutputContaining(arg1 string) {
 	}
 }
 
-func (s *suite) TheUserSendsUpoktToAnotherAddress(amount int64) {
-	addrs := s.getAddresses()
+func (s *suite) TheUserSendsUpoktFromAccountToAccount(amount int64, accName1, accName2 string) {
 	args := []string{
 		"tx",
 		"bank",
 		"send",
-		addrs[0],
-		addrs[1],
+		accNameToAddrMap[accName1],
+		accNameToAddrMap[accName2],
 		fmt.Sprintf("%dupokt", amount),
-		"--keyring-backend",
-		"test",
+		keyRingFlag,
 		"-y",
 	}
 	res, err := s.pocketd.RunCommandOnHost("", args...)
@@ -76,20 +85,78 @@ func (s *suite) TheUserSendsUpoktToAnotherAddress(amount int64) {
 	s.pocketd.result = res
 }
 
-func (s *suite) getAddresses() [2]string {
-	var strs [2]string
+func (s *suite) TheAccountHasABalanceGreaterThanUpokt(accName string, amount int64) {
+	bal := s.getAccBalance(accName)
+	if int64(bal) < amount {
+		s.Fatalf("account %s does not have enough upokt: %d < %d", accName, bal, amount)
+	}
+	s.scenarioState[accName] = bal // save the balance for later
+}
+
+func (s *suite) AnAccountExistsFor(accName string) {
+	bal := s.getAccBalance(accName)
+	s.scenarioState[accName] = bal // save the balance for later
+}
+
+func (s *suite) TheAccountBalanceOfShouldBeUpoktThanBefore(accName string, amount int64, condition string) {
+	prev, ok := s.scenarioState[accName]
+	if !ok {
+		s.Fatalf("no previous balance found for %s", accName)
+	}
+
+	bal := s.getAccBalance(accName)
+	switch condition {
+	case "more":
+		if bal <= prev.(int) {
+			s.Fatalf("account %s expected to have more upokt but: %d <= %d", accName, bal, prev)
+		}
+	case "less":
+		if bal >= prev.(int) {
+			s.Fatalf("account %s expected to have less upokt but: %d >= %d", accName, bal, prev)
+		}
+	default:
+		s.Fatalf("unknown condition %s", condition)
+	}
+}
+
+func (s *suite) TheUserShouldWaitForSeconds(dur int64) {
+	time.Sleep(time.Duration(dur) * time.Second)
+}
+
+func (s *suite) buildAddrMap() {
+	s.Helper()
 	res, err := s.pocketd.RunCommand(
-		"keys", "list", "--keyring-backend", "test",
+		"keys", "list", keyRingFlag,
 	)
 	if err != nil {
 		s.Fatalf("error getting keys: %s", err)
 	}
 	matches := addrRe.FindAllStringSubmatch(res.Stdout, -1)
-	if len(matches) >= 2 {
-		strs[0] = matches[0][1]
-		strs[1] = matches[len(matches)-1][1]
-	} else {
-		s.Fatalf("could not find two addresses in output: %s", res.Stdout)
+	for _, match := range matches {
+		name := match[2]
+		address := match[1]
+		accNameToAddrMap[name] = address
+	}
+}
+
+func (s *suite) getAccBalance(accName string) int {
+	s.Helper()
+	args := []string{
+		"query",
+		"bank",
+		"balances",
+		accNameToAddrMap[accName],
+	}
+	res, err := s.pocketd.RunCommandOnHost("", args...)
+	if err != nil {
+		s.Fatalf("error getting balance: %s", err)
+	}
+	s.pocketd.result = res
+	match := amountRe.FindStringSubmatch(res.Stdout)
+	if len(match) < 2 {
+		s.Fatalf("no balance found for %s", accName)
 	}
-	return strs
+	found, err := strconv.Atoi(match[1])
+	require.NoError(s, err)
+	return found
 }

e2e/tests/node.go

@@ -24,7 +24,7 @@ func init() {
 		defaultRPCURL = fmt.Sprintf("tcp://%s:%d", defaultRPCHost, defaultRPCPort)
 	}
 	if defaultHome == "" {
-		defaultHome = "./localnet/pocketd"
+		defaultHome = "../../localnet/pocketd"
 	}
 }
 
@@ -42,7 +42,7 @@ type PocketClient interface {
 }
 
 // Ensure that pocketdBin struct fulfills PocketClient
-var _ PocketClient = &pocketdBin{}
+var _ PocketClient = (*pocketdBin)(nil)
 
 // pocketdBin holds the reults of the last command that was run
 type pocketdBin struct {

e2e/tests/send.feature

@@ -2,7 +2,12 @@ Feature: Tx Namespace
 
   Scenario: User can send uPOKT
     Given the user has the pocketd binary installed
-    When the user sends 10000 uPOKT to another address
+    And the account "app1" has a balance greater than "1000" uPOKT
+    And an account exists for "app2"
+    When the user sends "1000" uPOKT from account "app1" to account "app2"
     Then the user should be able to see standard output containing "txhash:"
     And the user should be able to see standard output containing "code: 0"
     And the pocketd binary should exit without error
+    And the user should wait for "5" seconds
+    And the account balance of "app1" should be "1000" uPOKT "less" than before
+    And the account balance of "app2" should be "1000" uPOKT "more" than before

pkg/observable/channel/observable.go

@@ -2,8 +2,9 @@ package channel
 
 import (
 	"context"
-	"pocket/pkg/observable"
 	"sync"
+
+	"pocket/pkg/observable"
 )
 
 // TODO_DISCUSS: what should this be? should it be configurable? It seems to be most
@@ -12,7 +13,7 @@ import (
 // defaultSubscribeBufferSize is the buffer size of a observable's publish channel.
 const defaultPublishBufferSize = 50
 
-var _ observable.Observable[any] = &channelObservable[any]{}
+var _ observable.Observable[any] = (*channelObservable[any])(nil)
 
 // option is a function which receives and can modify the channelObservable state.
 type option[V any] func(obs *channelObservable[V])
@@ -50,7 +51,7 @@ func NewObservable[V any](opts ...option[V]) (observable.Observable[V], chan<- V
 	}
 
 	// start listening to the publishCh and emit values to observers
-	go obs.goPublish(obs.publishCh)
+	go obs.goPublish()
 
 	return obs, obs.publishCh
 }
@@ -63,6 +64,15 @@ func WithPublisher[V any](publishCh chan V) option[V] {
 	}
 }
 
+// Next synchronously returns the next value from the observable.
+func (obsvbl *channelObservable[V]) Next(ctx context.Context) V {
+	tempObserver := obsvbl.Subscribe(ctx)
+	defer tempObserver.Unsubscribe()
+
+	val := <-tempObserver.Ch()
+	return val
+}
+
 // Subscribe returns an observer which is notified when the publishCh channel
 // receives a value.
 func (obsvbl *channelObservable[V]) Subscribe(ctx context.Context) observable.Observer[V] {
@@ -110,8 +120,8 @@ func (obsvbl *channelObservable[V]) unsubscribeAll() {
 
 // goPublish to the publishCh and notify observers when values are received.
 // This function is blocking and should be run in a goroutine.
-func (obsvbl *channelObservable[V]) goPublish(publisher <-chan V) {
-	for notification := range publisher {
+func (obsvbl *channelObservable[V]) goPublish() {
+	for notification := range obsvbl.publishCh {
 		// Copy currentObservers to avoid holding the lock while notifying them.
 		// New or existing Observers may (un)subscribe while this notification
 		// is being fanned out.
@@ -154,9 +164,12 @@ func (obsvbl *channelObservable[V]) copyObservers() (observers []*channelObserve
 
 // goUnsubscribeOnDone unsubscribes from the subscription when the context is done.
 // It is a blocking function and intended to be called in a goroutine.
-func goUnsubscribeOnDone[V any](ctx context.Context, subscription observable.Observer[V]) {
+func goUnsubscribeOnDone[V any](ctx context.Context, observer observable.Observer[V]) {
 	<-ctx.Done()
-	subscription.Unsubscribe()
+	if observer.IsClosed() {
+		return
+	}
+	observer.Unsubscribe()
 }
 
 // onUnsubscribe returns a function that removes a given observer from the

pkg/observable/channel/observable_test.go

@@ -38,7 +38,7 @@ func TestChannelObservable_NotifyObservers(t *testing.T) {
 
 	tests := []test{
 		{
-			name:            "nil publisher",
+			name:            "nil publisher (default buffer size)",
 			publishCh:       nil,
 			inputs:          inputs,
 			expectedOutputs: inputs,
@@ -55,7 +55,13 @@ func TestChannelObservable_NotifyObservers(t *testing.T) {
 			inputs:          inputs,
 			expectedOutputs: inputs,
 		},
-		// TODO_INCOMPLETE: publisher channels which are full are proving harder to test
+		{
+			name:            "empty buffered len 1000 publisher",
+			publishCh:       make(chan int, 1000),
+			inputs:          inputs,
+			expectedOutputs: inputs,
+		},
+		// TODO_INCOMPLETE(#81): publisher channels which are full are proving harder to test
 		// robustly (no flakiness); perhaps it has to do with the lack of some
 		// kind of guarantee about the receiver order on the consumer side.
 		//
@@ -156,7 +162,7 @@ func TestChannelObservable_NotifyObservers(t *testing.T) {
 			err := group.Wait()
 			require.NoError(t, err)
 
-			// unsubscribing should unsubscribeAll obsvr channel(s)
+			// unsubscribing should close observer channel(s)
 			for _, observer := range observers {
 				observer.Unsubscribe()
 
@@ -205,6 +211,8 @@ func TestChannelObservable_UnsubscribeObservers(t *testing.T) {
 			},
 		},
 		{
+			// NOTE: this will log a warning that can be ignored:
+			// >  redundant unsubscribe: observer is closed
 			name: "cancel then unsubscribe",
 			lifecycleFn: func() observable.Observer[int] {
 				observer := obsvbl.Subscribe(ctx)
@@ -215,6 +223,8 @@ func TestChannelObservable_UnsubscribeObservers(t *testing.T) {
 			},
 		},
 		{
+			// NOTE: this will log a warning that can be ignored:
+			// >  redundant unsubscribe: observer is closed
 			name: "unsubscribe then cancel",
 			lifecycleFn: func() observable.Observer[int] {
 				observer := obsvbl.Subscribe(ctx)