wtf: Improve test cases

cratedb_toolkit/sqlalchemy/patch.py

@@ -5,9 +5,15 @@
 from decimal import Decimal
 from uuid import UUID
 
-import numpy as np
 import sqlalchemy as sa
 
+try:
+    import numpy as np
+
+    has_numpy = True
+except ImportError:
+    has_numpy = False
+
 
 def patch_inspector():
     """
@@ -65,10 +71,11 @@ def default(self, o):
 
         # NumPy ndarray and friends.
         # https://stackoverflow.com/a/49677241
-        if isinstance(o, np.integer):
-            return int(o)
-        elif isinstance(o, np.floating):
-            return float(o)
-        elif isinstance(o, np.ndarray):
-            return o.tolist()
+        if has_numpy:
+            if isinstance(o, np.integer):
+                return int(o)
+            elif isinstance(o, np.floating):
+                return float(o)
+            elif isinstance(o, np.ndarray):
+                return o.tolist()
         return json.JSONEncoder.default(self, o)

cratedb_toolkit/wtf/cli.py

@@ -144,16 +144,21 @@ def job_statistics(ctx: click.Context):
 
 
 @make_command(job_statistics, "collect", "Collect queries from sys.jobs_log.")
+@click.option("--once", is_flag=True, default=False, required=False, help="Whether to record only one sample")
 @click.pass_context
-def job_statistics_collect(ctx: click.Context):
+def job_statistics_collect(ctx: click.Context, once: bool):
     """
     Run jobs_log collector.
 
     # TODO: Forward `cratedb_sqlalchemy_url` properly.
     """
     import cratedb_toolkit.wtf.query_collector
 
-    cratedb_toolkit.wtf.query_collector.main()
+    cratedb_toolkit.wtf.query_collector.init()
+    if once:
+        cratedb_toolkit.wtf.query_collector.record_once()
+    else:
+        cratedb_toolkit.wtf.query_collector.record_forever()
 
 
 @make_command(job_statistics, "view", "View job statistics about collected queries.")
@@ -180,13 +185,17 @@ def job_statistics_view(ctx: click.Context):
 
 
 @make_command(cli, "record", "Record `info` and `job-info` outcomes.")
+@click.option("--once", is_flag=True, default=False, required=False, help="Whether to record only one sample")
 @click.pass_context
-def record(ctx: click.Context):
+def record(ctx: click.Context, once: bool):
     cratedb_sqlalchemy_url = ctx.meta["cratedb_sqlalchemy_url"]
     scrub = ctx.meta.get("scrub", False)
     adapter = DatabaseAdapter(dburi=cratedb_sqlalchemy_url, echo=False)
     recorder = InfoRecorder(adapter=adapter, scrub=scrub)
-    recorder.record_forever()
+    if once:
+        recorder.record_once()
+    else:
+        recorder.record_forever()
 
 
 @make_command(cli, "serve", help_serve)

cratedb_toolkit/wtf/http.py

@@ -2,9 +2,9 @@
 # Distributed under the terms of the AGPLv3 license, see LICENSE.
 import logging
 import os
-import typing as t
 from functools import lru_cache
 
+import typing_extensions as t
 from fastapi import Depends, FastAPI, HTTPException
 
 from cratedb_toolkit.util import DatabaseAdapter

cratedb_toolkit/wtf/query_collector.py

@@ -102,7 +102,7 @@ def init_stmts(stmts):
 
 
 def write_stats_to_db():
-    logger.info("Writing statistics to database")
+    logger.info(f"Writing statistics to database table: {stmt_log_table}")
     write_query_stmt = (
         f"INSERT INTO {stmt_log_table} "
         f"(id, stmt, calls, bucket, username, query_type, avg_duration, nodes, last_used) "
@@ -230,18 +230,23 @@ def scrape_db():
     last_scrape = next_scrape
 
 
-def run():
+def record_once():
+    logger.info("Recording information snapshot")
     scrape_db()
     write_stats_to_db()
 
 
-def main():
-    init()
+def record_forever():
     while True:
-        run()
+        record_once()
         logger.info(f"Sleeping for {interval} seconds")
         time.sleep(interval)
 
 
+def main():
+    init()
+    record_forever()
+
+
 if __name__ == "__main__":
     main()

cratedb_toolkit/wtf/recorder.py

@@ -55,4 +55,5 @@ def do_record_forever(self):
                 self.record_once()
             except Exception:
                 logger.exception("Failed to record information snapshot")
+            logger.info(f"Sleeping for {self.interval_seconds} seconds")
             time.sleep(self.interval_seconds)

pyproject.toml

@@ -164,6 +164,7 @@ service = [
 ]
 test = [
   "cratedb-toolkit[testing]",
+  "httpx<0.28",
   "pueblo[dataframe]",
   "pytest<9",
   "pytest-cov<6",
@@ -288,6 +289,7 @@ extend-exclude = [
 [tool.ruff.lint.per-file-ignores]
 "doc/conf.py" = ["A001", "ERA001"]
 "tests/*" = ["S101"]  # Allow use of `assert`, and `print`.
+"tests/wtf/test_http.py" = ["E402"]
 "examples/*" = ["T201"]  # Allow `print`
 "cratedb_toolkit/retention/cli.py" = ["T201"]  # Allow `print`
 "cratedb_toolkit/sqlalchemy/__init__.py" = ["F401"]  # Allow `module´ imported but unused

tests/cfr/test_cli.py

@@ -20,7 +20,7 @@ def filenames(path: Path):
     return sorted([item.name for item in path.iterdir()])
 
 
-def test_cfr_cli_export(cratedb, tmp_path, caplog):
+def test_cfr_cli_export_success(cratedb, tmp_path, caplog):
     """
     Verify `ctk cfr sys-export` works.
     """
@@ -49,7 +49,26 @@ def test_cfr_cli_export(cratedb, tmp_path, caplog):
     assert len(data_files) >= 10
 
 
-def test_cfr_cli_import(cratedb, tmp_path, caplog):
+def test_cfr_cli_export_failure(cratedb, tmp_path, caplog):
+    """
+    Verify `ctk cfr sys-export` failure.
+    """
+
+    # Invoke command.
+    runner = CliRunner(env={"CRATEDB_SQLALCHEMY_URL": "crate://foo.bar/", "CFR_TARGET": str(tmp_path)})
+    result = runner.invoke(
+        cli,
+        args="--debug sys-export",
+        catch_exceptions=False,
+    )
+    assert result.exit_code == 1
+
+    # Verify log output.
+    assert "Failed to establish a new connection" in caplog.text or "Failed to resolve" in caplog.text
+    assert result.output == ""
+
+
+def test_cfr_cli_import_success(cratedb, tmp_path, caplog):
     """
     Verify `ctk cfr sys-import` works.
     """
@@ -108,3 +127,22 @@ def test_cfr_cli_import(cratedb, tmp_path, caplog):
 
     cratedb.database.run_sql('REFRESH TABLE "sys-operations"')
     assert cratedb.database.count_records("sys-operations") == 1
+
+
+def test_cfr_cli_import_failure(cratedb, tmp_path, caplog):
+    """
+    Verify `ctk cfr sys-import` failure.
+    """
+
+    # Invoke command.
+    runner = CliRunner(env={"CRATEDB_SQLALCHEMY_URL": "crate://foo.bar/", "CFR_SOURCE": str(tmp_path)})
+    result = runner.invoke(
+        cli,
+        args="--debug sys-import",
+        catch_exceptions=False,
+    )
+    assert result.exit_code == 1
+
+    # Verify log output.
+    assert "Failed to establish a new connection" in caplog.text or "Failed to resolve" in caplog.text
+    assert result.output == ""

tests/conftest.py

@@ -14,6 +14,13 @@
 TESTDRIVE_DATA_SCHEMA = "testdrive-data"
 TESTDRIVE_EXT_SCHEMA = "testdrive-ext"
 RESET_TABLES = [
+    # FIXME: Let all subsystems use configured schema instead of hard-coded ones.
+    '"doc"."clusterinfo"',
+    '"doc"."jobinfo"',
+    '"ext"."clusterinfo"',
+    '"ext"."jobinfo"',
+    '"stats"."statement_log"',
+    '"stats"."last_execution"',
     f'"{TESTDRIVE_EXT_SCHEMA}"."retention_policy"',
     f'"{TESTDRIVE_DATA_SCHEMA}"."raw_metrics"',
     f'"{TESTDRIVE_DATA_SCHEMA}"."sensor_readings"',

tests/sqlalchemy/test_patch.py

@@ -1,6 +1,11 @@
+import datetime
+import json
+
+import pytest
 import sqlalchemy as sa
 
 from cratedb_toolkit.sqlalchemy import patch_inspector
+from cratedb_toolkit.sqlalchemy.patch import CrateJsonEncoderWithNumPy
 from tests.conftest import TESTDRIVE_DATA_SCHEMA
 
 
@@ -40,3 +45,23 @@ def test_inspector_patched(database):
 
     table_names = inspector.get_table_names()
     assert "foobar" in table_names
+
+
+def test_json_encoder_date():
+    """
+    Verify the extended JSON encoder also accepts Python's `date` types.
+    """
+    data = {"date": datetime.date(2024, 6, 4)}
+    encoded = json.dumps(data, cls=CrateJsonEncoderWithNumPy)
+    assert encoded == '{"date": 1717459200000}'
+
+
+def test_json_encoder_numpy():
+    """
+    Verify the extended JSON encoder also accepts NumPy types.
+    """
+    np = pytest.importorskip("numpy")
+
+    data = {"scalar-int": np.float32(42.42).astype(int), "scalar-float": np.float32(42.42), "ndarray": np.ndarray([1])}
+    encoded = json.dumps(data, cls=CrateJsonEncoderWithNumPy)
+    assert encoded == """{"scalar-int": 42, "scalar-float": 42.41999816894531, "ndarray": [2.08e-322]}"""

tests/util/test_platform.py

@@ -0,0 +1,15 @@
+from cratedb_toolkit.util.platform import PlatformInfo
+
+
+def test_platforminfo_application():
+    pi = PlatformInfo()
+    outcome = pi.application()
+    assert "name" in outcome
+    assert "version" in outcome
+    assert "platform" in outcome
+
+
+def test_platforminfo_libraries():
+    pi = PlatformInfo()
+    outcome = pi.libraries()
+    assert isinstance(outcome, dict)

tests/wtf/test_cli.py

@@ -2,6 +2,7 @@
 
 from boltons.iterutils import get_path
 from click.testing import CliRunner
+from yarl import URL
 
 from cratedb_toolkit.wtf.cli import cli
 
@@ -88,6 +89,40 @@ def test_wtf_cli_job_info(cratedb):
     assert "performance15min" in data_keys
 
 
+def test_wtf_cli_statistics_collect(cratedb, caplog):
+    """
+    Verify `cratedb-wtf job-statistics collect`.
+    """
+
+    uri = URL(cratedb.database.dburi)
+
+    # Invoke command.
+    runner = CliRunner(env={"CRATEDB_SQLALCHEMY_URL": cratedb.database.dburi})
+    result = runner.invoke(
+        cli,
+        args="job-statistics collect --once",
+        env={"HOSTNAME": f"{uri.host}:{uri.port}"},
+        catch_exceptions=False,
+    )
+    assert result.exit_code == 0
+
+    # Verify outcome: Log output.
+    assert "Recording information snapshot" in caplog.messages
+
+    # Verify outcome: Database content.
+    # stats.statement_log, stats.last_execution
+    results = cratedb.database.run_sql("SHOW TABLES", records=True)
+    assert {"table_name": "last_execution"} in results
+    assert {"table_name": "statement_log"} in results
+
+    # FIXME: Table is empty. Why?
+    cratedb.database.run_sql('REFRESH TABLE "stats"."statement_log"')
+    assert cratedb.database.count_records("stats.statement_log") == 0
+
+    cratedb.database.run_sql('REFRESH TABLE "stats"."last_execution"')
+    assert cratedb.database.count_records("stats.last_execution") == 1
+
+
 def test_wtf_cli_statistics_view(cratedb):
     """
     Verify `cratedb-wtf job-statistics view`.
@@ -109,3 +144,32 @@ def test_wtf_cli_statistics_view(cratedb):
 
     data_keys = list(info["data"].keys())
     assert "stats" in data_keys
+
+
+def test_wtf_cli_record(cratedb, caplog):
+    """
+    Verify `cratedb-wtf record`.
+    """
+
+    # Invoke command.
+    runner = CliRunner(env={"CRATEDB_SQLALCHEMY_URL": cratedb.database.dburi})
+    result = runner.invoke(
+        cli,
+        args="--debug record --once",
+        catch_exceptions=False,
+    )
+    assert result.exit_code == 0
+
+    # Verify outcome: Log output.
+    assert "Recording information snapshot" in caplog.messages
+
+    # Verify outcome: Database content.
+    results = cratedb.database.run_sql("SHOW TABLES", records=True)
+    assert {"table_name": "clusterinfo"} in results
+    assert {"table_name": "jobinfo"} in results
+
+    cratedb.database.run_sql('REFRESH TABLE "ext"."clusterinfo"')
+    assert cratedb.database.count_records("ext.clusterinfo") == 1
+
+    cratedb.database.run_sql('REFRESH TABLE "ext"."jobinfo"')
+    assert cratedb.database.count_records("ext.jobinfo") == 1