plane(1.1): PlaneProfile -- one TOML per cluster, env-overridable, no secrets

ops/k8s/profiles/kind.toml and k3s.toml hold the values that differ per cluster
(exec, kube_context, registry, images, router_url, ax_server, memory, atespace,
snapshots_bucket, image_pull_secret, pools); kind-images.json carries the digest
pin ops/ax/up.sh wrote. load_profile() reads OUSAST_PLANE_PROFILE (name or path,
default kind), applies OUSAST_<FIELD> and OUSAST_POOL_<NAME>_<FIELD>, and refuses
an unknown key, a credential-shaped key or value, an exec outside local|remote
and an images file without digest pins. `python -m openultrasast.plane.profile
--print FIELD` serves shell scripts. The module-audit manifest is regenerated
for the new module (counts 147/5, no orphan).

Gates: pytest full suite 2002 passed, 1 failed (the module-audit manifest, fixed
by regenerating it; tests/test_module_audit.py re-run exit 0), ruff check 0,
ruff format 0, mypy 0.

Co-Authored-By: Claude Fable 5.1 <[email protected]>
Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo
This commit is contained in:
Marius Ciepluch
2026-10-02 20:16:23 +02:00
co-authored by Claude Fable 5.1
parent 6bd44caa50
commit 8eac1b2c69
7 changed files with 2250 additions and 1767 deletions
+1 -1
View File
@@ -10,7 +10,7 @@ its kind proof and the record it writes under `benchmarks/measurements/<date>-k8
go-ahead** with a ceiling; steps that need a hand on a cluster that does not exist yet are marked **manual**.
- [ ] 1. Profile and literals (design section 1; open-question checks 2, 3, 7)
- [ ] 1.1 `PlaneProfile`
- [x] 1.1 `PlaneProfile`
- `src/openultrasast/plane/profile.py`: frozen dataclass with the fields of design section 1; `load_profile()`
reads `ops/k8s/profiles/<OUSAST_PLANE_PROFILE>.toml` (a name or a path; default `kind`), then applies
`OUSAST_<FIELD>` overrides; rejects an unknown key, a secret-looking key (`*_KEY`, `*_SECRET`), an `exec`
File diff suppressed because it is too large Load Diff
+26
View File
@@ -0,0 +1,26 @@
# The k3s profile: the plane on the maintainer's production cluster (plane-on-kubernetes design section 1;
# requirements "Target Environment"). Documented values, not secrets: the kubeconfig for the context, the GHCR
# pull secret's content and the store's keys live outside this file. Every field is overridable by OUSAST_<FIELD>,
# a pool by OUSAST_POOL_<NAME>_<FIELD>. Nothing has been applied to a k3s cluster yet; docs/deployment.md says
# which items are planned.
exec = "remote" # the same CLI from a laptop or a CI job, through the cluster's kube context
kube_context = "ousast-k3s" # the maintainer's context name for the k3s cluster
registry = "ghcr.io/norandom" # images built by GitHub Actions from tagged commits (design section 8)
images = "images.json" # the release asset {"runner": "ghcr.io/norandom/ousast-runner@sha256:...", "engine": ...}
router_url = "" # empty: port-forward to atenet-router through the kube context (ax-tunnel)
ax_server = "" # empty: the ax CLI's own tunnel through the kube context (ax-tunnel)
memory = "s3://sast-memory" # the memory store; S3_ENDPOINT and the keys come from the CI secrets or .env
atespace = "default"
snapshots_bucket = "s3://ax-snapshots" # ax's AX_SNAPSHOTS_BUCKET, ours (rendered into the ax-server patch)
image_pull_secret = "ghcr-pull" # only needed while the GHCR packages are private
[pools.default] # production sizing: four verify-class workers; the HPA (design section 5) scales it
replicas = 4
cpu = "1"
memory = "1536Mi"
[pools.engine] # two engine workers sized from the measured 2.1 GiB peak under a 3 GiB limit
replicas = 2
cpu = "2"
memory = "4Gi"
+3
View File
@@ -0,0 +1,3 @@
{
"runner": "localhost:5001/ousast-runner@sha256:412db0a1504af8f1b23a395af8d1dcbd152c1bb6d16594f2b733067163a31884"
}
+25
View File
@@ -0,0 +1,25 @@
# The kind profile: the plane on this laptop's single-node kind cluster, which ops/ax/up.sh creates and which is
# the development default (plane-on-kubernetes design section 1). Values only; a credential never goes here
# (keys stay in the environment or .env). Every field is overridable by OUSAST_<FIELD> in the environment, a pool
# by OUSAST_POOL_<NAME>_<FIELD>. This file may name the kind cluster; the code must not (tests/test_plane_literals.py).
exec = "local" # the reconciler runs on this host against the cluster below
kube_context = "kind-ousast" # kind's context for KIND_CLUSTER_NAME=ousast
registry = "localhost:5001" # the kind-local registry up.sh creates (Substrate rewrites it for the node)
images = "kind-images.json" # digest-pinned image references, written by up.sh after each build
router_url = "" # empty: port-forward to atenet-router through the kube context (ax-tunnel)
ax_server = "" # empty: the ax CLI's own tunnel through the kube context (ax-tunnel)
memory = "s3://sast-memory" # the memory store; S3_ENDPOINT and the keys come from .env
atespace = "default"
snapshots_bucket = "s3://ax-snapshots" # ax's AX_SNAPSHOTS_BUCKET, ours (rendered into the ax-server patch)
image_pull_secret = "" # the kind registry needs none
[pools.default] # ousast-pool: two gVisor workers fit the 7 GB host beside the control planes
replicas = 2
cpu = "1"
memory = "1536Mi"
[pools.engine] # engine-pool (design section 4): one worker for the Joern engine task
replicas = 1
cpu = "2"
memory = "4Gi"
+266
View File
@@ -0,0 +1,266 @@
"""The plane's configuration surface: one ``PlaneProfile`` per cluster (plane-on-kubernetes design section 1).
A profile is a TOML file under ``ops/k8s/profiles/`` (``kind.toml`` for the laptop's kind cluster, ``k3s.toml`` for
the production cluster) holding every value that differs between clusters: the execution mode, the kube context,
the registry and the digest-pinned images file, the router and ax-server addresses (empty: tunnel through the
Kubernetes API), the memory store, the atespace, ax's snapshot bucket, the worker pools and the image pull secret.
:func:`load_profile` reads the file ``OUSAST_PLANE_PROFILE`` names (a profile name or a path; default ``kind``) and
then applies the environment: every scalar field is overridable by ``OUSAST_<FIELD>`` (``OUSAST_KUBE_CONTEXT``,
``OUSAST_REGISTRY``, ``OUSAST_MEMORY``, ``OUSAST_ROUTER_URL``, ...), a pool by ``OUSAST_POOL_<NAME>_<FIELD>``.
A profile holds no secret: a key whose name ends in ``key``, ``secret``, ``token`` or ``password`` is refused, and so
is any unknown key, an ``exec`` outside ``local``/``remote``, and an images file whose references are not pinned by
digest (Agent Substrate rejects tags). Provider and store credentials stay in the environment or ``.env``.
``python -m openultrasast.plane.profile --print FIELD`` prints one value for shell scripts (``ops/ax/up.sh``):
``registry``, ``kube_context``, ``images.runner``, ``pools.default.replicas``, ...
"""
from __future__ import annotations
import json
import os
import re
import sys
import tomllib
from collections.abc import Mapping
from dataclasses import dataclass, field, fields
from pathlib import Path
from typing import Any
__all__ = ["DEFAULT_PROFILE", "PROFILES_DIR", "PlaneProfile", "Pool", "ProfileError", "load_images", "load_profile", "profiles_dir"]
DEFAULT_PROFILE = "kind"
PROFILES_DIR = Path("ops/k8s/profiles")
EXEC_MODES = ("local", "remote")
_DIGEST = re.compile(r"\S+@sha256:[0-9a-f]{64}")
_SECRET_KEY = re.compile(r"(key|secret|token|password)$", re.IGNORECASE)
_SECRET_VALUE = re.compile(r"(sk-[A-Za-z0-9]{8,}|AKIA[0-9A-Z]{12,}|[A-Za-z0-9+/]{40,}={0,2}$)")
_ENV_PREFIX = "OUSAST_"
class ProfileError(ValueError):
"""A profile that cannot be used; the message names the file and the field."""
@dataclass(frozen=True)
class Pool:
"""One WorkerPool's size: how many warm workers, and each worker's CPU and memory limit."""
replicas: int
cpu: str
memory: str
@dataclass(frozen=True)
class PlaneProfile:
"""The values of one cluster. ``images`` is resolved against the profile file's directory."""
name: str
path: Path
exec: str = "local"
kube_context: str = ""
registry: str = ""
images: Path = Path("images.json")
router_url: str = ""
ax_server: str = ""
memory: str = ""
atespace: str = "default"
snapshots_bucket: str = ""
image_pull_secret: str = ""
pools: Mapping[str, Pool] = field(default_factory=dict)
def load_images(self) -> dict[str, str]:
"""The digest-pinned image references of ``images`` (``{"runner": ..., "engine": ...}``)."""
return load_images(self.images)
def addresses(self) -> dict[str, str]:
"""Every configured address, for ``ousast plane doctor`` (Req 1.2); an empty router or ax-server address
means the Kubernetes API tunnel."""
return {
"exec": self.exec,
"kube_context": self.kube_context,
"registry": self.registry,
"images": str(self.images),
"router_url": self.router_url or "(tunnel through the kube context)",
"ax_server": self.ax_server or "(tunnel through the kube context)",
"memory": self.memory,
"atespace": self.atespace,
"snapshots_bucket": self.snapshots_bucket,
"image_pull_secret": self.image_pull_secret or "(none)",
}
_SCALARS = {f.name: str(f.type) for f in fields(PlaneProfile) if f.name not in ("name", "path", "pools")}
_POOL_FIELDS = {f.name: str(f.type) for f in fields(Pool)}
def profiles_dir() -> Path:
"""``ops/k8s/profiles`` of this checkout (``OUSAST_PLANE_PROFILES`` overrides; an installed package falls back
to the working directory)."""
override = os.environ.get("OUSAST_PLANE_PROFILES")
if override:
return Path(override)
checkout = Path(__file__).resolve().parents[3] / PROFILES_DIR
return checkout if checkout.is_dir() else PROFILES_DIR
def load_images(path: Path) -> dict[str, str]:
"""``{"runner": "<registry>/<name>@sha256:<64 hex>", ...}`` from a JSON images file; every value must be
digest-pinned. A one-line file holding a single reference (``ops/ax/up.sh`` before this profile) reads as
``{"runner": <line>}``."""
try:
text = path.read_text(encoding="utf-8").strip()
except OSError as exc:
raise ProfileError(f"images file {path}: {exc.strerror or exc}") from None
try:
loaded: Any = json.loads(text) if text.startswith("{") else {"runner": text}
except ValueError as exc:
raise ProfileError(f"images file {path}: not JSON ({exc})") from None
if not isinstance(loaded, dict) or not loaded or not all(isinstance(k, str) and isinstance(v, str) for k, v in loaded.items()):
raise ProfileError(f"images file {path}: expected an object of image names to references")
for name, reference in loaded.items():
if not _DIGEST.fullmatch(reference):
raise ProfileError(f"images file {path}: {name} is not digest-pinned ({reference!r}); Agent Substrate rejects tags")
return dict(loaded)
def _reject_secret(where: str, key: str, value: object, declared: bool = False) -> None:
"""A key that is not a declared field and is named like a credential, or any value shaped like one, is refused.
``image_pull_secret`` is declared: its value is the *name* of a Kubernetes Secret, never its content."""
if not declared and _SECRET_KEY.search(key):
raise ProfileError(f"{where}: {key} looks like a credential; profiles hold no secret (use the environment or .env)")
if isinstance(value, str) and _SECRET_VALUE.search(value):
raise ProfileError(f"{where}: the value of {key} looks like a credential; profiles hold no secret")
def _coerce(where: str, key: str, value: object, kind: str) -> Any:
if kind == "int":
if isinstance(value, bool) or not isinstance(value, int):
raise ProfileError(f"{where}: {key} must be an integer, got {value!r}")
return value
if kind == "Path":
if not isinstance(value, str) or not value:
raise ProfileError(f"{where}: {key} must be a file name")
return Path(value)
if not isinstance(value, str):
raise ProfileError(f"{where}: {key} must be a string, got {value!r}")
return value
def _pools(where: str, raw: object) -> dict[str, Pool]:
if not isinstance(raw, dict):
raise ProfileError(f"{where}: pools must be a table of pool names")
out: dict[str, Pool] = {}
for name, spec in raw.items():
if not isinstance(spec, dict):
raise ProfileError(f"{where}: pools.{name} must be a table")
unknown = sorted(set(spec) - set(_POOL_FIELDS))
if unknown:
raise ProfileError(f"{where}: pools.{name} has unknown keys: {', '.join(unknown)}")
missing = sorted(set(_POOL_FIELDS) - set(spec))
if missing:
raise ProfileError(f"{where}: pools.{name} lacks {', '.join(missing)}")
for key, value in spec.items():
_reject_secret(where, f"pools.{name}.{key}", value, declared=True)
out[str(name)] = Pool(**{k: _coerce(where, f"pools.{name}.{k}", spec[k], _POOL_FIELDS[k]) for k in _POOL_FIELDS})
return out
def _env_overrides(environ: Mapping[str, str], where: str, values: dict[str, Any], pools: dict[str, Pool]) -> None:
for name, kind in _SCALARS.items():
raw = environ.get(_ENV_PREFIX + name.upper())
if raw is not None and raw != "":
value: Any = raw
if kind == "int":
try:
value = int(raw)
except ValueError:
raise ProfileError(f"{where}: {_ENV_PREFIX}{name.upper()} must be an integer, got {raw!r}") from None
values[name] = _coerce(where, _ENV_PREFIX + name.upper(), value, kind)
pattern = re.compile(rf"^{_ENV_PREFIX}POOL_([A-Z0-9]+)_(REPLICAS|CPU|MEMORY)$")
for variable, raw in environ.items():
found = pattern.match(variable)
if not found or raw == "":
continue
pool_name, key = found.group(1).lower(), found.group(2).lower()
current = pools.get(pool_name)
if current is None:
raise ProfileError(f"{where}: {variable} names a pool the profile does not declare ({pool_name})")
value = int(raw) if key == "replicas" and raw.isdigit() else raw
pools[pool_name] = Pool(**{**vars(current), key: _coerce(where, variable, value, _POOL_FIELDS[key])})
def load_profile(name: str | None = None, environ: Mapping[str, str] | None = None) -> PlaneProfile:
"""The profile ``name`` (a name under :func:`profiles_dir` or a path to a ``.toml``; default
``OUSAST_PLANE_PROFILE``, else ``kind``), with the environment's overrides applied; :class:`ProfileError`
names what is wrong."""
env = os.environ if environ is None else environ
chosen = name or env.get("OUSAST_PLANE_PROFILE") or DEFAULT_PROFILE
path = Path(chosen) if chosen.endswith(".toml") or "/" in chosen else profiles_dir() / f"{chosen}.toml"
where = str(path)
try:
raw = tomllib.loads(path.read_text(encoding="utf-8"))
except OSError as exc:
raise ProfileError(f"profile {where}: {exc.strerror or exc}") from None
except tomllib.TOMLDecodeError as exc:
raise ProfileError(f"profile {where}: {exc}") from None
for key, value in raw.items(): # a credential-shaped key is named as such before it is merely unknown
_reject_secret(where, key, value, declared=key in _SCALARS)
unknown = sorted(set(raw) - set(_SCALARS) - {"pools"})
if unknown:
raise ProfileError(f"{where}: unknown keys: {', '.join(unknown)}")
values: dict[str, Any] = {}
for key, value in raw.items():
if key != "pools":
values[key] = _coerce(where, key, value, _SCALARS[key])
pools = _pools(where, raw.get("pools", {}))
_env_overrides(env, where, values, pools)
if values.get("exec", "local") not in EXEC_MODES:
raise ProfileError(f"{where}: exec must be one of {', '.join(EXEC_MODES)}, got {values.get('exec')!r}")
images = Path(values.get("images", "images.json"))
if not images.is_absolute():
images = path.resolve().parent / images
values["images"] = images
if images.is_file():
load_images(images) # a committed file must be pinned; a file the release has not published yet is doctor's finding
return PlaneProfile(name=path.stem, path=path, pools=pools, **values)
def _lookup(profile: PlaneProfile, dotted: str) -> str:
head, _, rest = dotted.partition(".")
if head == "images":
images = profile.load_images()
if rest not in images:
raise ProfileError(f"{profile.images}: no image named {rest!r} (have {', '.join(sorted(images))})")
return images[rest]
if head == "pools":
pool_name, _, key = rest.partition(".")
if pool_name not in profile.pools or key not in _POOL_FIELDS:
raise ProfileError(f"{profile.path}: no such pool field {dotted!r}")
return str(getattr(profile.pools[pool_name], key))
if head not in _SCALARS or rest:
raise ProfileError(f"{profile.path}: no such field {dotted!r}")
return str(getattr(profile, head))
def main(argv: list[str] | None = None) -> int:
"""``python -m openultrasast.plane.profile [--profile NAME] --print FIELD [FIELD...]``: one value per line."""
import argparse
parser = argparse.ArgumentParser(prog="openultrasast.plane.profile", description=main.__doc__)
parser.add_argument("--profile", help="profile name or path (default OUSAST_PLANE_PROFILE, else kind)")
parser.add_argument("--print", dest="fields", nargs="+", required=True, metavar="FIELD")
args = parser.parse_args(argv)
try:
profile = load_profile(args.profile)
for dotted in args.fields:
print(_lookup(profile, dotted))
except ProfileError as exc:
print(f"profile: {exc}", file=sys.stderr)
return 2
return 0
if __name__ == "__main__":
sys.exit(main())
+156
View File
@@ -0,0 +1,156 @@
"""plane-on-kubernetes task 1.1: the ``PlaneProfile`` (Req 1.1). Defaults, file-then-environment precedence, every
rejection naming its field, both committed profiles loading, and no committed value that looks like a credential."""
from __future__ import annotations
import json
import re
from pathlib import Path
import pytest
from openultrasast.plane import profile as profile_module
from openultrasast.plane.profile import PlaneProfile, Pool, ProfileError, load_images, load_profile, profiles_dir
ROOT = Path(__file__).resolve().parents[1]
DIGEST = "sha256:" + "ab" * 32
MINIMAL = 'exec = "local"\nkube_context = "kind-test"\nregistry = "registry.test:5000"\nimages = "images.json"\nmemory = "s3://m"\n'
def write_profile(tmp_path: Path, text: str = MINIMAL, name: str = "test") -> Path:
path = tmp_path / f"{name}.toml"
path.write_text(text, encoding="utf-8")
(tmp_path / "images.json").write_text(json.dumps({"runner": f"registry.test:5000/ousast-runner@{DIGEST}"}), encoding="utf-8")
return path
def test_defaults_and_the_images_path_resolve_against_the_profile_file(tmp_path: Path) -> None:
loaded = load_profile(str(write_profile(tmp_path)), environ={})
assert loaded.name == "test" and loaded.exec == "local" and loaded.kube_context == "kind-test"
assert loaded.images == tmp_path / "images.json" and loaded.load_images() == {"runner": f"registry.test:5000/ousast-runner@{DIGEST}"}
assert loaded.atespace == "default" and loaded.router_url == "" and loaded.pools == {}
assert "(tunnel through the kube context)" in loaded.addresses()["router_url"]
def test_the_environment_overrides_the_file_and_names_the_profile(tmp_path: Path) -> None:
path = write_profile(tmp_path, MINIMAL + '[pools.default]\nreplicas = 2\ncpu = "1"\nmemory = "1536Mi"\n')
env = {
"OUSAST_PLANE_PROFILE": str(path),
"OUSAST_KUBE_CONTEXT": "other-cluster",
"OUSAST_MEMORY": "s3://elsewhere/prefix",
"OUSAST_ROUTER_URL": "https://router.test",
"OUSAST_POOL_DEFAULT_REPLICAS": "5",
"OUSAST_POOL_DEFAULT_MEMORY": "2Gi",
"OUSAST_EXEC": "", # an empty variable is not an override
}
loaded = load_profile(environ=env)
assert (loaded.kube_context, loaded.memory, loaded.router_url, loaded.exec) == (
"other-cluster",
"s3://elsewhere/prefix",
"https://router.test",
"local",
)
assert loaded.pools == {"default": Pool(replicas=5, cpu="1", memory="2Gi")}
assert load_profile(str(path), environ={}).kube_context == "kind-test", "the file is unchanged"
@pytest.mark.parametrize(
("text", "field"),
[
(MINIMAL + 'colour = "blue"\n', "unknown keys: colour"),
(MINIMAL + 'aws_secret_key = "x"\n', "aws_secret_key looks like a credential"),
(MINIMAL + 'api_token = "x"\n', "api_token looks like a credential"),
(MINIMAL.replace('exec = "local"', 'exec = "cluster"'), "exec must be one of local, remote"),
(MINIMAL + '[pools.default]\nreplicas = "two"\ncpu = "1"\nmemory = "1Gi"\n', "pools.default.replicas must be an integer"),
(MINIMAL + '[pools.default]\nreplicas = 1\ncpu = "1"\n', "pools.default lacks memory"),
(MINIMAL + '[pools.default]\nreplicas = 1\ncpu = "1"\nmemory = "1Gi"\nnodes = 3\n', "pools.default has unknown keys: nodes"),
(MINIMAL.replace('kube_context = "kind-test"', "kube_context = 3"), "kube_context must be a string"),
],
)
def test_each_rejection_names_the_field(tmp_path: Path, text: str, field: str) -> None:
path = write_profile(tmp_path, text)
with pytest.raises(ProfileError, match=re.escape(field)) as exc:
load_profile(str(path), environ={})
assert str(path) in str(exc.value)
def test_a_credential_shaped_value_is_refused(tmp_path: Path) -> None:
path = write_profile(tmp_path, MINIMAL.replace("s3://m", "sk-" + "a1b2c3d4" * 3))
with pytest.raises(ProfileError, match="value of memory looks like a credential"):
load_profile(str(path), environ={})
def test_an_images_file_without_digest_pins_is_refused(tmp_path: Path) -> None:
path = write_profile(tmp_path)
(tmp_path / "images.json").write_text(json.dumps({"runner": "registry.test:5000/ousast-runner:dev"}), encoding="utf-8")
with pytest.raises(ProfileError, match="runner is not digest-pinned"):
load_profile(str(path), environ={})
(tmp_path / "images.json").write_text(f"registry.test:5000/ousast-runner@{DIGEST}\n", encoding="utf-8")
assert load_images(tmp_path / "images.json") == {"runner": f"registry.test:5000/ousast-runner@{DIGEST}"}, "the one-line form"
(tmp_path / "images.json").unlink()
loaded = load_profile(str(path), environ={}) # a file the release has not published yet: doctor's finding, not a load error
with pytest.raises(ProfileError, match="images file"):
loaded.load_images()
def test_an_environment_override_of_the_wrong_shape_is_refused(tmp_path: Path) -> None:
path = write_profile(tmp_path, MINIMAL + '[pools.default]\nreplicas = 1\ncpu = "1"\nmemory = "1Gi"\n')
with pytest.raises(ProfileError, match="OUSAST_POOL_DEFAULT_REPLICAS must be an integer"):
load_profile(str(path), environ={"OUSAST_POOL_DEFAULT_REPLICAS": "many"})
with pytest.raises(ProfileError, match="names a pool the profile does not declare"):
load_profile(str(path), environ={"OUSAST_POOL_ENGINE_REPLICAS": "1"})
with pytest.raises(ProfileError, match="exec must be one of"):
load_profile(str(path), environ={"OUSAST_EXEC": "both"})
def test_a_missing_profile_names_its_path(tmp_path: Path) -> None:
with pytest.raises(ProfileError, match="no-such.toml"):
load_profile(str(tmp_path / "no-such.toml"), environ={})
with pytest.raises(ProfileError, match="unknown"):
load_profile("unknown", environ={"OUSAST_PLANE_PROFILES": str(tmp_path)})
@pytest.mark.parametrize("name", ["kind", "k3s"])
def test_both_committed_profiles_load_with_the_designs_values(name: str) -> None:
loaded = load_profile(name, environ={})
assert loaded.path == profiles_dir() / f"{name}.toml" and profiles_dir() == ROOT / "ops/k8s/profiles"
assert loaded.exec == {"kind": "local", "k3s": "remote"}[name]
assert loaded.memory == "s3://sast-memory" and loaded.atespace == "default" and loaded.snapshots_bucket == "s3://ax-snapshots"
assert set(loaded.pools) == {"default", "engine"} and loaded.pools["engine"].memory == "4Gi"
assert loaded.router_url == "" and loaded.ax_server == "", "both profiles tunnel through the Kubernetes API"
if name == "kind":
assert loaded.image_pull_secret == "" and loaded.images.name == "kind-images.json" and loaded.images.is_file()
assert set(loaded.load_images()) == {"runner"} and loaded.load_images()["runner"].startswith(loaded.registry + "/")
else:
assert loaded.image_pull_secret == "ghcr-pull" and loaded.registry.startswith("ghcr.io/") and loaded.images.name == "images.json"
assert not loaded.images.is_file(), "the release asset is downloaded, never committed (design section 8)"
def test_no_committed_profile_value_matches_a_credential_pattern() -> None:
patterns = (
re.compile(r"(?i)(api_key|secret_key|access_key|token|password)\s*="), # image_pull_secret names a Secret, holds none
re.compile(r"sk-[A-Za-z0-9]{8,}"),
re.compile(r"AKIA[0-9A-Z]{12,}"),
)
for path in sorted(profiles_dir().glob("*.toml")):
for number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
code = line.split("#", 1)[0]
for pattern in patterns:
assert not pattern.search(code), f"{path}:{number}: {line.strip()}"
def test_print_for_shell_scripts(capsys: pytest.CaptureFixture[str], tmp_path: Path) -> None:
path = write_profile(tmp_path, MINIMAL + '[pools.default]\nreplicas = 3\ncpu = "1"\nmemory = "1Gi"\n')
fields = ["registry", "kube_context", "images.runner", "pools.default.replicas"]
assert profile_module.main(["--profile", str(path), "--print", *fields]) == 0
assert capsys.readouterr().out.splitlines() == ["registry.test:5000", "kind-test", f"registry.test:5000/ousast-runner@{DIGEST}", "3"]
assert profile_module.main(["--profile", str(path), "--print", "images.engine"]) == 2
assert "no image named 'engine'" in capsys.readouterr().err
assert profile_module.main(["--profile", str(path), "--print", "nothing"]) == 2
def test_profile_is_frozen_and_typed() -> None:
loaded = load_profile("kind", environ={})
assert isinstance(loaded, PlaneProfile)
with pytest.raises(AttributeError):
loaded.exec = "remote" # type: ignore[misc]