Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add overridden duration timeout to StreamTestKit #1468

Merged
merged 3 commits into from
Jan 25, 2025
Merged

Add overridden duration timeout to StreamTestKit #1468

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
4763be2
Add overridden duration timeout to StreamTestKit
mdedetrich Sep 9, 2024
f2070ba
add license to file
pjfanning Jan 25, 2025
c1bc36b
Update pr-1468-override-timeout-stream-testkit.backwards.excludes
pjfanning Jan 25, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 19 additions & 0 deletions ...lters/1.1.1.backwards.excludes/pr-1468-override-timeout-stream-testkit.backwards.excludes
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,19 @@
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.

# StreamTestKit.assertNoChildren is an internal API and can be changed
ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertNoChildren")
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When adding mima excludes, please add the reason why it is acceptable. In this case, the reason might be we don't promise binary compatiblity for -testkit artifacts (https://pekko.apache.org/docs/pekko/current/common/binary-compatibility-rules.html), and that this method is private[testkit]

20 changes: 19 additions & 1 deletion stream-testkit/src/main/scala/org/apache/pekko/stream/testkit/javadsl/StreamTestKit.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,10 @@ import pekko.stream.{ Materializer, SystemMaterializer }
import pekko.stream.impl.PhasedFusingActorMaterializer
import pekko.stream.testkit.scaladsl

import java.time.Duration
import java.util.concurrent.TimeUnit
import scala.concurrent.duration.FiniteDuration

