Update s3/localstack_test

Based in part on #111

When provided with --localstack_binary, localstack_test will start
localstack in host mode (via package localstack[runtime]).

When provided with --localstack_endpoint, localstack_test will connect
to a running localstack instance.

In order to integrate it into the CI, we need to install localstack[runtime],
however there is presently a pypa conflict between apache-beam and localstack.

PiperOrigin-RevId: 574269409
Change-Id: I511f6a7c28b9d14d9836030988f6d9bcb68a0549

tensorstore/internal/subprocess.cc

@@ -37,6 +37,7 @@
 #include <string.h>  // IWYU pragma: keep
 
 #include <atomic>  // IWYU pragma: keep
+#include <cassert>
 #include <limits>  // IWYU pragma: keep
 #include <memory>
 #include <string>
@@ -274,13 +275,14 @@ bool retry(int e) {
 }  // namespace
 
 struct Subprocess::Impl {
+  Impl(pid_t pid) : child_pid_(pid), exit_code_(-1) {}
   ~Impl();
 
   absl::Status Kill(int signal);
   Result<int> Join(bool block);
 
-  std::atomic<pid_t> child_pid_{-1};
-  std::atomic<int> exit_code_{-1};
+  std::atomic<pid_t> child_pid_;
+  std::atomic<int> exit_code_;
 };
 
 Subprocess::Impl::~Impl() {
@@ -412,19 +414,22 @@ Result<Subprocess> SpawnSubprocess(const SubprocessOptions& options) {
                              options.executable, " failed");
   }
   ABSL_CHECK_GT(child_pid, 0);
-
-  auto impl = std::make_shared<Subprocess::Impl>();
-  impl->child_pid_ = child_pid;
-  return Subprocess(std::move(impl));
+  return Subprocess(std::make_shared<Subprocess::Impl>(child_pid));
 }
 
 #endif  // !_WIN32
 
 Subprocess::~Subprocess() = default;
 
-absl::Status Subprocess::Kill(int signal) const { return impl_->Kill(signal); }
+absl::Status Subprocess::Kill(int signal) const {
+  assert(impl_);
+  return impl_->Kill(signal);
+}
 
-Result<int> Subprocess::Join(bool block) const { return impl_->Join(block); }
+Result<int> Subprocess::Join(bool block) const {
+  assert(impl_);
+  return impl_->Join(block);
+}
 
 }  // namespace internal
 }  // namespace tensorstore

tensorstore/internal/subprocess_test.cc

@@ -47,7 +47,7 @@ TEST(SubprocessTest, Join) {
   opts.args = {kSubprocessArg};
 
   auto child = SpawnSubprocess(opts);
-  EXPECT_TRUE(child.ok());
+  ASSERT_TRUE(child.ok());
 
   int exit_code;
   TENSORSTORE_ASSERT_OK_AND_ASSIGN(exit_code, child->Join());
@@ -60,7 +60,7 @@ TEST(SubprocessTest, Kill) {
   opts.args = {kSleepArg, kSubprocessArg};
 
   auto child = SpawnSubprocess(opts);
-  EXPECT_TRUE(child.ok());
+  ASSERT_TRUE(child.ok());
 
   EXPECT_THAT(child->Join(/*block=*/false),
               tensorstore::MatchesStatus(absl::StatusCode::kUnavailable));
@@ -81,7 +81,7 @@ TEST(SubprocessTest, DontInherit) {
   opts.inherit_stderr = false;
 
   auto child = SpawnSubprocess(opts);
-  EXPECT_TRUE(child.ok());
+  ASSERT_TRUE(child.ok());
 
   int exit_code;
   TENSORSTORE_ASSERT_OK_AND_ASSIGN(exit_code, child->Join());
@@ -95,7 +95,7 @@ TEST(SubprocessTest, Drop) {
   opts.args = {kSubprocessArg};
 
   auto child = SpawnSubprocess(opts);
-  EXPECT_TRUE(child.ok());
+  ASSERT_TRUE(child.ok());
 
   // Kill changes the result;
   child->Kill().IgnoreError();
@@ -110,7 +110,7 @@ TEST(SubprocessTest, Env) {
       {{"SUBPROCESS_TEST_ENV", "1"}});
 
   auto child = SpawnSubprocess(opts);
-  EXPECT_TRUE(child.ok());
+  ASSERT_TRUE(child.ok());
 
   int exit_code;
   TENSORSTORE_ASSERT_OK_AND_ASSIGN(exit_code, child->Join());

tensorstore/kvstore/s3/BUILD

@@ -296,13 +296,16 @@ tensorstore_cc_test(
 tensorstore_cc_test(
     name = "localstack_test",
     srcs = ["localstack_test.cc"],
+    # args = ["--localstack_binary=$(location :localstack_cli)"],
+    # data = [":localstack_cli"],
     tags = [
         "manual",
         "notap",  # TODO: Enable before linking by default.
         "skip-cmake",  # localstack is python, which will not work in bazel_to_cmake.
     ],
     deps = [
         ":aws_credential_provider",
+        ":s3",
         ":s3_request_builder",
         "//tensorstore:context",
         "//tensorstore:json_serialization_options_base",
@@ -311,6 +314,7 @@ tensorstore_cc_test(
         "//tensorstore/internal:subprocess",
         "//tensorstore/internal/http",
         "//tensorstore/internal/http:curl_transport",
+        "//tensorstore/internal/http:transport_test_utils",
         "//tensorstore/kvstore",
         "//tensorstore/kvstore:test_util",
         "//tensorstore/util:future",
@@ -326,3 +330,18 @@ tensorstore_cc_test(
         "@com_google_googletest//:gtest_main",
     ],
 )
+
+# NOTE: For the localstack_test, it seems simples to add a proxy py_binary like:
+# py_binary(
+#     name = "localstack_cli",
+#     testonly = 1,
+#     srcs = ["localstack_cli.py"],
+#     deps = ["@pypa_localstack//:localstack"],
+# )
+#
+# Where localstack_cli.py is something like this:
+#   if __name__ == "__main__":
+#     localstack.cli.main.main()
+#
+# However localstack[runtime] resolves to a pypa dependency on dill which conflicts with
+# the apache-beam dependency on dill.