Don't purge all resources when loading fixtures a second time in tests

examples/integration/config.dist.yml

@@ -5,7 +5,9 @@ app:
     project: gosoline
     family: gosoline
     group: grp
-
+  namespace: "{app.tags.project}.{app.env}.{app.tags.family}.{app.tags.group}"
+  model_id:
+    domain_pattern: "{app.tags.project}.{app.env}.{app.tags.family}.{app.tags.group}"
 
 db:
   default:

examples/more_details/stream-consumer-test/stream_consumer_test.go

@@ -1,5 +1,4 @@
 //go:build integration && fixtures
-// +build integration,fixtures
 
 package stream_consumer_test
 
@@ -25,6 +24,7 @@ func (s *ConsumerTestSuite) SetupSuite() []suite.Option {
 		suite.WithLogLevel("debug"),
 		suite.WithConfigFile("../stream-consumer/config.dist.yml"),
 		suite.WithModule("consumerModule", stream.NewConsumer("uintConsumer", consumer.NewConsumer())),
+		suite.WithSharedEnvironment(),
 	}
 }
 
@@ -57,20 +57,29 @@ func (s *ConsumerTestSuite) TestSuccess() *suite.StreamTestCase {
 	}
 }
 
-func (s *ConsumerTestSuite) TestSuccessTwice(app suite.AppUnderTest) {
-	consumer := s.Env().StreamInput("consumerInput")
-	s.NotNil(consumer)
-
-	consumer.Publish(mdl.Box(uint(2)), nil)
-	consumer.Publish(mdl.Box(uint(3)), nil)
-
-	app.Stop()
-	app.WaitDone()
+func (s *ConsumerTestSuite) TestSuccessTwice() *suite.StreamTestCase {
+	return &suite.StreamTestCase{
+		Input: map[string][]suite.StreamTestCaseInput{
+			"consumerInput": {
+				{
+					Attributes: nil,
+					Body:       mdl.Box(uint(2)),
+				},
+				{
+					Attributes: nil,
+					Body:       mdl.Box(uint(3)),
+				},
+			},
+		},
+		Assert: func() error {
+			var result int
+			s.Env().StreamOutput("publisher-outputEvent").Unmarshal(0, &result)
+			s.Equal(3, result)
 
-	var result int
-	s.Env().StreamOutput("publisher-outputEvent").Unmarshal(0, &result)
-	s.Equal(3, result)
+			s.Env().StreamOutput("publisher-outputEvent").Unmarshal(1, &result)
+			s.Equal(4, result)
 
-	s.Env().StreamOutput("publisher-outputEvent").Unmarshal(1, &result)
-	s.Equal(4, result)
+			return nil
+		},
+	}
 }

examples/more_details/stream-consumer/config.dist.yml

@@ -5,6 +5,9 @@ app:
     project: project
     family: family
     group: grp
+  namespace: "{app.tags.project}.{app.env}.{app.tags.family}.{app.tags.group}"
+  model_id:
+    domain_pattern: "{app.tags.project}.{app.env}.{app.tags.family}.{app.tags.group}"
 
 mdlsub:
   publishers:

pkg/blob/fixture_writer.go

@@ -63,7 +63,9 @@ func NewBlobFixtureWriter(ctx context.Context, config cfg.Config, logger log.Log
 		return nil, fmt.Errorf("can not create blob store: %w", err)
 	}
 
-	return NewBlobFixtureWriterWithInterfaces(logger, bri, reader, store), nil
+	resourceId := fmt.Sprintf("blob/%s", storeName)
+
+	return fixtures.NewManagedFixtureWriter(NewBlobFixtureWriterWithInterfaces(logger, bri, reader, store), resourceId), nil
 }
 
 func NewBlobFixtureWriterWithInterfaces(logger log.Logger, batchRunner BatchRunner, reader FixtureReader, store Store) fixtures.FixtureWriter {

pkg/cloud/aws/credentials_integrationtest.go

@@ -1,5 +1,4 @@
 //go:build integration
-// +build integration
 
 package aws
 

pkg/cloud/aws/credentials_integrationtest_test.go

@@ -1,5 +1,4 @@
 //go:build integration
-// +build integration
 
 package aws_test
 

pkg/conc/ddb/ddb_lock.go

@@ -13,7 +13,7 @@ import (
 
 //go:generate go run github.com/vektra/mockery/v2 --name LockManager
 type LockManager interface {
-	RenewLock(ctx context.Context, lockTime time.Duration, resource string, token string) error
+	RenewLock(ctx context.Context, lockTime time.Duration, resource string, token string) (expiry time.Time, err error)
 	ReleaseLock(ctx context.Context, resource string, token string) error
 }
 
@@ -35,7 +35,7 @@ func NewDdbLockFromInterfaces(
 	ctx context.Context,
 	resource string,
 	token string,
-	expires int64,
+	expires time.Time,
 ) *ddbLock {
 	return &ddbLock{
 		manager:  manager,
@@ -44,44 +44,44 @@ func NewDdbLockFromInterfaces(
 		ctx:      ctx,
 		resource: resource,
 		token:    token,
-		expires:  expires,
+		expires:  expires.UnixMicro(),
 		released: conc.NewSignalOnce(),
 	}
 }
 
 func (l *ddbLock) Renew(ctx context.Context, lockTime time.Duration) error {
 	if l == nil {
-		return conc.ErrNotOwned
+		return conc.ErrLockNotOwned
 	}
 
-	err := l.manager.RenewLock(ctx, lockTime, l.resource, l.token)
+	expiry, err := l.manager.RenewLock(ctx, lockTime, l.resource, l.token)
 
 	if err == nil {
-		atomic.StoreInt64(&l.expires, l.clock.Now().Add(lockTime).Unix())
+		atomic.StoreInt64(&l.expires, expiry.UnixMicro())
 	}
 
 	return err
 }
 
 func (l *ddbLock) Release() error {
 	if l == nil {
-		return conc.ErrNotOwned
+		return conc.ErrLockNotOwned
 	}
 
 	// stop the debug thread if needed
 	l.released.Signal()
 
-	deadline := time.Unix(atomic.LoadInt64(&l.expires), 0)
+	deadline := time.UnixMicro(atomic.LoadInt64(&l.expires))
 	remainingLockTime := deadline.Sub(l.clock.Now())
 
 	if remainingLockTime <= 0 {
-		return conc.ErrNotOwned
+		return conc.ErrLockNotOwned
 	}
 
 	done := make(chan struct{})
 	defer close(done)
 
-	// we should always release the lock, even when our parent gets cancelled.
+	// we should always release the lock, even when our parent gets canceled.
 	// if we don't manage to do this until it expires anyway, there is no further point in trying.
 	ctx, cancel := exec.WithManualCancelContext(l.ctx)
 	go func() {
@@ -100,18 +100,20 @@ func (l *ddbLock) Release() error {
 }
 
 func (l *ddbLock) runWatcher() {
+	t := l.clock.NewTimer(l.expiresIn())
+	defer t.Stop()
+
 	for {
-		expires := atomic.LoadInt64(&l.expires)
-		now := l.clock.Now()
+		expiresIn := l.expiresIn()
 
-		if expires < now.Unix() {
+		if expiresIn <= 0 {
 			break
 		}
 
-		t := time.NewTimer(time.Unix(expires, 0).Sub(now))
+		t.Reset(expiresIn)
 
 		select {
-		case <-t.C:
+		case <-t.Chan():
 			continue
 		case <-l.released.Channel():
 			return
@@ -123,3 +125,10 @@ func (l *ddbLock) runWatcher() {
 		"ddb_lock_resource": l.resource,
 	}).Warn(l.ctx, "failed to release or renew the lock before the timeout")
 }
+
+func (l *ddbLock) expiresIn() time.Duration {
+	expires := time.UnixMicro(atomic.LoadInt64(&l.expires))
+	now := l.clock.Now()
+
+	return expires.Sub(now)
+}