code: /*
   * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
   */
  
  package io.airbyte.integrations.standardtest.source;
  
  import static org.junit.jupiter.api.Assertions.assertTrue;
  
  import com.fasterxml.jackson.databind.JsonNode;
  import com.google.common.collect.Lists;
  import com.google.common.collect.Streams;
  import io.airbyte.commons.io.IOs;
  import io.airbyte.commons.io.LineGobbler;
  import io.airbyte.commons.json.Jsons;
  import io.airbyte.config.EnvConfigs;
  import io.airbyte.protocol.models.AirbyteMessage;
  import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
  import io.airbyte.protocol.models.ConnectorSpecification;
  import io.airbyte.workers.WorkerConfigs;
  import io.airbyte.workers.WorkerUtils;
  import java.io.IOException;
  import java.nio.file.Files;
  import java.nio.file.Path;
  import java.util.List;
  import java.util.concurrent.TimeUnit;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  /**
   * Extends TestSource such that it can be called using resources pulled from the file system. Will
   * also add the ability to execute arbitrary scripts in the next version.
   */
  public class PythonSourceAcceptanceTest extends SourceAcceptanceTest {
  
    private static final Logger LOGGER = LoggerFactory.getLogger(PythonSourceAcceptanceTest.class);
    private static final String OUTPUT_FILENAME = "output.json";
  
    public static String IMAGE_NAME;
    public static String PYTHON_CONTAINER_NAME;
  
    private Path testRoot;
  
    @Override
    protected String getImageName() {
      return IMAGE_NAME;
    }
  
    @Override
    protected ConnectorSpecification getSpec() throws IOException {
      return runExecutable(Command.GET_SPEC, ConnectorSpecification.class);
    }
  
    @Override
    protected JsonNode getConfig() throws IOException {
      return runExecutable(Command.GET_CONFIG);
    }
  
    @Override
    protected ConfiguredAirbyteCatalog getConfiguredCatalog() throws IOException {
      return runExecutable(Command.GET_CONFIGURED_CATALOG, ConfiguredAirbyteCatalog.class);
    }
  
    @Override
    protected JsonNode getState() throws IOException {
      return runExecutable(Command.GET_STATE);
    }
  
    @Override
    protected void assertFullRefreshMessages(final List<AirbyteMessage> allMessages) throws IOException {
      final List<String> regexTests = Streams.stream(runExecutable(Command.GET_REGEX_TESTS).withArray("tests").elements())
          .map(JsonNode::textValue).toList();
      final List<String> stringMessages = allMessages.stream().map(Jsons::serialize).toList();
      LOGGER.info("Running " + regexTests.size() + " regex tests...");
      regexTests.forEach(regex -> {
        LOGGER.info("Looking for [" + regex + "]");
        assertTrue(stringMessages.stream().anyMatch(line -> line.matches(regex)), "Failed to find regex: " + regex);
      });
    }
  
    @Override
    protected void setupEnvironment(final TestDestinationEnv environment) throws Exception {
      testRoot = Files.createTempDirectory(Files.createDirectories(Path.of("/tmp/standard_test")), "pytest");
      runExecutableVoid(Command.SETUP);
    }
  
    @Override
    protected void tearDown(final TestDestinationEnv testEnv) throws Exception {
      runExecutableVoid(Command.TEARDOWN);
    }
  
    private enum Command {
      GET_SPEC,
      GET_CONFIG,
      GET_CONFIGURED_CATALOG,
      GET_STATE,
      GET_REGEX_TESTS,
      SETUP,
      TEARDOWN
    }
  
    private <T> T runExecutable(final Command cmd, final Class<T> klass) throws IOException {
      return Jsons.object(runExecutable(cmd), klass);
    }
  
    private JsonNode runExecutable(final Command cmd) throws IOException {
      return Jsons.deserialize(IOs.readFile(runExecutableInternal(cmd), OUTPUT_FILENAME));
    }
  
    private void runExecutableVoid(final Command cmd) throws IOException {
      runExecutableInternal(cmd);
    }
  
    private Path runExecutableInternal(final Command cmd) throws IOException {
      LOGGER.info("testRoot = " + testRoot);
      final List<String> dockerCmd =
          Lists.newArrayList(
              "docker",
              "run",
              "--rm",
              "-i",
              "-v",
              String.format("%s:%s", testRoot, "/test_root"),
              "-w",
              testRoot.toString(),
              "--network",
              "host",
              PYTHON_CONTAINER_NAME,
              cmd.toString().toLowerCase(),
              "--out",
              "/test_root");
  
      final Process process = new ProcessBuilder(dockerCmd).start();
      LineGobbler.gobble(process.getErrorStream(), LOGGER::error);
      LineGobbler.gobble(process.getInputStream(), LOGGER::info);
  
      WorkerUtils.gentleClose(new WorkerConfigs(new EnvConfigs()), process, 1, TimeUnit.MINUTES);
  
      final int exitCode = process.exitValue();
      if (exitCode != 0) {
        throw new RuntimeException("python execution failed");
      }
  
      return testRoot;
    }
  
  }
  