object StreamTestKit {

/**
Expand All @@ -29,7 +33,21 @@ object StreamTestKit {
def assertAllStagesStopped(mat: Materializer): Unit =
mat match {
case impl: PhasedFusingActorMaterializer =>
scaladsl.StreamTestKit.assertNoChildren(impl.system, impl.supervisor)
scaladsl.StreamTestKit.assertNoChildren(impl.system, impl.supervisor, None)
case _ =>
}

/**
* Assert that there are no stages running under a given materializer.
* Usually this assertion is run after a test-case to check that all of the
* stages have terminated successfully with an overridden duration that ignores
* `stream.testkit.all-stages-stopped-timeout`.
*/
def assertAllStagesStopped(mat: Materializer, overrideTimeout: Duration): Unit =
mat match {
case impl: PhasedFusingActorMaterializer =>
scaladsl.StreamTestKit.assertNoChildren(impl.system, impl.supervisor,
Some(FiniteDuration(overrideTimeout.toMillis, TimeUnit.MILLISECONDS)))
case _ =>
}

Expand Down
31 changes: 27 additions & 4 deletions stream-testkit/src/main/scala/org/apache/pekko/stream/testkit/scaladsl/StreamTestKit.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,12 +35,30 @@ object StreamTestKit {
* This assertion is useful to check that all of the stages have
* terminated successfully.
*/
def assertAllStagesStopped[T](block: => T, overrideTimeout: FiniteDuration)(implicit materializer: Materializer): T =
materializer match {
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There is code duplication here, I can clean it up if the premise of this PR is acceptable.

case impl: PhasedFusingActorMaterializer =>
stopAllChildren(impl.system, impl.supervisor)
val result = block
assertNoChildren(impl.system, impl.supervisor, Some(overrideTimeout))
result
case _ => block
}

/**
* Asserts that after the given code block is ran, no stages are left over
* that were created by the given materializer with an overridden duration
* that ignores `stream.testkit.all-stages-stopped-timeout`.
*
* This assertion is useful to check that all of the stages have
* terminated successfully.
*/
def assertAllStagesStopped[T](block: => T)(implicit materializer: Materializer): T =
materializer match {
case impl: PhasedFusingActorMaterializer =>
stopAllChildren(impl.system, impl.supervisor)
val result = block
assertNoChildren(impl.system, impl.supervisor)
assertNoChildren(impl.system, impl.supervisor, None)
result
case _ => block
}
Expand All @@ -53,10 +71,15 @@ object StreamTestKit {
}

/** INTERNAL API */
@InternalApi private[testkit] def assertNoChildren(sys: ActorSystem, supervisor: ActorRef): Unit = {
@InternalApi private[testkit] def assertNoChildren(sys: ActorSystem, supervisor: ActorRef,
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

add an overloaded method here to avoid the mima issue?

overrideTimeout: Option[FiniteDuration]): Unit = {
val probe = TestProbe()(sys)
val c = sys.settings.config.getConfig("pekko.stream.testkit")
val timeout = c.getDuration("all-stages-stopped-timeout", MILLISECONDS).millis

val timeout = overrideTimeout.getOrElse {
val c = sys.settings.config.getConfig("pekko.stream.testkit")
c.getDuration("all-stages-stopped-timeout", MILLISECONDS).millis
}

probe.within(timeout) {
try probe.awaitAssert {
supervisor.tell(StreamSupervisor.GetChildren, probe.ref)
Expand Down
40 changes: 40 additions & 0 deletions stream-testkit/src/test/scala/org/apache/pekko/stream/testkit/StreamConfiguration.scala
View file
Open in desktop
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This might look a bit un-orthodox by Pekko standards, but it is the idiomatic way to override a configuration when in ScalaTest (which is what StreamSpec extends).

The defaault behaviour is still the same (i.e. getting the value from pekko.stream.testkit.all-stages-stopped-timeout) but if you decide to override it then you can do it the ScalaTest specific way.

Original file line number Diff line number Diff line change
@@ -0,0 +1,40 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package org.apache.pekko.stream.testkit

import org.apache.pekko.testkit.TestKitBase
import org.scalatest.time.{ Millis, Span }

import java.util.concurrent.TimeUnit

trait StreamConfiguration extends TestKitBase {
case class StreamConfig(allStagesStoppedTimeout: Span = Span({
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The premise here is that by default we get the timeout value from Pekko typesafe config (as is expected when using Pekko) however if a user wants to override this then its done the ScalaTest idiomatic way (which is overriding a def in a trait, similar to how PatienceConfig works).

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It is possible to omit units and support more flexibility, but the code is slightly complex and for reference only.

import scala.jdk.DurationConverters._
import org.scalatest.time.Span._
system.settings.config.getDuration("pekko.stream.testkit.all-stages-stopped-timeout").toScala.convertDurationToSpan

I just saw this {} and wanted to make some changes. I looked at the Span code and found that it has convertDurationoSpan. You can do whatever you want.

val c = system.settings.config.getConfig("pekko.stream.testkit")
c.getDuration("all-stages-stopped-timeout", TimeUnit.MILLISECONDS)
}, Millis))

private val defaultStreamConfig = StreamConfig()

/**
* The default `StreamConfig` which is derived from the Actor System's `pekko.stream.testkit.all-stages-stopped-timeout`
* configuration value. If you want to provide a different StreamConfig for specific tests without having to re-specify
* `pekko.stream.testkit.all-stages-stopped-timeout` then you can override this value.
*/
implicit def streamConfig: StreamConfig = defaultStreamConfig

}
8 changes: 6 additions & 2 deletions stream-testkit/src/test/scala/org/apache/pekko/stream/testkit/StreamSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,10 @@ import org.scalatest.Failed

import com.typesafe.config.{ Config, ConfigFactory }

abstract class StreamSpec(_system: ActorSystem) extends PekkoSpec(_system) {
import java.util.concurrent.TimeUnit

abstract class StreamSpec(_system: ActorSystem) extends PekkoSpec(_system) with StreamConfiguration {

def this(config: Config) =
this(
ActorSystem(
Expand Down Expand Up @@ -73,7 +76,8 @@ abstract class StreamSpec(_system: ActorSystem) extends PekkoSpec(_system) {
case impl: PhasedFusingActorMaterializer =>
stopAllChildren(impl.system, impl.supervisor)
val result = test.apply()
assertNoChildren(impl.system, impl.supervisor)
assertNoChildren(impl.system, impl.supervisor,
Some(FiniteDuration(streamConfig.allStagesStoppedTimeout.millisPart, TimeUnit.MILLISECONDS)))
result
case _ => other
}
Expand Down
Loading