From 8a7e1c3cfb16ce471c3329ec84ea11e95915eb05 Mon Sep 17 00:00:00 2001 From: gkc Date: Mon, 19 Oct 2026 14:43:26 +0100 Subject: [PATCH 1/5] test: TestServer waits for its worker to leave accept on close closeServerSocket() closed the listening socket while the worker thread could still be blocked in accept() on it. The JDK keeps that socket listening until the blocked thread wakes, so a test that reopened the same port straight away could fail with "Address already in use", as NettyAtCommandExecutorTest.testOnReadyBeforeConnectionIsAccepted did in CI. The worker now marks when it is inside accept(), and closeServerSocket() spins, for at most a second, until it has left. TestServerTest checks that the worker has left accept() once close returns and that the port rebinds at once. Repeated 50 times on Linux with JDK 25, testOnReadyBeforeConnectionIsAccepted failed 20 times without this change and not at all with it. --- .../atsign/client/impl/netty/TestServer.java | 31 +++++++++++++++++-- .../client/impl/netty/TestServerTest.java | 27 ++++++++++++++++ 2 files changed, 55 insertions(+), 3 deletions(-) create mode 100644 at_client/src/test/java/org/atsign/client/impl/netty/TestServerTest.java diff --git a/at_client/src/test/java/org/atsign/client/impl/netty/TestServer.java b/at_client/src/test/java/org/atsign/client/impl/netty/TestServer.java index cf94bd36..36284053 100644 --- a/at_client/src/test/java/org/atsign/client/impl/netty/TestServer.java +++ b/at_client/src/test/java/org/atsign/client/impl/netty/TestServer.java @@ -1,6 +1,7 @@ package org.atsign.client.impl.netty; import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static java.util.concurrent.TimeUnit.SECONDS; import static org.atsign.client.impl.common.Preconditions.checkNotNull; import java.io.*; @@ -56,6 +57,8 @@ public class TestServer implements AutoCloseable { private AtomicBoolean isAccept = new AtomicBoolean(true); + private final AtomicBoolean isAccepting = new AtomicBoolean(); + private AtomicBoolean shutdown = new AtomicBoolean(); private final Thread workerThread; @@ -72,9 +75,15 @@ public TestServer() throws Exception { workerThread = new Thread(() -> { while (!shutdown.get()) { try { - if (serverSocket != null && socket == null && isAccept.get()) { - log.debug("accepting on {}", serverSocket.getLocalPort()); - socket = serverSocket.accept(); + ServerSocket listening = serverSocket; + if (listening != null && socket == null && isAccept.get()) { + log.debug("accepting on {}", listening.getLocalPort()); + isAccepting.set(true); + try { + socket = listening.accept(); + } finally { + isAccepting.set(false); + } writer = new PrintWriter(new BufferedWriter(new OutputStreamWriter(socket.getOutputStream())), false); reader = new BufferedReader(new InputStreamReader(socket.getInputStream())); requestHandler.accept(null); @@ -145,6 +154,10 @@ public void closeClientSocket() { } } + /** + * Closes the listening socket and any accepted one, returning once the worker thread has left + * {@code accept()}, so that the port is free to be bound again. + */ public void closeServerSocket() { try { closeNoThrow(serverSocket, socket, reader, writer); @@ -152,6 +165,18 @@ public void closeServerSocket() { serverSocket = null; socket = null; } + // NOTE the JDK keeps a socket listening until the thread blocked in its accept() wakes + long deadline = System.nanoTime() + SECONDS.toNanos(1); + while (isAccepting.get()) { + if (System.nanoTime() - deadline > 0) { + throw new IllegalStateException("the worker thread has not left accept()"); + } + Thread.onSpinWait(); + } + } + + public boolean isAccepting() { + return isAccepting.get(); } public void accept() { diff --git a/at_client/src/test/java/org/atsign/client/impl/netty/TestServerTest.java b/at_client/src/test/java/org/atsign/client/impl/netty/TestServerTest.java new file mode 100644 index 00000000..32ac66bb --- /dev/null +++ b/at_client/src/test/java/org/atsign/client/impl/netty/TestServerTest.java @@ -0,0 +1,27 @@ +package org.atsign.client.impl.netty; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.awaitility.Awaitility.await; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.is; + +import org.junit.jupiter.api.Test; + +class TestServerTest { + + @Test + void testCloseServerSocketReturnsOnceTheWorkerHasLeftAcceptSoThePortRebindsAtOnce() throws Exception { + try (TestServer server = new TestServer()) { + int port = server.getPort(); + await().pollInterval(1, MILLISECONDS).atMost(1, SECONDS).until(server::isAccepting); + + server.closeServerSocket(); + + assertThat("worker is still inside accept()", server.isAccepting(), is(false)); + server.newServerSocket(); + assertThat(server.getPort(), equalTo(port)); + } + } +} From 20a52ba571af55673ecedf689abb3ae96bcc711e Mon Sep 17 00:00:00 2001 From: gkc Date: Mon, 19 Oct 2026 14:58:44 +0100 Subject: [PATCH 2/5] test: VirtualEnv waits until every hosted atSign is provisioned VirtualEnv.setUp() waited at most 20 seconds for 40 provisioning log lines and carried on whether or not they had arrived. In each of the 16 recent CI runs checked, it waited the full 20 seconds, so the integration tests could start before install_PKAM_Keys had finished, and an atSign provisioned late failed its first test with "no legacy PKAM credential". setUp() now reads the atSigns the container hosts from its supervisor configuration, and returns only once install_PKAM_Keys has exited, which the compose command now reports on stderr, having provisioned every one of them. It throws, naming the atSigns not provisioned, when the script exits short or has not exited after 180 seconds, and when it finds no hosted atSigns. A failed start stops the container, and later calls rethrow that failure rather than starting another. Every line the container writes is logged, prefixed [virtualenv], so a CI log shows how provisioning went. The compose command now escapes its shell variables as $$, so compose no longer blanks $PID in "wait $PID" and warns that PID is unset. The README describes the wait and the line that marks a manually started virtual env as ready, and its docker compose commands no longer pass -f without a file. --- README.md | 17 +- .../org/atsign/virtualenv/VirtualEnv.java | 237 ++++++++++++++++-- .../org/atsign/virtualenv/VirtualEnvTest.java | 183 ++++++++++++++ .../org/atsign/virtualenv/docker-compose.yml | 7 +- 4 files changed, 408 insertions(+), 36 deletions(-) create mode 100644 at_client/src/test/java/org/atsign/virtualenv/VirtualEnvTest.java diff --git a/README.md b/README.md index 7d1c9b2d..8032aee2 100644 --- a/README.md +++ b/README.md @@ -73,6 +73,9 @@ Use **-DskipITs** to avoid running integration tests. The integration tests check to see if a virtual env is running. If not they will attempt to start one. This requires dockerd or desktop docker to be running. +They then wait until `install_PKAM_Keys` has exited having provisioned every +atSign the container hosts. If it misses any, or has not exited within three +minutes, the integration tests fail, naming the atSigns it did not provision. For CI, "standing up" and then "tearing down" the docker container is the intended behavior. However, for a developer this will be slow so it's preferable to "standup" @@ -80,7 +83,7 @@ the virtual env independently, like this: ```shell cd at_client/src/test/resources/org/atsign/virtualenv -docker compose -f up +docker compose up ``` The start-up can take a few minutes and involves running scripts that install the @@ -89,11 +92,13 @@ following log output: ```text ... -virtualenv-1 | SHOUT|...| install_PKAM_Keys |cramAndPkamAuth successful for @chris -virtualenv-1 | SHOUT|...| install_PKAM_Keys |cramAndPkamAuth successful for @policy1 -virtualenv-1 | SHOUT|...| install_PKAM_Keys |cramAndPkamAuth successful for @emoji +virtualenv-1 | install_PKAM_Keys exited with status 0 ``` +Each atSign the script provisions logs a `cramAndPkamAuth successful` or +`cramAuthOnly successful` line before that; a status other than 0, or an atSign +with no such line, means the environment is not fully provisioned. + The tests require test keys (CRAM keys and atKeys) which the virtual env was built with. The pom contains a plugin which downloads the [at_demo_data](https://pub.dev/packages/at_demo_data) package and unpacks it @@ -110,8 +115,8 @@ be necessary to reset the docker container, like this: ```shell cd at_client/src/test/resources/org/atsign/virtualenv -docker compose -f down -docker compose -f up +docker compose down +docker compose up ``` ## Virtual Environment diff --git a/at_client/src/test/java/org/atsign/virtualenv/VirtualEnv.java b/at_client/src/test/java/org/atsign/virtualenv/VirtualEnv.java index e69a4696..fd20f343 100644 --- a/at_client/src/test/java/org/atsign/virtualenv/VirtualEnv.java +++ b/at_client/src/test/java/org/atsign/virtualenv/VirtualEnv.java @@ -1,25 +1,41 @@ package org.atsign.virtualenv; import static java.util.concurrent.TimeUnit.HOURS; -import static java.util.concurrent.TimeUnit.SECONDS; +import static java.util.concurrent.TimeUnit.MILLISECONDS; import java.io.File; -import java.net.URL; +import java.time.Duration; +import java.util.List; +import java.util.Locale; +import java.util.Set; +import java.util.TreeSet; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; import java.util.function.Consumer; +import java.util.function.Function; import java.util.regex.Matcher; import java.util.regex.Pattern; +import java.util.stream.Collectors; +import org.slf4j.Logger; import org.testcontainers.containers.ComposeContainer; +import org.testcontainers.containers.Container.ExecResult; +import org.testcontainers.containers.ContainerState; import org.testcontainers.containers.output.OutputFrame; +import org.testcontainers.containers.output.Slf4jLogConsumer; import lombok.extern.slf4j.Slf4j; @Slf4j public class VirtualEnv { - private static ComposeContainer CONTAINER; + private static final String SERVICE = "virtualenv"; + + private static final Duration PROVISIONING_LIMIT = Duration.ofSeconds(180); + + private static final Pattern HOSTED = Pattern.compile("/secondary\\s+-a\\s+(@\\S+)"); + + private static final Lifecycle LIFECYCLE = new Lifecycle(VirtualEnv::startCompose, PROVISIONING_LIMIT); public static void main(String[] args) throws Exception { setUp(); @@ -29,44 +45,211 @@ public static void main(String[] args) throws Exception { tearDown(); } + /** + * Starts the virtualenv, returning once install_PKAM_Keys has exited having provisioned every + * atSign the container hosts. A failure stops the container, and every later call rethrows it + * without starting another. + */ public static void setUp() { + LIFECYCLE.setUp(); + } + + public static void tearDown() { + LIFECYCLE.tearDown(); + } + + /** + * Returns the consumer that writes each line of the container's output to the given logger, + * prefixed {@code [virtualenv]}. + */ + static Consumer containerLog(Logger logger) { + return new Slf4jLogConsumer(logger).withPrefix(SERVICE); + } + + /** + * Returns the atSigns the container runs an atServer for, folded, as named in its supervisor + * configuration. + */ + static Set hostedAtSigns(String supervisorConfig) { + Matcher matcher = HOSTED.matcher(supervisorConfig); + Set hosted = new TreeSet<>(); + while (matcher.find()) { + hosted.add(fold(matcher.group(1))); + } + return hosted; + } + + private static String fold(String atSign) { + return atSign.toLowerCase(Locale.ROOT); + } + + private static Started startCompose(Consumer provisioningWatch) { + ComposeContainer container; try { - URL resource = VirtualEnv.class.getResource("docker-compose.yml"); - StartUpLatch latch = new StartUpLatch(Pattern.compile("install_PKAM_Keys .*successful"), 40); - CONTAINER = new ComposeContainer(new File(resource.toURI())).withLogConsumer("virtualenv", latch); - CONTAINER.start(); - latch.await(20, SECONDS); + container = new ComposeContainer(new File(VirtualEnv.class.getResource("docker-compose.yml").toURI())) + .withLogConsumer(SERVICE, containerLog(log)) + .withLogConsumer(SERVICE, provisioningWatch); } catch (Exception e) { - CONTAINER = null; - throw new RuntimeException(e); + throw new IllegalStateException("could not load the virtualenv compose file", e); } + try { + container.start(); + } catch (RuntimeException e) { + container.stop(); + throw e; + } + return new Started() { + @Override + public String supervisorConfig() throws Exception { + ContainerState state = container.getContainerByServiceName(SERVICE) + .orElseThrow(() -> new IllegalStateException("no " + SERVICE + " service in the compose project")); + ExecResult result = state.execInContainer("sh", "-c", "cat /etc/supervisor/conf.d/*.conf"); + if (result.getExitCode() != 0) { + throw new IllegalStateException("reading the supervisor configuration exited " + result.getExitCode()); + } + return result.getStdout(); + } + + @Override + public void stop() { + container.stop(); + } + }; } - public static void tearDown() { - CONTAINER.stop(); + /** + * A started virtualenv container, as {@link Lifecycle} needs it. + */ + interface Started { + + /** + * Returns the text of the container's supervisor configuration. + */ + String supervisorConfig() throws Exception; + + void stop(); } - private static class StartUpLatch implements Consumer { + /** + * Starts a virtualenv at most once per run and waits for it to be provisioned. + */ + static final class Lifecycle { - private final Matcher matcher; - private final CountDownLatch latch; + private final Function, Started> starter; + private final Duration limit; + private Started started; + private RuntimeException failure; - public StartUpLatch(Pattern logPattern, int expectedMatches) { - this.matcher = logPattern.matcher(""); - this.latch = new CountDownLatch(expectedMatches); - }; + Lifecycle(Function, Started> starter, Duration limit) { + this.starter = starter; + this.limit = limit; + } + + /** + * Starts the virtualenv unless this lifecycle already has, returning once it is provisioned; + * throws, and stops the container, when it is not. + */ + synchronized void setUp() { + if (failure != null) { + throw new IllegalStateException("the virtualenv failed to start earlier in this run: " + + failure.getMessage(), failure); + } + if (started != null) { + return; + } + ProvisioningWatch watch = new ProvisioningWatch(); + Started candidate = null; + try { + candidate = starter.apply(watch); + watch.awaitProvisioned(readHostedAtSigns(candidate), limit); + started = candidate; + } catch (Exception e) { + if (e instanceof InterruptedException) { + Thread.currentThread().interrupt(); + } + failure = e instanceof RuntimeException ? (RuntimeException) e : new IllegalStateException(e); + if (candidate != null) { + try { + candidate.stop(); + } catch (RuntimeException stopFailure) { + failure.addSuppressed(stopFailure); + } + } + throw failure; + } + } + + synchronized void tearDown() { + if (started != null) { + started.stop(); + started = null; + } + } - public void await(long timeout, TimeUnit unit) throws InterruptedException { - log.info("awaiting container log to contain {} lines that match {}...", latch.getCount(), matcher.pattern()); - latch.await(timeout, unit); - log.info("container log indicates that start up is complete"); + private static Set readHostedAtSigns(Started candidate) { + try { + return hostedAtSigns(candidate.supervisorConfig()); + } catch (Exception e) { + throw new IllegalStateException("found no hosted atSigns: the virtualenv's supervisor configuration" + + " could not be read", e); + } } + } + + /** + * Records, from the container's output, each atSign install_PKAM_Keys reports provisioned and + * the status it exits with. + */ + static final class ProvisioningWatch implements Consumer { + + private static final Pattern PROVISIONED = Pattern.compile("install_PKAM_Keys .*successful for (@\\S+)"); + private static final Pattern EXITED = Pattern.compile("install_PKAM_Keys exited with status (\\d+)"); + + private final Set provisioned = ConcurrentHashMap.newKeySet(); + private final CountDownLatch exited = new CountDownLatch(1); + private volatile int exitStatus; @Override - public void accept(OutputFrame outputFrame) { - if (matcher.reset(outputFrame.getUtf8StringWithoutLineEnding()).find()) { - latch.countDown(); + public void accept(OutputFrame frame) { + String line = frame.getUtf8StringWithoutLineEnding(); + Matcher matcher = PROVISIONED.matcher(line); + if (matcher.find()) { + provisioned.add(fold(matcher.group(1))); + return; + } + matcher = EXITED.matcher(line); + if (matcher.find()) { + exitStatus = Integer.parseInt(matcher.group(1)); + exited.countDown(); + } + } + + /** + * Returns once install_PKAM_Keys has exited with status 0 having provisioned every hosted + * atSign; throws naming the hosted atSigns it has not provisioned otherwise, or once the limit + * passes. + */ + void awaitProvisioned(Set hosted, Duration limit) throws InterruptedException { + if (hosted.isEmpty()) { + throw new IllegalStateException("found no hosted atSigns in the virtualenv's supervisor configuration"); + } + log.info("awaiting install_PKAM_Keys provisioning {} hosted atSigns...", hosted.size()); + boolean hasExited = exited.await(limit.toMillis(), MILLISECONDS); + List missing = hosted.stream() + .filter(atSign -> !provisioned.contains(atSign)) + .collect(Collectors.toList()); + if (!hasExited) { + throw new IllegalStateException(String.format(Locale.ROOT, + "install_PKAM_Keys had not exited after %.1fs; %d of %d hosted atSigns not yet provisioned: %s", + limit.toMillis() / 1000.0, missing.size(), hosted.size(), + missing)); + } + if (exitStatus != 0 || !missing.isEmpty()) { + throw new IllegalStateException(String.format( + "install_PKAM_Keys exited with status %d; %d of %d hosted atSigns not provisioned: %s", + exitStatus, missing.size(), hosted.size(), missing)); } + log.info("install_PKAM_Keys provisioned all {} hosted atSigns", hosted.size()); } } } diff --git a/at_client/src/test/java/org/atsign/virtualenv/VirtualEnvTest.java b/at_client/src/test/java/org/atsign/virtualenv/VirtualEnvTest.java new file mode 100644 index 00000000..8b8335a1 --- /dev/null +++ b/at_client/src/test/java/org/atsign/virtualenv/VirtualEnvTest.java @@ -0,0 +1,183 @@ +package org.atsign.virtualenv; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.contains; +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.is; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockingDetails; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Consumer; +import java.util.function.Function; +import java.util.stream.Collectors; + +import org.junit.jupiter.api.Test; +import org.slf4j.Logger; +import org.slf4j.helpers.MessageFormatter; +import org.testcontainers.containers.output.OutputFrame; +import org.testcontainers.containers.output.OutputFrame.OutputType; + +class VirtualEnvTest { + + private static final String CONFIG = "[program:25026_colin]\n" + + "command = /atsign/secondary/secondary -a @colin -p 25026 -s colinsecret\n" + + "[program:25029_gary]\n" + + "command = /atsign/secondary/secondary -a @gary -p 25029 -s garysecret\n" + + "[program:pkamLoad]\n" + + "command = /bin/sh -c /tmp/pkam.sh\n"; + + private static final Duration LIMIT = Duration.ofSeconds(5); + + @Test + void testSetUpReturnsOnlyAfterInstallPkamKeysExitsHavingProvisionedEveryHostedAtSign() { + AtomicBoolean exitReported = new AtomicBoolean(); + FakeStarter starter = new FakeStarter(CONFIG, watch -> { + watch.accept(stderr("SHOUT|2026-10-19 06:10:26.1| install_PKAM_Keys |cramAndPkamAuth successful for @colin")); + watch.accept(stderr("SHOUT|2026-10-19 06:10:26.2| install_PKAM_Keys |cramAndPkamAuth successful for @gary")); + new Thread(() -> { + sleep(200); + exitReported.set(true); + watch.accept(stderr("install_PKAM_Keys exited with status 0")); + }).start(); + }); + + new VirtualEnv.Lifecycle(starter, LIMIT).setUp(); + + assertThat("setUp returned before install_PKAM_Keys exited", exitReported.get(), is(true)); + assertThat(starter.stops.get(), equalTo(0)); + } + + @Test + void testSetUpThrowsNamingTheHostedAtSignInstallPkamKeysExitedWithoutProvisioning() { + FakeStarter starter = new FakeStarter(CONFIG, watch -> { + watch.accept(stderr("SHOUT|2026-10-19 06:10:26.1| install_PKAM_Keys |cramAndPkamAuth successful for @colin")); + watch.accept(stderr("install_PKAM_Keys exited with status 0")); + }); + + IllegalStateException ex = assertThrows(IllegalStateException.class, + () -> new VirtualEnv.Lifecycle(starter, LIMIT).setUp()); + + assertThat(ex.getMessage(), containsString("1 of 2 hosted atSigns not provisioned: [@gary]")); + } + + @Test + void testSetUpThrowsOnceTheLimitPassesNamingTheHostedAtSignsNotYetProvisioned() { + FakeStarter starter = new FakeStarter(CONFIG, watch -> watch + .accept(stderr("SHOUT|2026-10-19 06:10:26.1| install_PKAM_Keys |cramAuthOnly successful for @colin"))); + + IllegalStateException ex = assertThrows(IllegalStateException.class, + () -> new VirtualEnv.Lifecycle(starter, Duration.ofMillis(100)) + .setUp()); + + assertThat(ex.getMessage(), containsString("had not exited after 0.1s")); + assertThat(ex.getMessage(), containsString("1 of 2 hosted atSigns not yet provisioned: [@gary]")); + } + + @Test + void testContainerOutputIsLoggedPrefixedVirtualenv() { + Logger logger = mock(Logger.class); + + VirtualEnv.containerLog(logger).accept(stderr("install_PKAM_Keys exited with status 0")); + + List logged = mockingDetails(logger).getInvocations().stream() + .filter(invocation -> invocation.getMethod().getName().equals("info")) + .map(invocation -> render(invocation.getArguments())) + .collect(Collectors.toList()); + assertThat(logged, contains("[virtualenv] STDERR: install_PKAM_Keys exited with status 0")); + } + + @Test + void testSetUpRethrowsTheFirstFailureWithoutStartingAnotherContainer() { + FakeStarter starter = + new FakeStarter(CONFIG, watch -> watch.accept(stderr("install_PKAM_Keys exited with status 0"))); + VirtualEnv.Lifecycle lifecycle = new VirtualEnv.Lifecycle(starter, LIMIT); + IllegalStateException first = assertThrows(IllegalStateException.class, lifecycle::setUp); + + IllegalStateException later = assertThrows(IllegalStateException.class, lifecycle::setUp); + + assertThat(later.getMessage(), containsString(first.getMessage())); + assertThat(starter.starts.get(), equalTo(1)); + assertThat(starter.stops.get(), equalTo(1)); + } + + @Test + void testSetUpThrowsWhenTheSupervisorConfigurationNamesNoHostedAtSigns() { + FakeStarter starter = new FakeStarter("[program:pkamLoad]\ncommand = /bin/sh -c /tmp/pkam.sh\n", + watch -> watch.accept(stderr("install_PKAM_Keys exited with status 0"))); + + IllegalStateException ex = assertThrows(IllegalStateException.class, + () -> new VirtualEnv.Lifecycle(starter, LIMIT).setUp()); + + assertThat(ex.getMessage(), containsString("found no hosted atSigns")); + } + + @Test + void testSetUpThrowsWhenTheSupervisorConfigurationCannotBeRead() { + FakeStarter starter = + new FakeStarter(null, watch -> watch.accept(stderr("install_PKAM_Keys exited with status 0"))); + + IllegalStateException ex = assertThrows(IllegalStateException.class, + () -> new VirtualEnv.Lifecycle(starter, LIMIT).setUp()); + + assertThat(ex.getMessage(), containsString("found no hosted atSigns")); + } + + private static OutputFrame stderr(String line) { + return new OutputFrame(OutputType.STDERR, (line + "\n").getBytes(StandardCharsets.UTF_8)); + } + + private static String render(Object[] arguments) { + return MessageFormatter.arrayFormat((String) arguments[0], Arrays.copyOfRange(arguments, 1, arguments.length)) + .getMessage(); + } + + private static void sleep(long millis) { + try { + Thread.sleep(millis); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + private static final class FakeStarter implements Function, VirtualEnv.Started> { + + private final String supervisorConfig; + private final Consumer> output; + private final AtomicInteger starts = new AtomicInteger(); + private final AtomicInteger stops = new AtomicInteger(); + + FakeStarter(String supervisorConfig, Consumer> output) { + this.supervisorConfig = supervisorConfig; + this.output = output; + } + + @Override + public VirtualEnv.Started apply(Consumer watch) { + starts.incrementAndGet(); + output.accept(watch); + return new VirtualEnv.Started() { + @Override + public String supervisorConfig() throws Exception { + if (supervisorConfig == null) { + throw new IOException("cat: /etc/supervisor/conf.d/*.conf: No such file or directory"); + } + return supervisorConfig; + } + + @Override + public void stop() { + stops.incrementAndGet(); + } + }; + } + } +} diff --git a/at_client/src/test/resources/org/atsign/virtualenv/docker-compose.yml b/at_client/src/test/resources/org/atsign/virtualenv/docker-compose.yml index 51a2aa14..c675cc39 100644 --- a/at_client/src/test/resources/org/atsign/virtualenv/docker-compose.yml +++ b/at_client/src/test/resources/org/atsign/virtualenv/docker-compose.yml @@ -13,8 +13,9 @@ services: DART_VM_OPTIONS: "--use_compactor,--force_evacuation,--idle_duration_micros=0,--compactor_tasks=5,--marker_tasks=5,--scavenger_tasks=5,--old_gen_growth_time_ratio=50,--old_gen_growth_space_ratio=10" command: > sh -c ' - supervisord -n & PID=$! && + supervisord -n & PID=$$! && sleep 5 && - /usr/local/bin/install_PKAM_Keys - wait $PID + /usr/local/bin/install_PKAM_Keys; + echo "install_PKAM_Keys exited with status $$?" >&2; + wait $$PID ' From 10633187bb5d59f0070ddd869e6a789006527729 Mon Sep 17 00:00:00 2001 From: gkc Date: Mon, 19 Oct 2026 16:30:27 +0100 Subject: [PATCH 3/5] test: rename TestServer's in-accept flag and give it a 5s bound isAccepting sat beside the existing isAccept, which says whether the worker should accept at all, so the flag that says the worker is inside accept() is now isWorkerInAccept. closeServerSocket() waits up to five seconds for the worker to leave accept() rather than one, so a slow runner does not turn a thread descheduled for a moment into a failure. --- .../org/atsign/client/impl/netty/TestServer.java | 14 +++++++------- .../atsign/client/impl/netty/TestServerTest.java | 4 ++-- 2 files changed, 9 insertions(+), 9 deletions(-) diff --git a/at_client/src/test/java/org/atsign/client/impl/netty/TestServer.java b/at_client/src/test/java/org/atsign/client/impl/netty/TestServer.java index 36284053..f5577484 100644 --- a/at_client/src/test/java/org/atsign/client/impl/netty/TestServer.java +++ b/at_client/src/test/java/org/atsign/client/impl/netty/TestServer.java @@ -57,7 +57,7 @@ public class TestServer implements AutoCloseable { private AtomicBoolean isAccept = new AtomicBoolean(true); - private final AtomicBoolean isAccepting = new AtomicBoolean(); + private final AtomicBoolean isWorkerInAccept = new AtomicBoolean(); private AtomicBoolean shutdown = new AtomicBoolean(); @@ -78,11 +78,11 @@ public TestServer() throws Exception { ServerSocket listening = serverSocket; if (listening != null && socket == null && isAccept.get()) { log.debug("accepting on {}", listening.getLocalPort()); - isAccepting.set(true); + isWorkerInAccept.set(true); try { socket = listening.accept(); } finally { - isAccepting.set(false); + isWorkerInAccept.set(false); } writer = new PrintWriter(new BufferedWriter(new OutputStreamWriter(socket.getOutputStream())), false); reader = new BufferedReader(new InputStreamReader(socket.getInputStream())); @@ -166,8 +166,8 @@ public void closeServerSocket() { socket = null; } // NOTE the JDK keeps a socket listening until the thread blocked in its accept() wakes - long deadline = System.nanoTime() + SECONDS.toNanos(1); - while (isAccepting.get()) { + long deadline = System.nanoTime() + SECONDS.toNanos(5); + while (isWorkerInAccept.get()) { if (System.nanoTime() - deadline > 0) { throw new IllegalStateException("the worker thread has not left accept()"); } @@ -175,8 +175,8 @@ public void closeServerSocket() { } } - public boolean isAccepting() { - return isAccepting.get(); + public boolean isWorkerInAccept() { + return isWorkerInAccept.get(); } public void accept() { diff --git a/at_client/src/test/java/org/atsign/client/impl/netty/TestServerTest.java b/at_client/src/test/java/org/atsign/client/impl/netty/TestServerTest.java index 32ac66bb..b090d39c 100644 --- a/at_client/src/test/java/org/atsign/client/impl/netty/TestServerTest.java +++ b/at_client/src/test/java/org/atsign/client/impl/netty/TestServerTest.java @@ -15,11 +15,11 @@ class TestServerTest { void testCloseServerSocketReturnsOnceTheWorkerHasLeftAcceptSoThePortRebindsAtOnce() throws Exception { try (TestServer server = new TestServer()) { int port = server.getPort(); - await().pollInterval(1, MILLISECONDS).atMost(1, SECONDS).until(server::isAccepting); + await().pollInterval(1, MILLISECONDS).atMost(1, SECONDS).until(server::isWorkerInAccept); server.closeServerSocket(); - assertThat("worker is still inside accept()", server.isAccepting(), is(false)); + assertThat("worker is still inside accept()", server.isWorkerInAccept(), is(false)); server.newServerSocket(); assertThat(server.getPort(), equalTo(port)); } From 8e180923fbb9b6e0b67036b50cedbf314f810dd0 Mon Sep 17 00:00:00 2001 From: gkc Date: Mon, 19 Oct 2026 16:30:27 +0100 Subject: [PATCH 4/5] test: VirtualEnv probes for a running env and fails fast on a dead one Every caller now calls VirtualEnv.setUp() unconditionally, and setUp() checks whether something already answers on vip.ve.atsign.zone:64 only when it has neither started a container nor failed. A recorded failure is therefore rethrown even when the failed container could not be stopped and still answers. setUp() fails at once when the container's output ends before install_PKAM_Keys has exited, rather than after 180 seconds. It counts an atSign as provisioned only from a cramAndPkamAuth or cramAuthOnly success line, not from the INFO line install_PKAM_Keys logs after the CRAM login alone. A stop that fails after a failed start no longer replaces the start's failure. New tests cover a non-zero exit status, a second setUp() after a successful start, atSigns compared folded, the interrupt flag, and the compose command's exit line in the form the watch parses. The README says a virtual env that is already running is used as it is. --- README.md | 2 + .../org/atsign/client/impl/AtClientsIT.java | 6 +- .../client/impl/AtCommandExecutorsIT.java | 6 +- .../atsign/client/impl/cli/ActivateIT.java | 5 +- .../org/atsign/client/impl/cli/CliIT.java | 6 +- .../impl/netty/NettyAtCommandExecutorIT.java | 6 +- .../cucumber/CucumberVirtualEnvPlugin.java | 7 +- .../org/atsign/virtualenv/VirtualEnv.java | 73 ++++++--- .../org/atsign/virtualenv/VirtualEnvTest.java | 150 +++++++++++++++++- 9 files changed, 206 insertions(+), 55 deletions(-) diff --git a/README.md b/README.md index 8032aee2..4043c29e 100644 --- a/README.md +++ b/README.md @@ -76,6 +76,8 @@ attempt to start one. This requires dockerd or desktop docker to be running. They then wait until `install_PKAM_Keys` has exited having provisioned every atSign the container hosts. If it misses any, or has not exited within three minutes, the integration tests fail, naming the atSigns it did not provision. +A virtual env that is already running (anything answering on +vip.ve.atsign.zone:64) is used as it is, with no check that it is provisioned. For CI, "standing up" and then "tearing down" the docker container is the intended behavior. However, for a developer this will be slow so it's preferable to "standup" diff --git a/at_client/src/test/java/org/atsign/client/impl/AtClientsIT.java b/at_client/src/test/java/org/atsign/client/impl/AtClientsIT.java index d2ed0d97..e5f72d6e 100644 --- a/at_client/src/test/java/org/atsign/client/impl/AtClientsIT.java +++ b/at_client/src/test/java/org/atsign/client/impl/AtClientsIT.java @@ -1,6 +1,5 @@ package org.atsign.client.impl; -import static java.util.concurrent.TimeUnit.SECONDS; import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.*; @@ -11,7 +10,6 @@ import org.atsign.client.api.AtSign; import org.atsign.client.impl.common.ReconnectStrategy; import org.atsign.client.impl.util.KeysUtils; -import org.atsign.cucumber.helpers.Helpers; import org.atsign.virtualenv.VirtualEnv; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; @@ -27,9 +25,7 @@ class AtClientsIT { @BeforeAll public static void classSetup() throws Exception { - if (!Helpers.isHostPortReachable("vip.ve.atsign.zone:64", SECONDS.toMillis(2))) { - VirtualEnv.setUp(); - } + VirtualEnv.setUp(); atSign = AtSign.createAtSign("colin"); keysFile = new File("target/at_demo_data/lib/assets/atkeys/@colin.atKeys"); keys = KeysUtils.loadKeys(keysFile); diff --git a/at_client/src/test/java/org/atsign/client/impl/AtCommandExecutorsIT.java b/at_client/src/test/java/org/atsign/client/impl/AtCommandExecutorsIT.java index 8a796e97..dbf39e7c 100644 --- a/at_client/src/test/java/org/atsign/client/impl/AtCommandExecutorsIT.java +++ b/at_client/src/test/java/org/atsign/client/impl/AtCommandExecutorsIT.java @@ -1,6 +1,5 @@ package org.atsign.client.impl; -import static java.util.concurrent.TimeUnit.SECONDS; import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.*; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -14,7 +13,6 @@ import org.atsign.client.impl.commands.ScanCommands; import org.atsign.client.impl.common.ReconnectStrategy; import org.atsign.client.impl.util.KeysUtils; -import org.atsign.cucumber.helpers.Helpers; import org.atsign.virtualenv.VirtualEnv; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; @@ -29,9 +27,7 @@ class AtCommandExecutorsIT { @BeforeAll public static void classSetup() throws Exception { - if (!Helpers.isHostPortReachable("vip.ve.atsign.zone:64", SECONDS.toMillis(2))) { - VirtualEnv.setUp(); - } + VirtualEnv.setUp(); atSign = AtSign.createAtSign("colin"); keys = KeysUtils.loadKeys(new File("target/at_demo_data/lib/assets/atkeys/@colin.atKeys")); } diff --git a/at_client/src/test/java/org/atsign/client/impl/cli/ActivateIT.java b/at_client/src/test/java/org/atsign/client/impl/cli/ActivateIT.java index 2f9609fc..1a8bc6e7 100644 --- a/at_client/src/test/java/org/atsign/client/impl/cli/ActivateIT.java +++ b/at_client/src/test/java/org/atsign/client/impl/cli/ActivateIT.java @@ -22,7 +22,6 @@ import org.atsign.client.impl.util.KeysUtils; import org.atsign.cucumber.helpers.AtDemoData; -import org.atsign.cucumber.helpers.Helpers; import org.atsign.virtualenv.VirtualEnv; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeAll; @@ -40,9 +39,7 @@ class ActivateIT { @BeforeAll public static void classSetup() { - if (!Helpers.isHostPortReachable("vip.ve.atsign.zone:64", SECONDS.toMillis(2))) { - VirtualEnv.setUp(); - } + VirtualEnv.setUp(); } @BeforeEach diff --git a/at_client/src/test/java/org/atsign/client/impl/cli/CliIT.java b/at_client/src/test/java/org/atsign/client/impl/cli/CliIT.java index 99b99b89..dea06889 100644 --- a/at_client/src/test/java/org/atsign/client/impl/cli/CliIT.java +++ b/at_client/src/test/java/org/atsign/client/impl/cli/CliIT.java @@ -1,6 +1,5 @@ package org.atsign.client.impl.cli; -import static java.util.concurrent.TimeUnit.SECONDS; import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.*; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -14,7 +13,6 @@ import java.util.stream.Collectors; import org.atsign.client.impl.util.KeysUtils; -import org.atsign.cucumber.helpers.Helpers; import org.atsign.virtualenv.VirtualEnv; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; @@ -31,9 +29,7 @@ class CliIT { @BeforeAll public static void classSetup() { - if (!Helpers.isHostPortReachable(VIRTUAL_ENV_ROOT, SECONDS.toMillis(2))) { - VirtualEnv.setUp(); - } + VirtualEnv.setUp(); AT_SIGN_KEYS_DIR = KeysUtils.expectedKeysFilesLocation; ATSIGN_KEYS_SUFFIX = KeysUtils.keysFileSuffix; KeysUtils.expectedKeysFilesLocation = "target/at_demo_data/lib/assets/atkeys"; diff --git a/at_client/src/test/java/org/atsign/client/impl/netty/NettyAtCommandExecutorIT.java b/at_client/src/test/java/org/atsign/client/impl/netty/NettyAtCommandExecutorIT.java index 30b8e805..0d8c09aa 100644 --- a/at_client/src/test/java/org/atsign/client/impl/netty/NettyAtCommandExecutorIT.java +++ b/at_client/src/test/java/org/atsign/client/impl/netty/NettyAtCommandExecutorIT.java @@ -1,6 +1,5 @@ package org.atsign.client.impl.netty; -import static java.util.concurrent.TimeUnit.SECONDS; import static org.atsign.client.api.AtSign.createAtSign; import static org.awaitility.Awaitility.await; import static org.hamcrest.MatcherAssert.assertThat; @@ -11,7 +10,6 @@ import org.atsign.client.api.AtSign; import org.atsign.client.impl.netty.NettyAtCommandExecutor.NettyAtCommandExecutorBuilder; -import org.atsign.cucumber.helpers.Helpers; import org.atsign.virtualenv.VirtualEnv; import org.hamcrest.Matchers; import org.junit.jupiter.api.BeforeAll; @@ -24,9 +22,7 @@ class NettyAtCommandExecutorIT { @BeforeAll public static void classSetup() { - if (!Helpers.isHostPortReachable("vip.ve.atsign.zone:64", SECONDS.toMillis(2))) { - VirtualEnv.setUp(); - } + VirtualEnv.setUp(); } @Test diff --git a/at_client/src/test/java/org/atsign/cucumber/CucumberVirtualEnvPlugin.java b/at_client/src/test/java/org/atsign/cucumber/CucumberVirtualEnvPlugin.java index 013331be..ec718429 100644 --- a/at_client/src/test/java/org/atsign/cucumber/CucumberVirtualEnvPlugin.java +++ b/at_client/src/test/java/org/atsign/cucumber/CucumberVirtualEnvPlugin.java @@ -1,8 +1,5 @@ package org.atsign.cucumber; -import static java.util.concurrent.TimeUnit.SECONDS; - -import org.atsign.cucumber.helpers.Helpers; import org.atsign.virtualenv.VirtualEnv; import io.cucumber.plugin.EventListener; @@ -13,8 +10,6 @@ public class CucumberVirtualEnvPlugin implements EventListener { @Override public void setEventPublisher(EventPublisher publisher) { - if (!Helpers.isHostPortReachable("vip.ve.atsign.zone:64", SECONDS.toMillis(2))) { - publisher.registerHandlerFor(TestRunStarted.class, e -> VirtualEnv.setUp()); - } + publisher.registerHandlerFor(TestRunStarted.class, e -> VirtualEnv.setUp()); } } diff --git a/at_client/src/test/java/org/atsign/virtualenv/VirtualEnv.java b/at_client/src/test/java/org/atsign/virtualenv/VirtualEnv.java index fd20f343..82240ad4 100644 --- a/at_client/src/test/java/org/atsign/virtualenv/VirtualEnv.java +++ b/at_client/src/test/java/org/atsign/virtualenv/VirtualEnv.java @@ -2,6 +2,7 @@ import static java.util.concurrent.TimeUnit.HOURS; import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static java.util.concurrent.TimeUnit.SECONDS; import java.io.File; import java.time.Duration; @@ -11,12 +12,14 @@ import java.util.TreeSet; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; +import java.util.function.BooleanSupplier; import java.util.function.Consumer; import java.util.function.Function; import java.util.regex.Matcher; import java.util.regex.Pattern; import java.util.stream.Collectors; +import org.atsign.cucumber.helpers.Helpers; import org.slf4j.Logger; import org.testcontainers.containers.ComposeContainer; import org.testcontainers.containers.Container.ExecResult; @@ -31,11 +34,14 @@ public class VirtualEnv { private static final String SERVICE = "virtualenv"; + private static final String ROOT_SERVER = "vip.ve.atsign.zone:64"; + private static final Duration PROVISIONING_LIMIT = Duration.ofSeconds(180); private static final Pattern HOSTED = Pattern.compile("/secondary\\s+-a\\s+(@\\S+)"); - private static final Lifecycle LIFECYCLE = new Lifecycle(VirtualEnv::startCompose, PROVISIONING_LIMIT); + private static final Lifecycle LIFECYCLE = new Lifecycle(VirtualEnv::startCompose, + () -> Helpers.isHostPortReachable(ROOT_SERVER, SECONDS.toMillis(2)), PROVISIONING_LIMIT); public static void main(String[] args) throws Exception { setUp(); @@ -46,14 +52,17 @@ public static void main(String[] args) throws Exception { } /** - * Starts the virtualenv, returning once install_PKAM_Keys has exited having provisioned every - * atSign the container hosts. A failure stops the container, and every later call rethrows it - * without starting another. + * Starts the virtualenv, unless one is already answering on its root server, returning once + * install_PKAM_Keys has exited having provisioned every atSign the container hosts. A failure + * stops the container, and every later call rethrows it without starting another. */ public static void setUp() { LIFECYCLE.setUp(); } + /** + * Stops the virtualenv this class started, if any. + */ public static void tearDown() { LIFECYCLE.tearDown(); } @@ -95,7 +104,11 @@ private static Started startCompose(Consumer provisioningWatch) { try { container.start(); } catch (RuntimeException e) { - container.stop(); + try { + container.stop(); + } catch (RuntimeException stopFailure) { + e.addSuppressed(stopFailure); + } throw e; } return new Started() { @@ -127,27 +140,32 @@ interface Started { */ String supervisorConfig() throws Exception; + /** + * Stops and removes the container. + */ void stop(); } /** - * Starts a virtualenv at most once per run and waits for it to be provisioned. + * Starts a virtualenv, at most once until it is torn down, and waits for it to be provisioned. */ static final class Lifecycle { private final Function, Started> starter; + private final BooleanSupplier alreadyRunning; private final Duration limit; private Started started; private RuntimeException failure; - Lifecycle(Function, Started> starter, Duration limit) { + Lifecycle(Function, Started> starter, BooleanSupplier alreadyRunning, Duration limit) { this.starter = starter; + this.alreadyRunning = alreadyRunning; this.limit = limit; } /** - * Starts the virtualenv unless this lifecycle already has, returning once it is provisioned; - * throws, and stops the container, when it is not. + * Starts the virtualenv unless this lifecycle already has or one is already running, returning + * once it is provisioned; throws, and stops the container, when it is not. */ synchronized void setUp() { if (failure != null) { @@ -157,6 +175,10 @@ synchronized void setUp() { if (started != null) { return; } + if (alreadyRunning.getAsBoolean()) { + log.info("using the virtualenv that is already running"); + return; + } ProvisioningWatch watch = new ProvisioningWatch(); Started candidate = null; try { @@ -197,20 +219,26 @@ private static Set readHostedAtSigns(Started candidate) { } /** - * Records, from the container's output, each atSign install_PKAM_Keys reports provisioned and - * the status it exits with. + * Records, from the container's output, each atSign install_PKAM_Keys reports provisioned, the + * status it exits with, and whether the output has ended. */ static final class ProvisioningWatch implements Consumer { - private static final Pattern PROVISIONED = Pattern.compile("install_PKAM_Keys .*successful for (@\\S+)"); + private static final Pattern PROVISIONED = + Pattern.compile("install_PKAM_Keys \\|(?:cramAndPkamAuth|cramAuthOnly) successful for (@\\S+)$"); private static final Pattern EXITED = Pattern.compile("install_PKAM_Keys exited with status (\\d+)"); private final Set provisioned = ConcurrentHashMap.newKeySet(); - private final CountDownLatch exited = new CountDownLatch(1); + private final CountDownLatch exitedOrEnded = new CountDownLatch(1); + private volatile boolean exitReported; private volatile int exitStatus; @Override public void accept(OutputFrame frame) { + if (frame.getType() == OutputFrame.OutputType.END) { + exitedOrEnded.countDown(); + return; + } String line = frame.getUtf8StringWithoutLineEnding(); Matcher matcher = PROVISIONED.matcher(line); if (matcher.find()) { @@ -220,32 +248,39 @@ public void accept(OutputFrame frame) { matcher = EXITED.matcher(line); if (matcher.find()) { exitStatus = Integer.parseInt(matcher.group(1)); - exited.countDown(); + exitReported = true; + exitedOrEnded.countDown(); } } /** * Returns once install_PKAM_Keys has exited with status 0 having provisioned every hosted - * atSign; throws naming the hosted atSigns it has not provisioned otherwise, or once the limit - * passes. + * atSign; throws naming the hosted atSigns it has not provisioned otherwise, when the + * container's output ends first, or once the limit passes. */ void awaitProvisioned(Set hosted, Duration limit) throws InterruptedException { if (hosted.isEmpty()) { throw new IllegalStateException("found no hosted atSigns in the virtualenv's supervisor configuration"); } log.info("awaiting install_PKAM_Keys provisioning {} hosted atSigns...", hosted.size()); - boolean hasExited = exited.await(limit.toMillis(), MILLISECONDS); + boolean finished = exitedOrEnded.await(limit.toMillis(), MILLISECONDS); List missing = hosted.stream() .filter(atSign -> !provisioned.contains(atSign)) .collect(Collectors.toList()); - if (!hasExited) { + if (!finished) { throw new IllegalStateException(String.format(Locale.ROOT, "install_PKAM_Keys had not exited after %.1fs; %d of %d hosted atSigns not yet provisioned: %s", limit.toMillis() / 1000.0, missing.size(), hosted.size(), missing)); } + if (!exitReported) { + throw new IllegalStateException(String.format(Locale.ROOT, + "the virtualenv container's output ended before install_PKAM_Keys exited; %d of %d hosted atSigns" + + " not yet provisioned: %s", + missing.size(), hosted.size(), missing)); + } if (exitStatus != 0 || !missing.isEmpty()) { - throw new IllegalStateException(String.format( + throw new IllegalStateException(String.format(Locale.ROOT, "install_PKAM_Keys exited with status %d; %d of %d hosted atSigns not provisioned: %s", exitStatus, missing.size(), hosted.size(), missing)); } diff --git a/at_client/src/test/java/org/atsign/virtualenv/VirtualEnvTest.java b/at_client/src/test/java/org/atsign/virtualenv/VirtualEnvTest.java index 8b8335a1..432d7d3b 100644 --- a/at_client/src/test/java/org/atsign/virtualenv/VirtualEnvTest.java +++ b/at_client/src/test/java/org/atsign/virtualenv/VirtualEnvTest.java @@ -10,14 +10,19 @@ import static org.mockito.Mockito.mockingDetails; import java.io.IOException; +import java.io.InputStream; import java.nio.charset.StandardCharsets; import java.time.Duration; import java.util.Arrays; import java.util.List; +import java.util.Set; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BooleanSupplier; import java.util.function.Consumer; import java.util.function.Function; +import java.util.regex.Matcher; +import java.util.regex.Pattern; import java.util.stream.Collectors; import org.junit.jupiter.api.Test; @@ -37,6 +42,8 @@ class VirtualEnvTest { private static final Duration LIMIT = Duration.ofSeconds(5); + private static final BooleanSupplier NOTHING_RUNNING = () -> false; + @Test void testSetUpReturnsOnlyAfterInstallPkamKeysExitsHavingProvisionedEveryHostedAtSign() { AtomicBoolean exitReported = new AtomicBoolean(); @@ -50,7 +57,7 @@ void testSetUpReturnsOnlyAfterInstallPkamKeysExitsHavingProvisionedEveryHostedAt }).start(); }); - new VirtualEnv.Lifecycle(starter, LIMIT).setUp(); + new VirtualEnv.Lifecycle(starter, NOTHING_RUNNING, LIMIT).setUp(); assertThat("setUp returned before install_PKAM_Keys exited", exitReported.get(), is(true)); assertThat(starter.stops.get(), equalTo(0)); @@ -64,7 +71,7 @@ void testSetUpThrowsNamingTheHostedAtSignInstallPkamKeysExitedWithoutProvisionin }); IllegalStateException ex = assertThrows(IllegalStateException.class, - () -> new VirtualEnv.Lifecycle(starter, LIMIT).setUp()); + () -> new VirtualEnv.Lifecycle(starter, NOTHING_RUNNING, LIMIT).setUp()); assertThat(ex.getMessage(), containsString("1 of 2 hosted atSigns not provisioned: [@gary]")); } @@ -75,7 +82,8 @@ void testSetUpThrowsOnceTheLimitPassesNamingTheHostedAtSignsNotYetProvisioned() .accept(stderr("SHOUT|2026-10-19 06:10:26.1| install_PKAM_Keys |cramAuthOnly successful for @colin"))); IllegalStateException ex = assertThrows(IllegalStateException.class, - () -> new VirtualEnv.Lifecycle(starter, Duration.ofMillis(100)) + () -> new VirtualEnv.Lifecycle(starter, NOTHING_RUNNING, + Duration.ofMillis(100)) .setUp()); assertThat(ex.getMessage(), containsString("had not exited after 0.1s")); @@ -99,7 +107,7 @@ void testContainerOutputIsLoggedPrefixedVirtualenv() { void testSetUpRethrowsTheFirstFailureWithoutStartingAnotherContainer() { FakeStarter starter = new FakeStarter(CONFIG, watch -> watch.accept(stderr("install_PKAM_Keys exited with status 0"))); - VirtualEnv.Lifecycle lifecycle = new VirtualEnv.Lifecycle(starter, LIMIT); + VirtualEnv.Lifecycle lifecycle = new VirtualEnv.Lifecycle(starter, NOTHING_RUNNING, LIMIT); IllegalStateException first = assertThrows(IllegalStateException.class, lifecycle::setUp); IllegalStateException later = assertThrows(IllegalStateException.class, lifecycle::setUp); @@ -115,7 +123,7 @@ void testSetUpThrowsWhenTheSupervisorConfigurationNamesNoHostedAtSigns() { watch -> watch.accept(stderr("install_PKAM_Keys exited with status 0"))); IllegalStateException ex = assertThrows(IllegalStateException.class, - () -> new VirtualEnv.Lifecycle(starter, LIMIT).setUp()); + () -> new VirtualEnv.Lifecycle(starter, NOTHING_RUNNING, LIMIT).setUp()); assertThat(ex.getMessage(), containsString("found no hosted atSigns")); } @@ -126,9 +134,139 @@ void testSetUpThrowsWhenTheSupervisorConfigurationCannotBeRead() { new FakeStarter(null, watch -> watch.accept(stderr("install_PKAM_Keys exited with status 0"))); IllegalStateException ex = assertThrows(IllegalStateException.class, - () -> new VirtualEnv.Lifecycle(starter, LIMIT).setUp()); + () -> new VirtualEnv.Lifecycle(starter, NOTHING_RUNNING, LIMIT).setUp()); assertThat(ex.getMessage(), containsString("found no hosted atSigns")); + assertThat(ex.getMessage(), containsString("could not be read")); + } + + @Test + void testSetUpDoesNotCountACramLoginLineAsProvisioned() { + FakeStarter starter = new FakeStarter(CONFIG, watch -> { + watch.accept(stderr("SHOUT|2026-10-19 06:10:26.1| install_PKAM_Keys |cramAndPkamAuth successful for @colin")); + watch + .accept(stderr("INFO|2026-10-19 06:10:26.2| install_PKAM_Keys |CRAM Authentication is successful for @gary")); + watch.accept(stderr("install_PKAM_Keys exited with status 0")); + }); + + IllegalStateException ex = assertThrows(IllegalStateException.class, + () -> new VirtualEnv.Lifecycle(starter, NOTHING_RUNNING, LIMIT).setUp()); + + assertThat(ex.getMessage(), containsString("1 of 2 hosted atSigns not provisioned: [@gary]")); + } + + @Test + void testSetUpThrowsNamingTheStatusWhenInstallPkamKeysExitsNonZero() { + FakeStarter starter = new FakeStarter(CONFIG, watch -> { + watch.accept(stderr("SHOUT|2026-10-19 06:10:26.1| install_PKAM_Keys |cramAndPkamAuth successful for @colin")); + watch.accept(stderr("SHOUT|2026-10-19 06:10:26.2| install_PKAM_Keys |cramAndPkamAuth successful for @gary")); + watch.accept(stderr("install_PKAM_Keys exited with status 1")); + }); + + IllegalStateException ex = assertThrows(IllegalStateException.class, + () -> new VirtualEnv.Lifecycle(starter, NOTHING_RUNNING, LIMIT).setUp()); + + assertThat(ex.getMessage(), containsString("exited with status 1")); + } + + @Test + void testSetUpAfterASuccessfulStartDoesNotStartAnotherContainer() { + FakeStarter starter = new FakeStarter(CONFIG, watch -> { + watch.accept(stderr("SHOUT|2026-10-19 06:10:26.1| install_PKAM_Keys |cramAndPkamAuth successful for @colin")); + watch.accept(stderr("SHOUT|2026-10-19 06:10:26.2| install_PKAM_Keys |cramAuthOnly successful for @gary")); + watch.accept(stderr("install_PKAM_Keys exited with status 0")); + }); + VirtualEnv.Lifecycle lifecycle = new VirtualEnv.Lifecycle(starter, NOTHING_RUNNING, LIMIT); + lifecycle.setUp(); + + lifecycle.setUp(); + + assertThat(starter.starts.get(), equalTo(1)); + assertThat(starter.stops.get(), equalTo(0)); + } + + @Test + void testSetUpComparesAtSignsFolded() { + String config = "command = /atsign/secondary/secondary -a @Colin -p 25026 -s colinsecret\n" + + "command = /atsign/secondary/secondary -a @gary -p 25029 -s garysecret\n"; + FakeStarter starter = new FakeStarter(config, watch -> { + watch.accept(stderr("SHOUT|2026-10-19 06:10:26.1| install_PKAM_Keys |cramAndPkamAuth successful for @colin")); + watch.accept(stderr("SHOUT|2026-10-19 06:10:26.2| install_PKAM_Keys |cramAndPkamAuth successful for @GARY")); + watch.accept(stderr("install_PKAM_Keys exited with status 0")); + }); + + new VirtualEnv.Lifecycle(starter, NOTHING_RUNNING, LIMIT).setUp(); + + assertThat(starter.stops.get(), equalTo(0)); + } + + @Test + void testSetUpRestoresTheInterruptWhenInterruptedWhileWaiting() { + FakeStarter starter = new FakeStarter(CONFIG, watch -> { + }); + Thread.currentThread().interrupt(); + + assertThrows(IllegalStateException.class, () -> new VirtualEnv.Lifecycle(starter, NOTHING_RUNNING, LIMIT).setUp()); + + assertThat("interrupt flag", Thread.interrupted(), is(true)); + } + + @Test + void testTheComposeCommandReportsTheExitOnStderrInTheFormTheWatchParses() throws Exception { + String compose; + try (InputStream in = VirtualEnv.class.getResourceAsStream("docker-compose.yml")) { + compose = new String(in.readAllBytes(), StandardCharsets.UTF_8); + } + Matcher echo = Pattern.compile("echo \"([^\"]*)\" >&2").matcher(compose); + assertThat("compose command echoes a line to stderr", echo.find(), is(true)); + String template = echo.group(1); + assertThat(template, containsString("$$?")); + VirtualEnv.ProvisioningWatch watch = new VirtualEnv.ProvisioningWatch(); + watch.accept(stderr("SHOUT|2026-10-19 06:10:26.1| install_PKAM_Keys |cramAndPkamAuth successful for @colin")); + + watch.accept(stderr(template.replace("$$?", "0"))); + + watch.awaitProvisioned(Set.of("@colin"), LIMIT); + } + + @Test + void testSetUpUsesAVirtualEnvThatIsAlreadyRunningWithoutStartingAContainer() { + FakeStarter starter = new FakeStarter(CONFIG, watch -> { + }); + + new VirtualEnv.Lifecycle(starter, () -> true, LIMIT).setUp(); + + assertThat(starter.starts.get(), equalTo(0)); + } + + @Test + void testSetUpRethrowsTheFirstFailureWhenTheFailedContainerIsStillAnswering() { + AtomicBoolean answering = new AtomicBoolean(); + FakeStarter starter = new FakeStarter(CONFIG, watch -> { + answering.set(true); + watch.accept(stderr("install_PKAM_Keys exited with status 0")); + }); + VirtualEnv.Lifecycle lifecycle = new VirtualEnv.Lifecycle(starter, answering::get, LIMIT); + IllegalStateException first = assertThrows(IllegalStateException.class, lifecycle::setUp); + + IllegalStateException later = assertThrows(IllegalStateException.class, lifecycle::setUp); + + assertThat(later.getMessage(), containsString(first.getMessage())); + assertThat(starter.starts.get(), equalTo(1)); + } + + @Test + void testSetUpThrowsAtOnceWhenTheContainerOutputEndsBeforeInstallPkamKeysExits() { + FakeStarter starter = new FakeStarter(CONFIG, watch -> { + watch.accept(stderr("SHOUT|2026-10-19 06:10:26.1| install_PKAM_Keys |cramAndPkamAuth successful for @colin")); + watch.accept(OutputFrame.END); + }); + + IllegalStateException ex = assertThrows(IllegalStateException.class, + () -> new VirtualEnv.Lifecycle(starter, NOTHING_RUNNING, LIMIT).setUp()); + + assertThat(ex.getMessage(), containsString("output ended before install_PKAM_Keys exited")); + assertThat(ex.getMessage(), containsString("1 of 2 hosted atSigns not yet provisioned: [@gary]")); } private static OutputFrame stderr(String line) { From 3544fd8a5e73ce140b3a0a2a7856333dc2dcaf99 Mon Sep 17 00:00:00 2001 From: gkc Date: Tue, 20 Oct 2026 17:03:49 +0100 Subject: [PATCH 5/5] test: TestServer closes a connection accepted while it closes closeServerSocket read the accepted socket before it waited for the worker to leave accept(). A connection that accept() returned during that wait was then either kept and served, or dropped with nothing referencing it, and stayed open either way. It now closes the listening socket, waits, and then closes whatever the worker accepted. --- .../test/java/org/atsign/client/impl/netty/TestServer.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/at_client/src/test/java/org/atsign/client/impl/netty/TestServer.java b/at_client/src/test/java/org/atsign/client/impl/netty/TestServer.java index f5577484..2ddd2f0e 100644 --- a/at_client/src/test/java/org/atsign/client/impl/netty/TestServer.java +++ b/at_client/src/test/java/org/atsign/client/impl/netty/TestServer.java @@ -160,10 +160,9 @@ public void closeClientSocket() { */ public void closeServerSocket() { try { - closeNoThrow(serverSocket, socket, reader, writer); + closeNoThrow(serverSocket); } finally { serverSocket = null; - socket = null; } // NOTE the JDK keeps a socket listening until the thread blocked in its accept() wakes long deadline = System.nanoTime() + SECONDS.toNanos(5); @@ -173,6 +172,7 @@ public void closeServerSocket() { } Thread.onSpinWait(); } + closeClientSocket(); } public boolean isWorkerInAccept() {