OSAC-3593: Migrate bmaas e2e test suite & core test infra from osac-test-infra (1/5)

pyproject.toml

@@ -0,0 +1,54 @@
+[project]
+name = "osac-e2e"
+version = "0.1.0"
+description = "OSAC end-to-end test infrastructure"
+requires-python = ">=3.11"
+dependencies = [
+    "pytest>=8.0",
+    "pytest-xdist>=3.0",
+    "pyyaml>=6.0",
+    "websocket-client>=1.6",
+]
+
+[tool.setuptools.packages.find]
+include = ["tests*"]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+pythonpath = ["."]
+addopts = "-n 4 --dist loadfile"
+markers = [
+    "sanity: fast/essential e2e coverage for PR checks",
+    "regression: heavier or cross-flow e2e for periodic runs",
+]
+log_level = "INFO"
+log_cli = true
+log_cli_date_format = "%Y-%m-%dT%H:%M:%S%z"
+log_cli_format = "%(asctime)s %(levelname)-8s %(message)s"
+log_cli_level = "INFO"
+log_file = "/tmp/test-output/e2e.log"
+log_file_date_format = "%Y-%m-%dT%H:%M:%S%z"
+log_file_format = "%(asctime)s %(levelname)-8s %(name)s %(message)s"
+log_file_level = "INFO"
+log_file_mode = "a"
+
+[tool.basedpyright]
+reportUnusedCallResult = false
+reportAny = false
+reportExplicitAny = false
+
+[tool.ruff]
+target-version = "py311"
+line-length = 120
+
+[tool.ruff.format]
+skip-magic-trailing-comma = true
+
+[tool.ruff.lint.isort]
+split-on-trailing-comma = false
+
+[tool.ruff.lint]
+select = ["E", "F", "W", "I", "UP", "ANN", "B", "SIM", "RUF"]
+
+[tool.ruff.lint.per-file-ignores]
+"tests/*" = ["ANN101", "ANN201"]

tests/bmaas/conftest.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+import subprocess
+import tempfile
+import uuid
+from collections.abc import Generator
+from pathlib import Path
+
+import pytest
+
+from tests.core.grpc_client import GRPCClient
+from tests.core.runner import env
+
+
+@pytest.fixture(scope="session")
+def bmi_template() -> str:
+    return env("OSAC_BMI_TEMPLATE", "bm-host-provisioning")
+
+
+@pytest.fixture(scope="session")
+def bmh_namespace() -> str:
+    return env("OSAC_BMH_NAMESPACE", "host-inventory")
+
+
+@pytest.fixture(scope="session")
+def test_run_id() -> str:
+    return str(uuid.uuid4())[:8]
+
+
+@pytest.fixture(scope="session")
+def ssh_public_key() -> Generator[str, None, None]:
+    with tempfile.TemporaryDirectory() as tmpdir:
+        key_path = Path(tmpdir) / "bmi-test-key"
+        subprocess.run(
+            ["ssh-keygen", "-t", "ed25519", "-f", str(key_path), "-N", "", "-C", "bmi-e2e-test"],
+            capture_output=True,
+            check=True,
+        )
+        yield (key_path.with_suffix(".pub")).read_text().strip()
+
+
+@pytest.fixture(scope="session")
+def catalog_item(private_grpc: GRPCClient, bmi_template: str, test_run_id: str) -> Generator[str, None, None]:
+    name = f"e2e-bmaas-{test_run_id}"
+    print(f"\nCreating BareMetalInstanceCatalogItem: {name}")
+    item_id: str = private_grpc.create_baremetal_instance_catalog_item(
+        name=name,
+        title=f"E2E BMaaS Test ({test_run_id})",
+        description="Temporary catalog item for BMaaS E2E tests",
+        template=bmi_template,
+        field_definitions=[{"path": "ssh_public_key", "display_name": "SSH Public Key", "editable": True}],
+    )
+    print(f"CatalogItem created: {item_id}")
+
+    yield item_id
+
+    try:
+        print(f"\nDeleting BareMetalInstanceCatalogItem {item_id}...")
+        private_grpc.delete_baremetal_instance_catalog_item(item_id=item_id)
+        print(f"CatalogItem {item_id} deleted")
+    except Exception as e:
+        print(f"WARNING: Failed to delete catalog item {item_id}: {e}")

tests/bmaas/networking/__init__.py

@@ -0,0 +1 @@
+from __future__ import annotations

tests/bmaas/networking/bmi_ssh.py

