Port xbstream_fifo_test.sh to pytest (xbstream_fifo_tests.py)

hashicorp_kmip_setup.py

@@ -58,15 +58,15 @@ def setup_directories(config_dir: Path, data_dir: Path, log_dir: Path, certs_dir
 def check_license(license_path: str | None) -> str:
     info("Checking for license file...")
 
-    # First check environment variable (like bash: export VAULT_LICENSE=...)
-    env_license = os.environ.get("VAULT_LICENSE")
+    # First check environment variable (like bash: export HASHICORP_VAULT_LICENSE=...)
+    env_license = os.environ.get("HASHICORP_VAULT_LICENSE")
     if env_license:
-        info("Using license from environment variable VAULT_LICENSE")
+        info("Using license from environment variable HASHICORP_VAULT_LICENSE")
         return env_license.strip()
 
     # Fall back to --license file path
     if not license_path:
-        error("No license provided. Either set VAULT_LICENSE env var or use --license=/path/to/vault.hclic")
+        error("No license provided. Either set HASHICORP_VAULT_LICENSE env var or use --license=/path/to/vault.hclic")
         sys.exit(1)
 
     p = Path(license_path)
@@ -286,7 +286,7 @@ def parse_args() -> argparse.Namespace:
         epilog="""
 LICENSE:
   Provide via environment variable (recommended):
-    export VAULT_LICENSE=$(cat /path/to/vault.hclic)
+    export HASHICORP_VAULT_LICENSE=$(cat /path/to/vault.hclic)
     python3 hashicorp-kmip-setup.py
 
   Or via file path:
@@ -296,7 +296,7 @@ def parse_args() -> argparse.Namespace:
   python3 hashicorp-kmip-setup.py --license=/path/to/vault.hclic
   python3 hashicorp-kmip-setup.py --license=/path/to/vault.hclic --cert-dir=/custom/path/certs
   python3 hashicorp-kmip-setup.py --license=/path/to/vault.hclic --verbose
-  export VAULT_LICENSE=$(cat vault.hclic) && python3 hashicorp-kmip-setup.py --verbose
+  export HASHICORP_VAULT_LICENSE=$(cat vault.hclic) && python3 hashicorp-kmip-setup.py --verbose
         """,
     )
     parser.add_argument(
@@ -314,7 +314,7 @@ def parse_args() -> argparse.Namespace:
         "--license",
         required=False,
         metavar="FILE",
-        help="Path to vault.hclic. Can also be set via: export VAULT_LICENSE=<content>",
+        help="Path to vault.hclic. Can also be set via: export HASHICORP_VAULT_LICENSE=<content>",
     )
     return parser.parse_args()
 

test_scripts/pxb/README.md

