Testing and CI Infrastructure
The testing infrastructure for dapcli prioritizes strict hermeticity and static hygiene to ensure code quality and security. By default, the test suite blocks all network connections and subprocess invocations, allowing only specific, read-only git operations required for repository introspection. This approach forces developers to explicitly mock external dependencies, preventing flaky tests and accidental data leakage. Additionally, automated checks enforce a hard limit on source file size to maintain readability and ease of onboarding.
Hermetic Test Environment
Section titled “Hermetic Test Environment”The core of the testing strategy is defined in tests/conftest.py, which provides two autouse session-scope fixtures that enforce a hermetic environment for every test 1.
- Network Blocking: The
_block_socket_connectfixture patchessocket.socket.connectto raise apytest.failerror if any in-process TCP connection is attempted . This catches pure-Python code that forgets to mock HTTP clients. Tests must explicitly patch libraries likehttpx.Clientto simulate network responses. - Subprocess Blocking: The
_block_subprocessfixture patchessubprocess.run,Popen, andcheck_outputto block all subprocess invocations except for specific allowed cases.- Git Carveout:
gitcommands are allowed because the hygiene scanner relies ongit ls-filesto enumerate tracked files, and self-tests may initialize temporary repositories. Thesegitoperations are read-only (e.g.,ls-files,init,add,commit) and never hit the network. - Ceiling Checker Carveout: A specific Python invocation of
tools/check_file_ceiling.pyis allowed, as this offline checker shells out only togit ls-files. - Failures: Any other subprocess invocation raises
pytest.fail.
- Git Carveout:
Tests that exercise the runner-proxy (dap.kube.runner) opt into mocked subprocess by patching dap.kube.runner.subprocess directly within the test function.
Blueprint Client API Tests
Section titled “Blueprint Client API Tests”The file tests/test_blueprint_client.py contains hermetic tests for the dap.blueprint.client module, locking in the API contract with the orchestrator’s /api/v3.1/* surface 2.
- Mocking Strategy: Tests use
httpx.MockTransportto intercept HTTP requests. A helper_mk_transporthandles OIDC token minting by returning a mock token for/api/v1/oidc/tokenrequests. - Patch Factory: A
_patch_factoryfunction createshttpx.Clientinstances with the mock transport, patching bothdap.auth.token.httpx.Clientanddap.orch_client.httpx.Client. - Sleep Patching: The
_no_sleephelper patchesdap.blueprint.client.time.sleepto a no-op, preventing tests from waiting during polling operations.
Key test cases verify:
- Tar Creation:
_tar_blueprintroots the archive at the source directory name and rejects missing directories. - Upload:
upload_blueprintsends a multipart/form-data request withparamsandblueprint_archivefields in the correct order. It raisesDapErrorif the upload state fails. - List/Delete:
list_blueprintsreturns items from the JSON response.delete_blueprintincludes theforce=truequery parameter when specified. - Deployment:
create_deploymentsendsblueprint_id,inputs, andvisibilityin the PUT body. - Workflow Execution:
execute_workflowposts a body withforce=Falseby default.wait_executionreturns the status or raisesDapErroron failure.latest_executionreturns the first item orNone. - Runtime Properties:
fetch_runtime_propertiesreturns node data.redact_runtime_propertiesmasks secret keys (e.g.,bmc_password,client_secret,auth_token,private_key,api_key) while preserving non-secret data.
Hygiene and File Size Checks
Section titled “Hygiene and File Size Checks”The repository enforces strict hygiene through automated tests that check file sizes and line counts.
- Source File Limit:
tests/test_file_size_limit.pyenforces a hard cap of 1000 lines for any first-party source file insrc/3. This is intended to keep modules readable in one screen and encourage splitting large modules early. The test iterates over all.pyfiles insrc/and fails if any exceed the limit. - Vendored Ceiling Checker:
tests/test_file_ceiling.pyruns the in-repo, stdlib-onlytools/check_file_ceiling.pyas a subprocess 4. This checker usesgit ls-filesto enumerate the tracked file set and fails if any hand-authored code file exceeds the 1000-line ceiling. This test is allowed to run as a subprocess because it is explicitly whitelisted in the hermetic fixtures 1.
"""Hermetic-test fixtures.
Two autouse session-scope fixtures fire for every test:
1. **Network block.** Patches ``socket.socket.connect`` so any
in-process attempt to open a TCP connection raises ``pytest.fail``.
Catches pure-Python code that forgot a mock.
2. **Subprocess block with `git` carveout.** Patches
``subprocess.run`` / ``Popen`` / ``check_output`` so that the only
subprocess invocations allowed during tests are ``git`` calls.
The hygiene scanner relies on ``git ls-files`` against the repo
root to enumerate the tracked file set, and the self-test file
legitimately initialises throwaway repos under ``tmp_path`` to
prove the scanner works. ``git`` is read-only for the operations
in use (``ls-files``, ``init``, ``add``, ``commit``) and never
hits the network, so allowing the binary by name is the cleanest
line to draw. The vendored file-size ceiling test
(``tests/test_file_ceiling.py``) runs the in-repo, stdlib-only
``tools/check_file_ceiling.py`` as a subprocess; that checker is
offline (it shells out only to ``git ls-files``), so a Python
invocation of that specific in-repo script is allowed too. Any
other non-``git`` subprocess raises ``pytest.fail``.
The hermeticity claim is therefore:
> no in-process socket connections, and no subprocess invocations
> other than ``git`` operations against test-provided temporary
> directories.
Tests that exercise the runner-proxy (``dap.kube.runner``) opt in to
mocked subprocess by patching ``dap.kube.runner.subprocess`` directly
inside the test function.
"""
from __future__ import annotations
import socket
import subprocess
import sys
from pathlib import Path
"""Hermetic tests for :mod:`dap.blueprint.client`.
These lock in the load-bearing API contract on the orchestrator's
`/api/v3.1/*` surface, every detail of which took an evening of
probing against a live cluster to discover.
"""
from __future__ import annotations
import json
import tarfile
from pathlib import Path
from unittest.mock import patch
import httpx
import pytest
from dap.blueprint.client import (
_tar_blueprint,
create_deployment,
delete_blueprint,
execute_workflow,
fetch_runtime_properties,
latest_execution,
list_blueprints,
redact_runtime_properties,
upload_blueprint,
wait_execution,
)
from dap.bootstrap import BootstrapResult
from dap.config import ClusterConfig
CFG = ClusterConfig(
name="ut",
portal_url="https://dap-portal.example.com",
orchestrator_url="https://dap-orchestrator.example.com",
realm="00000000-0000-0000-0000-000000000001",
org_id="00000000-0000-0000-0000-000000000001",
)
ADMIN = BootstrapResult(
"""Hard cap: no first-party source file may exceed 1000 lines.
Large command modules are split into packages instead. Splitting a
growing module early is cheaper than untangling a 2000-line "utils.py"
later, and keeping every module readable in one screen is a force-
multiplier when onboarding new contributors.
"""
from __future__ import annotations
from pathlib import Path
MAX_LINES = 1000
_SRC = Path(__file__).resolve().parent.parent / "src"
def test_no_source_file_exceeds_1000_lines() -> None:
offenders: list[str] = []
for path in sorted(_SRC.rglob("*.py")):
rel = str(path.relative_to(_SRC.parent))
with path.open("rb") as fh:
n = sum(1 for _ in fh)
if n > MAX_LINES:
offenders.append(f"{rel}: {n} lines")
assert not offenders, (
f"file(s) exceed the {MAX_LINES}-line hard limit. Split into a package/modules:\n " + "\n ".join(offenders)
)
"""Enforce the house 1000-line ceiling in this repo's own test suite.
Fails if any hand-authored code file exceeds the ceiling, so CI blocks a
regression with no allowlist or grandfathering. See
``tools/check_file_ceiling.py`` for the underlying check.
"""
from __future__ import annotations
import subprocess
import sys
from pathlib import Path
def _repo_root() -> Path:
res = subprocess.run(
["git", "rev-parse", "--show-toplevel"],
cwd=str(Path(__file__).resolve().parent),
capture_output=True,
text=True,
check=True,
)
return Path(res.stdout.strip())
def test_no_code_file_over_the_ceiling() -> None:
root = _repo_root()
checker = root / "tools" / "check_file_ceiling.py"
assert checker.exists(), f"vendored ceiling checker missing: {checker}"
proc = subprocess.run(
[sys.executable, str(checker), str(root)],
capture_output=True,
text=True,
check=False,
)
assert proc.returncode == 0, proc.stderr