@@ -0,0 +1,106 @@
+from __future__ import annotations
+
+import logging
+import subprocess
+
+log = logging.getLogger(__name__)
+
+_SSH_OPTS = [
+    "-o",
+    "ConnectTimeout=10",
+    "-o",
+    "StrictHostKeyChecking=no",
+    "-o",
+    "UserKnownHostsFile=/dev/null",
+    "-i",
+    "/root/.ssh/id_rsa",
+]
+
+
+def get_bmc_ip(bmh_name: str) -> str:
+    domiflist = subprocess.run(["virsh", "domiflist", bmh_name], capture_output=True, text=True, timeout=10, check=True)
+    bmc_mac = ""
+    for line in domiflist.stdout.splitlines():
+        if "bmc-net" in line:
+            bmc_mac = line.split()[4]
+            break
+    if not bmc_mac:
+        raise RuntimeError(f"No bmc-net interface found for VM {bmh_name}")
+
+    leases = subprocess.run(
+        ["virsh", "net-dhcp-leases", "bmc-net"], capture_output=True, text=True, timeout=10, check=True
+    )
+    for line in leases.stdout.splitlines():
+        if bmc_mac in line:
+            ip_with_prefix = line.split()[4]
+            return ip_with_prefix.split("/")[0]
+    raise RuntimeError(f"No DHCP lease found for MAC {bmc_mac} on bmc-net")
+
+
+def ssh_bmi(bmc_ip: str, command: str, timeout: int = 30) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        ["ssh", *_SSH_OPTS, f"fedora@{bmc_ip}", command], capture_output=True, text=True, timeout=timeout, check=True
+    )
+
+
+def ssh_bmi_unchecked(bmc_ip: str, command: str, timeout: int = 30) -> tuple[str, int]:
+    try:
+        result = subprocess.run(
+            ["ssh", *_SSH_OPTS, f"fedora@{bmc_ip}", command],
+            capture_output=True,
+            text=True,
+            timeout=timeout,
+            check=False,
+        )
+    except subprocess.TimeoutExpired:
+        log.warning("ssh_bmi_unchecked(%s): subprocess timed out after %ds", bmc_ip, timeout)
+        return f"ssh timed out after {timeout}s", 255
+    return (result.stdout.strip() + "\n" + result.stderr.strip()).strip(), result.returncode
+
+
+def arping(bmc_ip: str, target_ip: str, interface: str = "ens5", count: int = 3) -> bool:
+    _, rc = ssh_bmi_unchecked(bmc_ip, f"arping -c {count} -I {interface} {target_ip}", timeout=30)
+    return rc == 0
+
+
+def ping(bmc_ip: str, target_ip: str, count: int = 3, wait: int = 3) -> bool:
+    _, rc = ssh_bmi_unchecked(bmc_ip, f"ping -c {count} -W {wait} {target_ip}", timeout=30)
+    return rc == 0
+
+
+def curl_status(bmc_ip: str, url: str, timeout: int = 15) -> int:
+    output, rc = ssh_bmi_unchecked(
+        bmc_ip,
+        f"curl -s -o /dev/null -w '%{{http_code}}' --connect-timeout {timeout} {url}",
+        timeout=timeout + 30,
+    )
+    log.info("curl_status(%s, %s): ssh_rc=%d, output=%r", bmc_ip, url, rc, output[-200:])
+    for line in output.strip().splitlines():
+        line = line.strip()
+        if line.isdigit() and len(line) == 3:
+            return int(line)
+    log.warning("curl_status: no HTTP status code found in output, returning 0")
+    return 0
+
+
+def ssh_via_external_ip(external_ip: str, command: str = "hostname", timeout: int = 15) -> str:
+    result = subprocess.run(
+        [
+            "ssh",
+            "-o",
+            f"ConnectTimeout={timeout}",
+            "-o",
+            "StrictHostKeyChecking=no",
+            "-o",
+            "UserKnownHostsFile=/dev/null",
+            "-i",
+            "/root/.ssh/id_rsa",
+            f"fedora@{external_ip}",
+            command,
+        ],
+        capture_output=True,
+        text=True,
+        timeout=timeout + 10,
+        check=True,
+    )
+    return result.stdout.strip()

tests/bmaas/networking/conftest.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+import uuid
+from pathlib import Path
+
+import pytest
+
+from tests.core.grpc_client import GRPCClient
+from tests.core.runner import env
+
+
+@pytest.fixture(scope="session")
+def external_ip_pool_name() -> str:
+    return env("OSAC_EXTERNAL_IP_POOL", "tenant-external-pool")
+
+
+@pytest.fixture(scope="session")
+def external_ip_pool_cidr() -> str:
+    return env("OSAC_EXTERNAL_IP_POOL_CIDR", "198.51.100.24/29")
+
+
+@pytest.fixture(scope="session")
+def auto_eip_catalog_item_name() -> str:
+    return env("OSAC_BMI_AUTO_EIP_CATALOG_ITEM", "ci-bm-auto-eip")
+
+
+@pytest.fixture(scope="session")
+def mgmt_cluster_ip() -> str:
+    return env("OSAC_MGMT_CLUSTER_IP", "192.168.40.2")
+
+
+@pytest.fixture(scope="session")
+def net_test_run_id() -> str:
+    return uuid.uuid4().hex[:8]
+
+
+@pytest.fixture(scope="session")
+def bmi_template() -> str:
+    return env("OSAC_BMI_TEMPLATE", "bm-host-provisioning")
+
+
+@pytest.fixture(scope="session")
+def bmh_namespace() -> str:
+    return env("OSAC_BMH_NAMESPACE", "host-inventory")
+
+
+@pytest.fixture(scope="session")
+def catalog_item_name() -> str:
+    return env("OSAC_BMI_CATALOG_ITEM", "ci-bm-default")
+
+
+@pytest.fixture(scope="session")
+def net_ssh_public_key() -> str:
+    key_path = Path(env("OSAC_BMI_SSH_PUBLIC_KEY", "/root/.ssh/id_rsa.pub"))
+    return key_path.read_text().strip()