Function s3_base

dask/bytes/tests/test_s3.py:82–116

()

@pytest.fixture(scope="module")
def s3_base():
    with ensure_safe_environment_variables():
        os.environ["AWS_ACCESS_KEY_ID"] = "foobar_key"
        os.environ["AWS_SECRET_ACCESS_KEY"] = "foobar_secret"
        # Ignore any local AWS credentials/config files as they can interfere with moto
        os.environ["AWS_SHARED_CREDENTIALS_FILE"] = ""
        os.environ["AWS_CONFIG_FILE"] = ""

        # pipe to null to avoid logging in terminal
        proc = subprocess.Popen(
            shlex.split("moto_server s3 -p 5555"), stdout=subprocess.DEVNULL
        )

        timeout = time.perf_counter() + 8
        while True:
            try:
                # OK to go once server is accepting connections
                r = urllib.request.urlopen(endpoint_uri)
                if r.status == 200:
                    break
            except urllib.request.URLError:
                pass
            time.sleep(0.1)
            assert time.perf_counter() < timeout, "Timed out waiting for moto server"
        yield

        # shut down external process
        proc.terminate()
        try:
            proc.wait(timeout=3)
        except subprocess.TimeoutExpired:
            proc.kill()
            if sys.platform == "win32":
                # belt & braces
                subprocess.call(f"TASKKILL /F /PID {proc.pid} /T")


@pytest.fixture