@@ -41,6 +41,10 @@
     - [Run a specific test suite](#run-a-specific-test-suite-3)
     - [Run all tests](#run-all-tests-3)
   - [Test reference — upgrade\_backup\_tests.py](#test-reference--upgrade_backup_testspy)
+- [xbstream\_fifo\_tests.py — FIFO-streamed cloud backup tests](#xbstream_fifo_testspy--fifo-streamed-cloud-backup-tests)
+  - [Additional environment variables](#additional-environment-variables-3)
+  - [How to run tests](#how-to-run-tests-4)
+  - [Test reference — xbstream\_fifo\_tests.py](#test-reference--xbstream_fifo_testspy)
 
 ---
 
@@ -105,7 +109,7 @@ If any of these are unset, KMS tests are skipped. PS 8.0+ only; skipped on MS.
 
 **For KMIP tests** (`test_kmip_component_backup`, `test_crash_backup_encrypted_kmip`):
 
-- Vault types come from `KMIP_CONFIGS` in `test_helper.py` (currently `pykmip`, `fortanix`).
+- Vault types come from `KMIP_CONFIGS` in `test_helper.py` (currently `pykmip`, `fortanix`, `hashicorp`).
 - KMIP tests require PS 8.0+ (skipped on 5.7); `keyring_kmip` tests are skipped on MS.
 - For **Fortanix** vault variants, export:
 
@@ -116,6 +120,14 @@ export FORTANIX_PASSWORD=<your-fortanix-password>
 
 If Fortanix vars are not set, Fortanix-only variants are skipped.
 
+- For **HashiCorp Vault** variants, export a real Vault Enterprise license:
+
+```bash
+export HASHICORP_VAULT_LICENSE=$(cat /path/to/vault.hclic)
+```
+
+If unset, `hashicorp`-only variants are skipped.
+
 ### How to run tests
 
 Use **pytest** with the test file. Useful options:
@@ -290,8 +302,8 @@ pytest inc_backup_load_tests.py --collect-only -q
 | `test_page_tracking_backup`| Non-param     | Page tracking; skipped on 5.7 |
 | `test_crash_backup`         | Param         | `[innodb-no_pt]`, `[innodb-pt]`, `[rocksdb-no_pt]`, `[rocksdb-pt]` |
 | `test_crash_backup_encrypted_keyring_file` | Param `[no_pt]`, `[pt]` | Encrypted crash flow using keyring_file component |
-| `test_crash_backup_encrypted_kmip` | Param | One id per `vault_type` x page tracking, e.g. `[pykmip-no_pt]`, `[pykmip-pt]`; Fortanix variants require `FORTANIX_EMAIL`, `FORTANIX_PASSWORD` |
-| `test_kmip_component_backup`| Param         | One id per vault in `KMIP_CONFIGS` (e.g. `[pykmip]`, `[fortanix]`); Fortanix variants require `FORTANIX_EMAIL`, `FORTANIX_PASSWORD` |
+| `test_crash_backup_encrypted_kmip` | Param | One id per `vault_type` x page tracking, e.g. `[pykmip-no_pt]`, `[pykmip-pt]`; Fortanix variants require `FORTANIX_EMAIL`, `FORTANIX_PASSWORD`; hashicorp variants require `HASHICORP_VAULT_LICENSE` |
+| `test_kmip_component_backup`| Param         | One id per vault in `KMIP_CONFIGS` (e.g. `[pykmip]`, `[fortanix]`, `[hashicorp]`); Fortanix variants require `FORTANIX_EMAIL`, `FORTANIX_PASSWORD`; hashicorp variants require `HASHICORP_VAULT_LICENSE` |
 | `test_kms_component_backup` | Param `[no_pt]`, `[pt]` | keyring_kms component; requires `KMS_KEYID`, `KMS_SECRET_KEY`, `KMS_AUTH_KEY`, `KMS_REGION`; skipped on 5.7 and MS |
 
 ---
@@ -830,3 +842,60 @@ python upgrade_backup_tests.py All
 | `test_upgrade_full_backup` | 1. full backup with previous PXB → prepare/restore with current PXB | Plain `--log-bin=binlog` server options; sysbench load runs in background |
 | `test_upgrade_inc_backup`  | 1. full + inc with previous PXB → prepare/restore with current PXB; 2. full with previous PXB + inc with current PXB → prepare/restore with current PXB | Two scenarios run sequentially against the same primary; sysbench load on each |
 | `test_upgrade_backup_encrypt` | 1. full prev → prepare/restore current; 2. full + inc prev → prepare/restore current; 3. full prev + inc current → prepare/restore current | Auto-detects server type (PS / MS / 5.7) and applies matching keyring + encryption options; uses `--keyring_file_data=<MYSQLDIR>/keyring` and per-binary `--xtrabackup-plugin-dir` for both PXB binaries |
+
+---
+
+## xbstream_fifo_tests.py — FIFO-streamed cloud backup tests
+
+Tests in `xbstream_fifo_tests.py` run full/incremental/compressed/partition-table/encrypted backups streamed through `xtrabackup`/`xbcloud`/`xbstream` named pipes (`--fifo-streams`/`--fifo-dir`) to a local [SeaweedFS](https://github.com/seaweedfs/seaweedfs) S3-gateway container, instead of a single-process shell pipe (which is what `take_backup(backup_type="cloud")` in the other test files uses). This exercises `xbcloud`'s multi-stream parallel upload/download path against an S3-compatible backend, without needing real cloud credentials.
+
+**Additional requirement:** Docker must be installed and running — `xbstream_fifo_tests.py` starts a `chrislusf/seaweedfs` container (`seaweedfs_helper.py`) once per test session (port `9000` on the host by default) and stops it when the session ends.
+
+### Additional environment variables
+
+```bash
+export LOAD_TOOL=pstress
+export LOAD_TOOL_DIR=$HOME/lab/pstress/src
+export FIFO_STREAM=30                # number of named pipes; default 30
+export FIFO_DIR=/tmp/xbstream_fifo   # FIFO pipe directory; default /tmp/xbstream_fifo
+```
+
+`test_fifo_partition_tables`, `test_fifo_keyring_file_backup`, and `test_fifo_kmip_backup` require `LOAD_TOOL=pstress` (they're skipped otherwise, since they rely on pstress's partition-table/encrypted-table DDL).
+
+**For KMIP tests** (`test_fifo_kmip_backup`): same requirements as `inc_backup_load_tests.py`'s `test_kmip_component_backup` — vault types come from `KMIP_CONFIGS`, skipped on 5.7/MS, Fortanix variants need `FORTANIX_EMAIL`/`FORTANIX_PASSWORD`, and hashicorp variants need `HASHICORP_VAULT_LICENSE` (a real Vault Enterprise license).
+
+### How to run tests
+
+```bash
+# Single test
+pytest xbstream_fifo_tests.py -v -s -k test_fifo_full_backup_and_restore
+
+# Multiple tests
+pytest xbstream_fifo_tests.py -v -s -k "test_fifo_full_backup_and_restore or test_fifo_compressed_backup"
+
+# One KMIP vault variant
+pytest xbstream_fifo_tests.py -v -s -k "test_fifo_kmip_backup[pykmip]"
+
+# Whole file
+pytest xbstream_fifo_tests.py -v -s
+
+# Via the script's built-in suites
+python xbstream_fifo_tests.py Fifo_Backup_tests
+python xbstream_fifo_tests.py Fifo_Partition_tests
+python xbstream_fifo_tests.py Fifo_Encryption_tests
+python xbstream_fifo_tests.py Fifo_Kmip_tests
+```
+
+Redirecting `-s` output to a log file for `tail -f`? Python fully buffers stdout when it isn't a TTY, so a genuinely-progressing test can look stuck. Use `python3 -u -m pytest ...` (or `export PYTHONUNBUFFERED=1`) to keep the log flushing in real time.
+
+### Test reference — xbstream_fifo_tests.py
+
+| Test | Type | Notes |
+|------|------|-------|
+| `test_fifo_full_backup_and_restore` | Non-param | Full backup and restore |
+| `test_fifo_incremental_backup` | Non-param | Full + 3 incrementals (5s apart), each restored/prepared separately (mirrors the original bash script's fixed-count design, not the load-until-done loop `take_backup()` uses elsewhere) |
+| `test_fifo_compressed_backup` | Non-param | Full backup with `--compress=zstd --compress-zstd-level=19` |
+| `test_fifo_partition_tables` | Non-param | Incremental backup of pstress-generated partitioned tables; requires `LOAD_TOOL=pstress` |
+| `test_fifo_keyring_file_backup` | Non-param | keyring_file component encrypted incremental backup; requires `LOAD_TOOL=pstress`. As of PXB 8.4.0-7/PS 8.4.10-10 this can occasionally crash `xtrabackup` with an InnoDB assertion (`fil0fil.cc:...:page_id.space() != TRX_SYS_SPACE`) while parsing the redo log of an encrypted incremental — a product-level PXB/InnoDB bug, not a bug in this test |
+| `test_fifo_kmip_backup` | Param | One id per vault in `KMIP_CONFIGS` (e.g. `[pykmip]`, `[fortanix]`, `[hashicorp]`); requires `LOAD_TOOL=pstress`; skipped on 5.7/MS; Fortanix variants require `FORTANIX_EMAIL`, `FORTANIX_PASSWORD`; hashicorp variants require `HASHICORP_VAULT_LICENSE` |
+| `test_fifo_encrypted_backup` | Non-param | Full backup encrypted with xbcrypt (`--encrypt=AES256`, not keyring-based) |

test_scripts/pxb/kmip_helper.py

@@ -35,10 +35,10 @@
     """Helper class for managing KMIP servers."""
 
     # Default KMIP configurations
     DEFAULT_KMIP_CONFIGS = {
         "pykmip": "addr=127.0.0.1,image=satyapercona/kmip:latest,port=5696,name=kmip_pykmip",
         # "fortanix": "addr=216.180.120.88,port=5696,name=kmip_fortanix,setup_script=fortanix_kmip_setup.py",
-        # "hashicorp": "addr=127.0.0.1,port=5696,name=kmip_hashicorp,setup_script=hashicorp_kmip_setup.py",
+        "hashicorp": "addr=127.0.0.1,port=5696,name=kmip_hashicorp,setup_script=hashicorp_kmip_setup.py",
         # "ciphertrust": "addr=127.0.0.1,port=5696,name=kmip_ciphertrust,setup_script=setup_kmip_api.py",
     }
 
@@ -402,7 +402,8 @@
                 script_content = response.read().decode("utf-8")
 
             if not script_content:
-                print("Downloaded script is empty")
+                self.last_error = "Downloaded HashiCorp setup script is empty"
+                print(self.last_error)
                 return False
 
             if os.path.exists(cert_dir):
@@ -411,21 +412,29 @@
 
             os.makedirs(cert_dir, exist_ok=True)
 
-            # Execute the script
+            # Execute the script (it's a real Python script, run via
+            # `python3 -` to read it from stdin, matching how the retired
+            # bash kmip_helper.sh invoked it: `python3 - --cert-dir=...`)
             result = subprocess.run(
-                ["bash", "-s", "--", f"--cert-dir={cert_dir}"],
+                ["python3", "-", f"--cert-dir={cert_dir}"],
                 input=script_content,
                 text=True,
                 capture_output=True,
                 check=False,
             )
             if result.returncode != 0:
-                print(f"Failed to execute script {setup_script}, (exit code: {result.returncode})")
+                detail = (result.stderr or result.stdout or "").strip()
+                self.last_error = (
+                    f"HashiCorp setup script exited with code {result.returncode}"
+                    + (f": {detail}" if detail else "")
+                )
+                print(f"Failed to execute script {setup_script}: {self.last_error}")
                 return False
 
             self.generate_kmip_config(kmip_type, addr, port, cert_dir)
         except Exception as e:
-            print(f"Failed to setup HashiCorp: {e}")
+            self.last_error = f"Failed to setup HashiCorp: {e}"
+            print(self.last_error)
             return False
 
         self.kmip_config["cert_dir"] = cert_dir

test_scripts/pxb/seaweedfs_helper.py

@@ -0,0 +1,143 @@
+#!/usr/bin/env python3
+"""
+SeaweedFS Helper Library
+Usage: from seaweedfs_helper import SeaweedFSHelper
+
+Manages a local SeaweedFS S3-gateway container used as the S3-compatible
+backend for xbstream_fifo_tests.py's cloud/FIFO backup tests. Ported from
+xbstream_fifo_test.sh's start_seaweedfs()/cleanup_exit() shell functions.
+Required: Docker must be installed and running.
+"""
+
+import json
+import os
+import subprocess
+import time
+import urllib.request
+import urllib.error
+from typing import Optional
+
+CONTAINER_NAME = "seaweedfs"
+IMAGE = "chrislusf/seaweedfs:latest"
+HOST_PORT = 9000
+CONTAINER_S3_PORT = 8333
+S3_ACCESS_KEY = "admin"
+S3_SECRET_KEY = "password"
+
+
+class SeaweedFSHelper:
+    """Helper class for starting/stopping the SeaweedFS S3-gateway container."""
+
+    def __init__(self, data_dir: Optional[str] = None, host_port: int = HOST_PORT):
+        self.data_dir = data_dir or os.path.join(os.path.expanduser("~"), "seaweedfs", "data")
+        self.config_dir = os.path.dirname(self.data_dir)
+        self.host_port = host_port
+        self.last_error: str = ""
+
+    def _container_status(self) -> Optional[str]:
+        """Return 'running', 'stopped', or None if the container doesn't exist."""
+        result = subprocess.run(
+            ["docker", "ps", "--filter", f"name={CONTAINER_NAME}", "--filter", "status=running",
+             "--format", "{{.Names}}"],
+            capture_output=True, text=True, check=False,
+        )
+        if CONTAINER_NAME in result.stdout.split():
+            return "running"
+        result = subprocess.run(
+            ["docker", "ps", "-a", "--filter", f"name={CONTAINER_NAME}", "--format", "{{.Names}}"],
+            capture_output=True, text=True, check=False,
+        )
+        if CONTAINER_NAME in result.stdout.split():
+            return "stopped"
+        return None
+
+    def _write_s3_config(self) -> str:
+        """Write the S3 identity config (admin/password) SeaweedFS's S3 gateway needs."""
+        os.makedirs(self.config_dir, exist_ok=True)
+        os.makedirs(self.data_dir, exist_ok=True)
+        config = {
+            "identities": [
+                {
+                    "name": "admin",
+                    "credentials": [{"accessKey": S3_ACCESS_KEY, "secretKey": S3_SECRET_KEY}],
+                    "actions": ["Admin", "Read", "Write"],
+                }
+            ]
+        }
+        config_path = os.path.join(self.config_dir, "s3.json")
+        with open(config_path, "w", encoding="utf-8") as f:
+            json.dump(config, f, indent=2)
+        return config_path
+
+    def _wait_ready(self, timeout: int = 60) -> bool:
+        """Poll the S3 gateway port. SeaweedFS has no MinIO-style
+        /minio/health/ready endpoint, so any HTTP response (even an error
+        status) is enough to know xbcloud can reach it."""
+        url = f"http://localhost:{self.host_port}/"
+        for _ in range(timeout):
+            try:
+                urllib.request.urlopen(url, timeout=1)
+                return True
+            except urllib.error.HTTPError:
+                # Any HTTP response (e.g. 403/404) means the gateway is up.
+                return True
+            except (urllib.error.URLError, OSError):
+                time.sleep(1)
+        return False
+
+    def start(self) -> bool:
+        """Start the SeaweedFS container if it isn't already running."""
+        status = self._container_status()
+        if status == "running":
+            print("SeaweedFS is already running.")
+        elif status == "stopped":
+            print("Found stopped SeaweedFS container. Starting it...")
+            result = subprocess.run(
+                ["docker", "start", CONTAINER_NAME],
+                capture_output=True, text=True, check=False,
+            )
+            if result.returncode != 0:
+                self.last_error = result.stderr or result.stdout
+                print(f"ERR: Failed to start SeaweedFS container: {self.last_error}")
+                return False
+        else:
+            if os.path.isdir(self.data_dir):
+                for entry in os.listdir(self.data_dir):
+                    path = os.path.join(self.data_dir, entry)
+                    subprocess.run(["rm", "-rf", path], check=False)
+            config_path = self._write_s3_config()
+            print("No SeaweedFS container found. Creating and starting one...")
+            result = subprocess.run(
+                [
+                    "docker", "run", "-d",
+                    "-p", f"{self.host_port}:{CONTAINER_S3_PORT}",
+                    "--name", CONTAINER_NAME,
+                    "-v", f"{self.data_dir}:/data",
+                    "-v", f"{config_path}:/etc/seaweedfs/s3.json",
+                    IMAGE,
+                    "server", "-s3", f"-s3.port={CONTAINER_S3_PORT}",
+                    "-s3.config=/etc/seaweedfs/s3.json", "-dir=/data",
+                ],
+                capture_output=True, text=True, check=False,
+            )
+            if result.returncode != 0:
+                self.last_error = result.stderr
+                print(f"ERR: Failed to start SeaweedFS container: {result.stderr}")
+                return False
+
+        print("Waiting for SeaweedFS to become ready", end="", flush=True)
+        if self._wait_ready():
+            print("\nSeaweedFS is ready!\n")
+            return True
+
+        print("\nSeaweedFS failed to become ready in time.")
+        logs = subprocess.run(["docker", "logs", CONTAINER_NAME], capture_output=True, text=True, check=False)
+        self.last_error = logs.stdout + logs.stderr
+        print(self.last_error)
+        return False
+
+    def stop(self) -> None:
+        """Stop the SeaweedFS container if it's running."""
+        if self._container_status() == "running":
+            print("Stopping SeaweedFS container...")
+            subprocess.run(["docker", "stop", CONTAINER_NAME], capture_output=True, check=False)