Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
6 changes: 6 additions & 0 deletions .github/workflows/tests.yml
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,12 @@ jobs:
- name: Setup env
run: uv pip install --group dev -e .

- name: Cache CrateDB
uses: actions/cache@v6
with:
path: parts/crate
key: cratedb-${{ runner.os }}-${{ runner.arch }}-${{ hashFiles('tests/conftest.py') }}

- name: Run Linters
run: |
ruff check .
Expand Down
9 changes: 9 additions & 0 deletions CHANGES.rst
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,15 @@
Changes for crate
=================

Unreleased
================
- Testing: ``CrateLayer.start()`` now raises ``SystemError`` when CrateDB does
not report its HTTP address. The startup timeout is raised from 30s to 90s.
- Testing: ``CrateLayer`` can be used as a context manager, which starts and
stops CrateDB.
- Testing: Fixed ``CrateLayer`` using the parent of the ``crate_config``
directory as ``path.conf``, instead of the directory itself.

2026/10/30 2.4.0
================
- Breaking change: Raised the minimum ``urllib3`` version to 2.8.0. This
Expand Down
14 changes: 10 additions & 4 deletions docs/by-example/http.rst
Original file line number Diff line number Diff line change
Expand Up @@ -32,11 +32,17 @@ If no ``server`` argument (or no argument at all) is passed, the default one
['http://127.0.0.1:4200']
>>> http_client.close()

When using a list of servers, the servers are selected by round-robin:
The following examples use unreachable servers. A short connect timeout makes
them fail fast, while the read timeout leaves CrateDB enough time to respond:

>>> import urllib3
>>> invalid_host1 = "192.0.2.1:9999"
>>> invalid_host2 = "192.0.2.2:9999"
>>> http_client = HttpClient([crate_host, invalid_host1, invalid_host2], timeout=0.3)
>>> timeout = urllib3.Timeout(connect=0.3, read=10)

When using a list of servers, the servers are selected by round-robin:

>>> http_client = HttpClient([crate_host, invalid_host1, invalid_host2], timeout=timeout)
>>> http_client._get_server()
'http://127.0.0.1:44209'

Expand All @@ -50,7 +56,7 @@ When using a list of servers, the servers are selected by round-robin:

Servers with connection errors will be removed from the active server list:

>>> http_client = HttpClient([invalid_host1, invalid_host2, crate_host], timeout=0.3)
>>> http_client = HttpClient([invalid_host1, invalid_host2, crate_host], timeout=timeout)
>>> result = http_client.sql('select name from locations')
>>> http_client._active_servers
['http://127.0.0.1:44209']
Expand All @@ -70,7 +76,7 @@ sleep after the first request::
If no active servers are available and the retry interval is not reached, just use the oldest
inactive one:

>>> http_client = HttpClient([invalid_host1, invalid_host2, crate_host], timeout=0.3)
>>> http_client = HttpClient([invalid_host1, invalid_host2, crate_host], timeout=timeout)
>>> result = http_client.sql('select name from locations')
>>> http_client._active_servers = []
>>> http_client._get_server()
Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -104,7 +104,7 @@ non_interactive = true


[tool.pytest.ini_options]
addopts = "-rA --verbosity=3 --doctest-modules"
addopts = "-ra --verbosity=1 --durations=10 --doctest-modules"
doctest_optionflags = "NORMALIZE_WHITESPACE ELLIPSIS"
minversion = "2.0"
testpaths = [
Expand Down
53 changes: 42 additions & 11 deletions src/crate/testing/layer.py
Original file line number Diff line number Diff line change
Expand Up @@ -93,7 +93,11 @@ def _download_and_extract(uri, directory):
def wait_for_http_url(log, timeout=30, verbose=False):
start = time.monotonic()
while True:
line = log.readline().decode("utf-8").strip()
raw = log.readline()
if not raw:
# EOF: the process exited before publishing its HTTP address.
return None
line = raw.decode("utf-8").strip()
elapsed = time.monotonic() - start
if verbose:
sys.stderr.write("[{:>4.1f}s]{}\n".format(elapsed, line))
Expand All @@ -107,6 +111,7 @@ def wait_for_http_url(log, timeout=30, verbose=False):
class OutputMonitor:
def __init__(self):
self.consumers = []
self._out_thread: Optional[threading.Thread] = None

def consume(self, iterable):
for line in iterable:
Expand Down Expand Up @@ -144,6 +149,7 @@ class CrateLayer:

tmpdir = tempfile.gettempdir()
wait_interval = 0.2
startup_timeout = 90

@staticmethod
def from_uri(
Expand Down Expand Up @@ -244,6 +250,7 @@ def __init__(
self.http_url = http_url_from_host_port(host, port)

self.process: Optional[subprocess.Popen] = None
self.monitor = OutputMonitor()
self.verbose = verbose
self.env = env or {}
self.env.setdefault("CRATE_USE_IPV4", "true")
Expand Down Expand Up @@ -299,14 +306,18 @@ def create_settings(
transport_port,
further_settings=None,
):
if os.path.isdir(crate_config):
conf_dir = crate_config
else:
conf_dir = os.path.dirname(crate_config)
settings = {
"discovery.type": "zen",
"discovery.initial_state_timeout": 0,
"node.name": node_name,
"cluster.name": cluster_name,
"network.host": host,
"http.port": http_port,
"path.conf": os.path.dirname(crate_config),
"path.conf": conf_dir,
"transport.tcp.port": transport_port,
}
if further_settings:
Expand Down Expand Up @@ -346,25 +357,36 @@ def start(self):
# try to read http_url from startup logs
# this is necessary if no static port is assigned
self.http_url = wait_for_http_url(
self.process.stdout, verbose=self.verbose
self.process.stdout,
timeout=self.startup_timeout,
verbose=self.verbose,
)

self.monitor = OutputMonitor()
self.monitor.start(self.process)

if not self.http_url:
self.stop()
else:
sys.stderr.write("HTTP: {}\n".format(self.http_url))
self._wait_for_start()
self._wait_for_master()
sys.stderr.write("\nCrate instance ready.\n")
raise SystemError(
"Failed to start Crate instance: HTTP address not found in "
"startup logs within {0}s. Use verbose=True to see the "
"logs.".format(self.startup_timeout)
)

sys.stderr.write("HTTP: {}\n".format(self.http_url))
self._wait_for_start()
self._wait_for_master()
sys.stderr.write("\nCrate instance ready.\n")

def stop(self):
self.conn_pool.clear()
if self.process:
self.process.terminate()
self.process.communicate(timeout=10)
try:
self.process.communicate(timeout=10)
except subprocess.TimeoutExpired:
self.process.kill()
self.process.communicate()
stdout = self.process.stdout
if stdout:
stdout.close()
Expand All @@ -375,6 +397,13 @@ def stop(self):
def tearDown(self):
self.stop()

def __enter__(self):
self.start()
return self

def __exit__(self, *exc_info):
self.stop()

def _wait_for(self, validator):
start = time.monotonic()

Expand All @@ -390,7 +419,7 @@ def _wait_for(self, validator):
self.stop()
raise e

if wait_time > 90:
if wait_time > self.startup_timeout:
for line in line_buf.lines:
log.error(line)
self.stop()
Expand All @@ -408,7 +437,9 @@ def _wait_for_start(self):
# after the layer starts don't result in 503
def validator():
try:
resp = self.conn_pool.request("HEAD", self.http_url)
resp = self.conn_pool.request(
"HEAD", self.http_url, retries=False
)
return resp.status == 200
except Exception:
return False
Expand Down
19 changes: 16 additions & 3 deletions tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,8 @@

log = logging.getLogger("tests.conftest")

logging.getLogger("urllib3.connectionpool").setLevel(logging.ERROR)


REQUEST_PATH = "crate.client.http.Server.request"
URL_TMPL = "https://cdn.crate.io/downloads/releases/cratedb/{arch}_{os}/crate-6.1.2.{ext}"
Expand Down Expand Up @@ -146,12 +148,23 @@ def get_crate_url() -> str:


def download_cratedb(path: Path):
if not path.exists():
_download_and_extract(path)


def _download_and_extract(path: Path, attempts: int = 3):
url = get_crate_url()
if path.exists():
return
if not url.startswith("https:"):
raise ValueError("Invalid url")
filename, _msg = urlretrieve(url)
for attempt in range(1, attempts + 1):
try:
filename, _msg = urlretrieve(url)
break
except OSError as ex:
if attempt == attempts:
raise
log.warning("Downloading %s failed, retrying: %s", url, ex)
time.sleep(5 * attempt)
if sys.platform.startswith("win32"):
with zipfile.ZipFile(filename) as z:
first_file = z.namelist()[0]
Expand Down
Loading
Loading