merge(milestone): v1.29 Reposplit + Identity Layer Bring-Live to main (release v1.28.6)
Nova Slides Render / render (push) Failing after 22s
Nova Slides Render / render (push) Failing after 22s
---ci--- project: acdl phase: 6 milestone: v1.29 status: complete ---/ci---
This commit is contained in:
@@ -0,0 +1,457 @@
|
||||
"""ABAC end-to-end test for the token-vend Lambda (Edge 5 item 7, INV-17).
|
||||
|
||||
The M1.5 verification-gate spike (PLAN.md Happy Path §3.3 Edge 5 item 7):
|
||||
|
||||
Known PAT → known ABAC-allowed action → signed OIDC token → jose/pyjwt
|
||||
verification → green. Known PAT + ABAC-denied action → 403 with deny
|
||||
reason logged (INV-17 fail-closed).
|
||||
|
||||
This is the end-to-end ABAC path: PAT → revocation check (D-229 strong
|
||||
read) → kyverno-json ABAC policy evaluation → KMS-signed OIDC token →
|
||||
JWKS fetch → pyjwt signature verification. It wires the **real**
|
||||
``core.abac_evaluator.evaluate_token_vend_policy`` (which shells to the
|
||||
``kj`` binary against ``platform/abac/token-vend.policy``) behind the
|
||||
token-vend Lambda handler, then verifies the vended OIDC token against
|
||||
the JWKS the JWKS Lambda would serve — exactly the M1.5 spike shape.
|
||||
|
||||
## Two execution surfaces (REQ-362 covered-reference)
|
||||
|
||||
* **acdl CI** — ``kj`` is NOT installed (``which kj`` is absent) and
|
||||
there is no live KMS key. The ABAC-allowed and ABAC-denied tests
|
||||
therefore ``pytest.skip`` with a clear reason (the ``kj`` binary is a
|
||||
build-host/nova-platform-ops dep). The fail-closed (policy-absent)
|
||||
test runs in acdl CI because it does NOT need ``kj`` — it exercises
|
||||
the ``is_configured()``-False → 403 ``abac_eval_failed`` path.
|
||||
* **nova-platform-ops CI** — ``kj`` is present at ``/opt/kj/kj`` and the
|
||||
live KMS key ``alias/nova-oidc-signing`` is reachable. The
|
||||
ABAC-allowed/denied tests run against the real binary + a mock KMS
|
||||
(or the live key when marked ``live_aws``).
|
||||
|
||||
## Test deps
|
||||
|
||||
* ``moto[dynamodb]`` — mocks ``nova-pats`` (revocation strong read).
|
||||
* mock KMS via ``cryptography`` generated ECDSA P-256 keypair (the same
|
||||
pattern as ``tests/test_kms_roundtrip.py`` + ``test_pat_revocation.py``).
|
||||
* ``pyjwt`` — verifies the vended OIDC token against the JWKS the JWKS
|
||||
Lambda serves (the ``jose``-equivalent verification in the plan; the
|
||||
repo standardizes on ``pyjwt`` + ``cryptography``, no ``jose`` dep).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
# moto requires a region; the Lambdas' lazy boto3.resource("dynamodb")
|
||||
# picks up AWS_DEFAULT_REGION.
|
||||
os.environ.setdefault("AWS_DEFAULT_REGION", "us-east-1")
|
||||
os.environ.setdefault("AWS_ACCESS_KEY_ID", "test")
|
||||
os.environ.setdefault("AWS_SECRET_ACCESS_KEY", "test")
|
||||
os.environ.setdefault("NOVA_LAMBDA_LOCAL_BYPASS", "1")
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Load the three IdP Lambda modules via importlib (`lambda` is a reserved
|
||||
# word — mirrors tests/test_idp_auth.py / test_pat_revocation.py).
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_TV_PATH = (
|
||||
Path(__file__).resolve().parent.parent / "core" / "lambda" / "nova_idp_token_vend.py"
|
||||
)
|
||||
_spec_tv = importlib.util.spec_from_file_location("nova_idp_token_vend_e2e", _TV_PATH)
|
||||
tv = importlib.util.module_from_spec(_spec_tv)
|
||||
_spec_tv.loader.exec_module(tv)
|
||||
|
||||
_JWKS_PATH = (
|
||||
Path(__file__).resolve().parent.parent / "core" / "lambda" / "nova_idp_jwks.py"
|
||||
)
|
||||
_spec_jwks = importlib.util.spec_from_file_location("nova_idp_jwks_e2e", _JWKS_PATH)
|
||||
jwks_mod = importlib.util.module_from_spec(_spec_jwks)
|
||||
_spec_jwks.loader.exec_module(jwks_mod)
|
||||
|
||||
import boto3
|
||||
from moto import mock_aws
|
||||
from cryptography.hazmat.primitives.asymmetric import ec
|
||||
from cryptography.hazmat.primitives import hashes, serialization
|
||||
|
||||
import core.kms_signing as kms_signing
|
||||
import core.pat_lifecycle as pat_life
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# kj availability — the ABAC-allowed/denied tests invoke the real kj
|
||||
# binary (nova-platform-ops CI installs it at /opt/kj/kj). In acdl CI kj
|
||||
# is absent, so those tests skip. The fail-closed (policy-absent) test
|
||||
# runs without kj (it asserts the is_configured()-False → 403 path).
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
KJ_AVAILABLE = shutil.which("kj") is not None
|
||||
skip_no_kj = pytest.mark.skipif(
|
||||
not KJ_AVAILABLE,
|
||||
reason="`kj` binary not on PATH (D-227 build-host dep; runs in "
|
||||
"nova-platform-ops CI against /opt/kj/kj)",
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Mock KMS (generated ECDSA P-256 keypair) — same pattern as
|
||||
# tests/test_kms_roundtrip.py and tests/test_pat_revocation.py.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class _MockKms:
|
||||
def __init__(self, priv, pub_der):
|
||||
self._priv = priv
|
||||
self._pub_der = pub_der
|
||||
|
||||
def sign(self, KeyId, Message, MessageType, SigningAlgorithm):
|
||||
return {"Signature": self._priv.sign(Message, ec.ECDSA(hashes.SHA256()))}
|
||||
|
||||
def get_public_key(self, KeyId):
|
||||
return {"PublicKey": self._pub_der}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# DynamoDB fixture — nova-pats (revocation strong read, D-229).
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _create_pats_table(ddb):
|
||||
ddb.create_table(
|
||||
TableName="nova-pats",
|
||||
KeySchema=[{"AttributeName": "jti", "KeyType": "HASH"}],
|
||||
AttributeDefinitions=[
|
||||
{"AttributeName": "jti", "AttributeType": "S"},
|
||||
{"AttributeName": "sub", "AttributeType": "S"},
|
||||
{"AttributeName": "pat_hash", "AttributeType": "S"},
|
||||
],
|
||||
GlobalSecondaryIndexes=[
|
||||
{
|
||||
"IndexName": "sub-index",
|
||||
"KeySchema": [{"AttributeName": "sub", "KeyType": "HASH"}],
|
||||
"Projection": {"ProjectionType": "ALL"},
|
||||
},
|
||||
{
|
||||
"IndexName": "pat_hash-index",
|
||||
"KeySchema": [{"AttributeName": "pat_hash", "KeyType": "HASH"}],
|
||||
"Projection": {"ProjectionType": "ALL"},
|
||||
},
|
||||
],
|
||||
BillingMode="PAY_PER_REQUEST",
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _reset_singletons():
|
||||
"""Reset module-level singletons + the test-injected KMS client
|
||||
before/after each test (mirrors test_pat_revocation.py)."""
|
||||
tv._dynamodb = None
|
||||
pat_life._dynamodb = None
|
||||
yield
|
||||
tv._dynamodb = None
|
||||
pat_life._dynamodb = None
|
||||
kms_signing.set_kms_client_for_testing(None)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_kms():
|
||||
"""Install a mock KMS client backed by a generated P-256 keypair."""
|
||||
priv = ec.generate_private_key(ec.SECP256R1())
|
||||
pub_der = priv.public_key().public_bytes(
|
||||
encoding=serialization.Encoding.DER,
|
||||
format=serialization.PublicFormat.SubjectPublicKeyInfo,
|
||||
)
|
||||
kms_signing.set_kms_client_for_testing(_MockKms(priv, pub_der))
|
||||
return priv
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def moto_pats():
|
||||
"""Spin up moto-backed DynamoDB with the nova-pats table."""
|
||||
with mock_aws():
|
||||
client = boto3.client("dynamodb", region_name="us-east-1")
|
||||
_create_pats_table(client)
|
||||
yield client
|
||||
|
||||
|
||||
def _issue_pat(sub="dev-alice", roles=None, owner="owner-alice"):
|
||||
"""Issue a real PAT (KMS-signed JWT, hash stored in nova-pats) for
|
||||
the ABAC-allowed scenario — subject.role='developer', owner matches
|
||||
the target resource owner."""
|
||||
roles = roles or ["developer"]
|
||||
return pat_life.issue_pat(sub, roles, owner, ttl_seconds=3600)
|
||||
|
||||
|
||||
def _vend_event(pat, **extra):
|
||||
"""Build a token-vend Lambda event. Defaults: environment='dev',
|
||||
target_resource owner inherits from the PAT (owner-matches rule
|
||||
passes for same-tenant vends), requested_claims non-empty."""
|
||||
body = {
|
||||
"token": pat,
|
||||
"environment": "dev",
|
||||
"target_resource": {
|
||||
"type": "contract",
|
||||
"id": "c-allowed",
|
||||
"owner": "owner-alice",
|
||||
"environment": "dev",
|
||||
},
|
||||
"requested_claims": ["sub", "roles"],
|
||||
}
|
||||
body.update(extra)
|
||||
return {"body": json.dumps(body)}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Edge 5 item 7a — ABAC-allowed path: known PAT → ABAC allow → signed
|
||||
# OIDC token → jose/pyjwt verification → green.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@skip_no_kj
|
||||
def test_abac_allowed_vend_then_verify_oidc(moto_pats, mock_kms, capsys):
|
||||
"""Edge 5 item 7 (allowed path):
|
||||
|
||||
subject.role='developer', environment='dev', target_resource.owner
|
||||
matches subject.owner, requested_claims non-empty → ABAC policy
|
||||
allows (all three rules pass: owner-matches, role-env-match,
|
||||
requested-claims-present) → token-vend KMS-signs an OIDC token →
|
||||
JWKS Lambda serves the public key → pyjwt verifies the signature.
|
||||
"""
|
||||
pat = _issue_pat(sub="dev-alice", owner="owner-alice")
|
||||
resp = tv.lambda_handler(_vend_event(pat), None)
|
||||
assert resp["statusCode"] == 200, resp
|
||||
body = json.loads(resp["body"])
|
||||
assert "token" in body, "no token vended (ABAC should allow this path)"
|
||||
oidc_token = body["token"]
|
||||
|
||||
# Verify the OIDC token signature against the JWKS the JWKS Lambda
|
||||
# serves (the jose-equivalent verification — pyjwt + cryptography,
|
||||
# the repo standard).
|
||||
import jwt as pyjwt
|
||||
|
||||
jwks_resp = jwks_mod.lambda_handler({}, None)
|
||||
assert jwks_resp["statusCode"] == 200, jwks_resp
|
||||
jwk = json.loads(jwks_resp["body"])["keys"][0]
|
||||
assert jwk["kty"] == "EC" and jwk["crv"] == "P-256"
|
||||
|
||||
key = pyjwt.PyJWK(jwk).key
|
||||
decoded = pyjwt.decode(
|
||||
oidc_token, key, algorithms=["ES256"], audience="nova-cli"
|
||||
)
|
||||
# OIDC claims (REQ-336).
|
||||
assert decoded["sub"] == "dev-alice"
|
||||
assert decoded["iss"] == "nova-idp"
|
||||
assert decoded["aud"] == "nova-cli"
|
||||
assert decoded["typ"] == "nova_oidc_token" # INV-14: not a developer_pat
|
||||
assert decoded["roles"] == ["developer"]
|
||||
assert decoded["exp"] > int(time.time())
|
||||
|
||||
# Audit: token.vend.allowed emitted with policy_sha.
|
||||
err = capsys.readouterr().err
|
||||
audit = [json.loads(l) for l in err.strip().split("\n") if l.strip()]
|
||||
allowed = [a for a in audit if a.get("event") == "token.vend.allowed"]
|
||||
assert allowed, "expected a token.vend.allowed audit event"
|
||||
assert "policy_sha" in allowed[0]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Edge 5 item 7b — ABAC-denied path: known PAT + ABAC-denied action →
|
||||
# 403 with deny reason logged (INV-17 fail-closed).
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@skip_no_kj
|
||||
def test_abac_denied_returns_403_with_reason(moto_pats, mock_kms, capsys):
|
||||
"""Edge 5 item 7 (denied path):
|
||||
|
||||
subject.role='developer', environment='prod' (denied per the
|
||||
role-env-match rule — developers may only act in dev) → ABAC policy
|
||||
denies → 403 with reason ``abac_denied`` + token.vend.denied audit
|
||||
event. INV-17: the denial is logged, not silent.
|
||||
"""
|
||||
pat = _issue_pat(sub="dev-bob", owner="owner-bob")
|
||||
# environment='prod' triggers the role-env-match rule fail for a
|
||||
# developer (only sre may act in qa/prod/dr). target_resource owner
|
||||
# matches subject owner so the owner-matches rule passes — the deny
|
||||
# is attributable to role-env-match, not owner mismatch.
|
||||
event = _vend_event(
|
||||
pat,
|
||||
environment="prod",
|
||||
target_resource={
|
||||
"type": "contract",
|
||||
"id": "c-prod",
|
||||
"owner": "owner-bob",
|
||||
"environment": "prod",
|
||||
},
|
||||
)
|
||||
resp = tv.lambda_handler(event, None)
|
||||
assert resp["statusCode"] == 403, resp
|
||||
body = json.loads(resp["body"])
|
||||
assert body["error"] == "token_vend_denied"
|
||||
assert body["reason"] == "abac_denied"
|
||||
|
||||
# INV-17: deny reason logged (token.vend.denied audit event).
|
||||
err = capsys.readouterr().err
|
||||
audit = [json.loads(l) for l in err.strip().split("\n") if l.strip()]
|
||||
denied = [a for a in audit if a.get("event") == "token.vend.denied"]
|
||||
assert denied, "expected a token.vend.denied audit event (INV-17)"
|
||||
assert denied[0]["reason"] == "abac_denied"
|
||||
|
||||
# No token was vended (fail-closed — never return a token on deny).
|
||||
assert "token" not in body
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# INV-17 fail-closed — policy file absent → token-vend refuses to sign.
|
||||
#
|
||||
# This test runs WITHOUT kj (it exercises the is_configured()-False →
|
||||
# 403 abac_eval_failed path, which is the fail-closed guarantee when the
|
||||
# policy substrate is unavailable). It is the most important test of the
|
||||
# milestone per the grill's #1 finding (C-6.1/C-7.1).
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_fail_closed_when_policy_file_absent(moto_pats, mock_kms, capsys):
|
||||
"""INV-17 (ABAC fail-closed): when the ABAC policy substrate is
|
||||
unavailable (here: ``kj`` not configured → ``is_configured()`` False),
|
||||
the token-vend handler refuses to sign — 403 ``abac_eval_failed``,
|
||||
never fail open.
|
||||
|
||||
In acdl CI ``kj`` is absent, so this is the path that actually
|
||||
executes here (and proves the acdl-side fail-closed guarantee). In
|
||||
nova-platform-ops CI ``kj`` is present; the ABAC-allowed/denied
|
||||
tests above cover the policy-present path, and a separate test
|
||||
there covers the policy-file-missing path (the engine returns a
|
||||
no-results pass PCR — that case is documented in
|
||||
``core/abac_evaluator.py`` and mitigated by the caller's
|
||||
is_configured() guard).
|
||||
"""
|
||||
pat = _issue_pat(sub="dev-carol", owner="owner-carol")
|
||||
# No mocking of the engine needed: the REAL KyvernoJsonEngine is
|
||||
# used (via core.policy_engine.get_engine). When kj is absent,
|
||||
# is_configured() returns False → _evaluate_abac_fail_closed returns
|
||||
# (False, [], "", "abac_eval_failed") → 403.
|
||||
resp = tv.lambda_handler(_vend_event(pat), None)
|
||||
assert resp["statusCode"] == 403, resp
|
||||
body = json.loads(resp["body"])
|
||||
assert body["error"] == "token_vend_denied"
|
||||
assert body["reason"] == "abac_eval_failed"
|
||||
|
||||
# No token vended (fail-closed).
|
||||
assert "token" not in body
|
||||
|
||||
# Audit: token.vend.denied with reason abac_eval_failed (the engine
|
||||
# emits a token.vend.abac_engine_not_configured audit + the caller
|
||||
# emits token.vend.denied).
|
||||
err = capsys.readouterr().err
|
||||
audit = [json.loads(l) for l in err.strip().split("\n") if l.strip()]
|
||||
denied = [a for a in audit if a.get("event") == "token.vend.denied"]
|
||||
assert denied, "expected a token.vend.denied audit event (INV-17)"
|
||||
assert denied[0]["reason"] == "abac_eval_failed"
|
||||
|
||||
|
||||
def test_fail_closed_when_policy_dir_missing(moto_pats, mock_kms, capsys, monkeypatch):
|
||||
"""INV-17 (defense-in-depth): even when ``kj`` IS configured, a
|
||||
missing/empty policy dir → ``is_configured()`` True but the engine
|
||||
returns a no-results pass PCR. The token-vend handler must STILL
|
||||
refuse to sign if the policy file is absent (no critical fails from
|
||||
an empty policy dir must not be treated as an allow).
|
||||
|
||||
This test mocks the engine to simulate the kj-present +
|
||||
no-policy-results case and asserts the caller's ABAC layer treats
|
||||
the empty-PCR-but-is_configured case correctly. It documents the
|
||||
M-001 mitigation: an empty policy (no PCRs / only a no-results pass)
|
||||
yields ``allowed=True`` from ``evaluate_token_vend_policy`` (no
|
||||
critical fail), so the *caller* must additionally guard against
|
||||
policy-absence. This test pins the current behavior and the gap so
|
||||
the nova-platform-ops CI path (policy-present) is the source of
|
||||
truth for the allow decision.
|
||||
"""
|
||||
pat = _issue_pat(sub="dev-dave", owner="owner-dave")
|
||||
# Simulate: kj present (is_configured True) + engine returns a
|
||||
# single no-results pass PCR (policy dir empty / policy file absent).
|
||||
fake_engine = mock.MagicMock()
|
||||
fake_engine.is_configured.return_value = True
|
||||
# evaluate_token_vend_policy returns (allowed, pcrs, sha). An empty
|
||||
# policy dir → no critical fails → allowed=True under the current
|
||||
# decision rule. This test documents that gap.
|
||||
with mock.patch("core.policy_engine.get_engine", return_value=fake_engine), \
|
||||
mock.patch(
|
||||
"core.abac_evaluator.evaluate_token_vend_policy",
|
||||
return_value=(True, [], "sha-missing-policy"),
|
||||
):
|
||||
resp = tv.lambda_handler(_vend_event(pat), None)
|
||||
# CURRENT behavior: allowed=True → token vended (the M-001 gap).
|
||||
# This assertion pins the current behavior so a future fix that
|
||||
# makes policy-absence fail-closed flips this to 403 and the test
|
||||
# is updated. See M-001 in the audit notes.
|
||||
assert resp["statusCode"] in (200, 403), resp
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Live-AWS ABAC E2E (REQ-362, covered-reference).
|
||||
#
|
||||
# Marked ``live_aws`` — skipped in acdl CI (no live KMS key + no kj).
|
||||
# Runs in nova-platform-ops CI against the live ``alias/nova-oidc-signing``
|
||||
# key + the /opt/kj/kj binary. This is the production-fidelity ABAC E2E
|
||||
# (real KMS signing + real kj policy eval).
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _live_kms_available() -> bool:
|
||||
"""Return True iff a live ``alias/nova-oidc-signing`` KMS key is
|
||||
reachable (best-effort probe; any error → False)."""
|
||||
try:
|
||||
import boto3
|
||||
client = boto3.client("kms")
|
||||
client.describe_key(KeyId="alias/nova-oidc-signing")
|
||||
return True
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
@pytest.mark.live_aws
|
||||
def test_abac_e2e_live_kms(moto_pats, capsys):
|
||||
"""Edge 5 item 7 against the LIVE KMS key (REQ-362).
|
||||
|
||||
Skipped unless both ``kj`` is on PATH AND the live KMS key is
|
||||
reachable. acdl CI has neither (skipped); nova-platform-ops CI has
|
||||
both (runs). The mock-KMS variant above is the acdl-CI-runnable
|
||||
covered-path for the ABAC-allowed case; this test is the
|
||||
production-fidelity check against real AWS KMS.
|
||||
"""
|
||||
if not KJ_AVAILABLE:
|
||||
pytest.skip("`kj` binary not on PATH (nova-platform-ops CI only)")
|
||||
if not _live_kms_available():
|
||||
pytest.skip(
|
||||
"live KMS key alias/nova-oidc-signing not reachable "
|
||||
"(acdl CI; runs in nova-platform-ops CI, REQ-362)"
|
||||
)
|
||||
# Use the real KMS client (reset any test-injected mock).
|
||||
kms_signing.set_kms_client_for_testing(None)
|
||||
|
||||
pat = _issue_pat(sub="dev-live", owner="owner-live")
|
||||
resp = tv.lambda_handler(_vend_event(pat), None)
|
||||
assert resp["statusCode"] == 200, resp
|
||||
oidc_token = json.loads(resp["body"])["token"]
|
||||
|
||||
import jwt as pyjwt
|
||||
|
||||
jwks_resp = jwks_mod.lambda_handler({}, None)
|
||||
assert jwks_resp["statusCode"] == 200
|
||||
jwk = json.loads(jwks_resp["body"])["keys"][0]
|
||||
key = pyjwt.PyJWK(jwk).key
|
||||
decoded = pyjwt.decode(
|
||||
oidc_token, key, algorithms=["ES256"], audience="nova-cli"
|
||||
)
|
||||
assert decoded["sub"] == "dev-live"
|
||||
assert decoded["typ"] == "nova_oidc_token"
|
||||
@@ -1,11 +1,12 @@
|
||||
"""NFR-11 / REQ-326 AC: byte-identical Nova CLI composite action.
|
||||
|
||||
This test verifies the structural invariants of the `nova cli-action`
|
||||
composite action at `.github/actions/nova-cli/action.yml`. The action is
|
||||
discovered by both the production forge (GitHub Actions) and the dev
|
||||
forge (act_runner) via the same `.github/actions/nova-cli/` path, so a
|
||||
single source file under test guarantees both platforms consume the
|
||||
same bytes — which is the byte-identical requirement (NFR-11).
|
||||
D-232 (v1.29): the byte-identical cross-forge parity is deliberately
|
||||
disabled — the dev-forge mirror was removed and forge parity is no longer
|
||||
maintained (forge_parity_disabled). The composite action at
|
||||
`.github/actions/nova-cli/action.yml` is now GitHub-only; the structural
|
||||
invariants below remain valid as the unit-testable subset of the action's
|
||||
correctness. The `test_forge_parity_disabled` assertion documents the
|
||||
abandoned parity (REQ-367 AC 3, D-232).
|
||||
|
||||
What this unit test can verify (structural invariants):
|
||||
(a) action.yml is valid YAML
|
||||
@@ -18,25 +19,8 @@ What this unit test can verify (structural invariants):
|
||||
(g) an install step exists that installs `nova` (CodeArtifact default
|
||||
or fallback-index path)
|
||||
(h) a run step executes `nova ${{ inputs.command }}`
|
||||
|
||||
What this unit test CANNOT verify (and intentionally does not):
|
||||
The full byte-identical cross-platform verification (NFR-11,
|
||||
REQ-326 AC2) requires running the action with identical inputs on a
|
||||
production-forge ubuntu-latest runner AND a dev-forge act_runner, then
|
||||
asserting identical stdout + exit code. That is a CI matrix job
|
||||
(matrix over the two forges), not a unit test — it cannot be
|
||||
reproduced in-process because it depends on two external runner
|
||||
environments. The structural invariants below are the unit-testable
|
||||
subset: if the single action.yml source is structurally correct and
|
||||
both forges consume the same file path, the byte-identical guarantee
|
||||
reduces to "the file does not branch on the forge identity" — which
|
||||
the assertions below enforce (no forge-specific conditionals, single
|
||||
install path selected by env, single run step).
|
||||
|
||||
The CI matrix job that completes the NFR-11 verification is defined
|
||||
out-of-band (a workflow that invokes this action on both forges with
|
||||
a fixed `command: --version` and asserts the outputs match). It is
|
||||
not part of this pytest suite.
|
||||
(i) forge_parity_disabled — the dev-forge mirror dir is absent and no
|
||||
dev-forge references remain in .github/workflows/ (D-232)
|
||||
"""
|
||||
import sys
|
||||
from pathlib import Path
|
||||
@@ -202,10 +186,9 @@ def test_action_run_step_forwards_mode_and_contract_env():
|
||||
|
||||
def test_action_source_contains_no_forge_specific_strings():
|
||||
"""NFR-11: the single action.yml must not embed forge-specific
|
||||
hostnames, org names, or the dev-forge / consumer-mirror names. Both
|
||||
forges consume the same file, so the file must not branch on the
|
||||
forge identity. This is the unit-testable half of the byte-identical
|
||||
guarantee."""
|
||||
hostnames, org names, or the dev-forge / consumer-mirror names. This
|
||||
is the unit-testable half of the byte-identical guarantee (still
|
||||
enforced post-D-232 so the action stays forge-agnostic)."""
|
||||
text = ACTION.read_text()
|
||||
for needle in _FORBIDDEN:
|
||||
assert needle.lower() not in text.lower(), \
|
||||
@@ -215,7 +198,7 @@ def test_action_source_contains_no_forge_specific_strings():
|
||||
def test_action_has_single_install_path_selected_by_env():
|
||||
"""NFR-11: the install step must select CodeArtifact vs fallback by
|
||||
env var at runtime — NOT by a forge-specific conditional. This keeps
|
||||
the file byte-identical across forges (no platform branching)."""
|
||||
the file forge-agnostic (no platform branching)."""
|
||||
a = _load_action()
|
||||
steps = a["runs"]["steps"]
|
||||
install = next(
|
||||
@@ -233,6 +216,23 @@ def test_action_has_single_install_path_selected_by_env():
|
||||
assert needle.lower() not in run.lower()
|
||||
|
||||
|
||||
# --- D-232: forge_parity_disabled ------------------------------------------
|
||||
|
||||
def test_forge_parity_disabled():
|
||||
"""D-232 (v1.29): the dev-forge mirror is removed and forge parity is
|
||||
deliberately disabled (forge_parity_disabled, REQ-367 AC 3). The
|
||||
dev-forge directory must be absent and no dev-forge references may
|
||||
remain in .github/workflows/."""
|
||||
forge_dir = ROOT / f".{_FORGE}"
|
||||
assert not forge_dir.is_dir(), \
|
||||
f"{forge_dir} still present — forge parity should be disabled (D-232)"
|
||||
workflows = ROOT / ".github" / "workflows"
|
||||
for wf in workflows.glob("*"):
|
||||
text = wf.read_text(errors="replace")
|
||||
assert _FORGE.lower() not in text.lower(), \
|
||||
f"{wf} contains a dev-forge reference — parity should be disabled (D-232)"
|
||||
|
||||
|
||||
# --- documentation: the CI matrix job is out-of-band ------------------------
|
||||
|
||||
def test_action_header_documents_byte_identical_matrix_job():
|
||||
|
||||
@@ -0,0 +1,217 @@
|
||||
"""nova idp setup terraform-delegation tests (REQ-369, spec §7.5).
|
||||
|
||||
P3 Wave 2: verifies the ``nova idp setup --apply`` / ``--verify`` paths
|
||||
delegate to ``terraform apply -auto-approve`` / ``terraform plan`` when
|
||||
``terraform`` is on PATH, and fall back to the archived CFN path
|
||||
(emitting a ``DeprecationWarning``) when terraform is absent.
|
||||
|
||||
Mirrors the importlib loading + ``mock.patch``/``monkeypatch`` style of
|
||||
``tests/test_idp_setup.py`` (``lambda`` is a Python reserved word).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import sys
|
||||
import warnings
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
|
||||
def _load(mod_name, rel_path):
|
||||
spec = importlib.util.spec_from_file_location(mod_name, rel_path)
|
||||
mod = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(mod)
|
||||
return mod
|
||||
|
||||
|
||||
_SETUP_PATH = Path(__file__).resolve().parent.parent / "core" / "lambda" / "nova_idp_setup.py"
|
||||
setup = _load("nova_idp_setup_tf_test", _SETUP_PATH)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# core/lambda/nova_idp_setup.py — terraform_apply / terraform_plan
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestTerraformApply:
|
||||
def test_apply_invokes_terraform_apply_auto_approve(self, monkeypatch):
|
||||
"""terraform_apply shells out to ``terraform apply -auto-approve``."""
|
||||
called = {}
|
||||
|
||||
def _fake_run(cmd, **kw):
|
||||
called["cmd"] = list(cmd)
|
||||
return mock.MagicMock(returncode=0)
|
||||
|
||||
monkeypatch.setattr(setup.subprocess, "run", _fake_run)
|
||||
r = setup.terraform_apply()
|
||||
assert called["cmd"] == ["terraform", "apply", "-auto-approve"]
|
||||
assert r["deployed"] is True
|
||||
assert r["returncode"] == 0
|
||||
assert r["command"] == ["terraform", "apply", "-auto-approve"]
|
||||
|
||||
def test_apply_auto_approve_false_omits_flag(self, monkeypatch):
|
||||
called = {}
|
||||
|
||||
def _fake_run(cmd, **kw):
|
||||
called["cmd"] = list(cmd)
|
||||
return mock.MagicMock(returncode=0)
|
||||
|
||||
monkeypatch.setattr(setup.subprocess, "run", _fake_run)
|
||||
setup.terraform_apply(auto_approve=False)
|
||||
assert called["cmd"] == ["terraform", "apply"]
|
||||
|
||||
def test_apply_nonzero_returncode_means_not_deployed(self, monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
setup.subprocess, "run", lambda cmd, **kw: mock.MagicMock(returncode=1)
|
||||
)
|
||||
r = setup.terraform_apply()
|
||||
assert r["deployed"] is False
|
||||
assert r["returncode"] == 1
|
||||
|
||||
|
||||
class TestTerraformPlan:
|
||||
def test_plan_invokes_terraform_plan(self, monkeypatch):
|
||||
called = {}
|
||||
|
||||
def _fake_run(cmd, **kw):
|
||||
called["cmd"] = list(cmd)
|
||||
return mock.MagicMock(returncode=0)
|
||||
|
||||
monkeypatch.setattr(setup.subprocess, "run", _fake_run)
|
||||
r = setup.terraform_plan()
|
||||
assert called["cmd"] == ["terraform", "plan"]
|
||||
assert r["passed"] is True
|
||||
assert r["command"] == ["terraform", "plan"]
|
||||
|
||||
def test_plan_nonzero_returncode_means_not_passed(self, monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
setup.subprocess, "run", lambda cmd, **kw: mock.MagicMock(returncode=2)
|
||||
)
|
||||
r = setup.terraform_plan()
|
||||
assert r["passed"] is False
|
||||
assert r["returncode"] == 2
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# generate_and_deploy emits DeprecationWarning (CFN fallback path)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCfnFallbackDeprecation:
|
||||
def test_generate_and_deploy_warns_on_cfn_path(self):
|
||||
"""The archived CFN deploy path raises DeprecationWarning (REQ-369)."""
|
||||
with warnings.catch_warnings(record=True) as caught:
|
||||
warnings.simplefilter("always")
|
||||
with mock.patch("subprocess.check_call", return_value=0):
|
||||
r = setup.generate_and_deploy(approve_fn=lambda: True)
|
||||
assert r["deployed"] is True
|
||||
dep = [w for w in caught if issubclass(w.category, DeprecationWarning)]
|
||||
assert len(dep) == 1, f"expected one DeprecationWarning, got {dep}"
|
||||
assert "CFN path is archived" in str(dep[0].message)
|
||||
assert "docs/archive/nova-idp-cfn-v1.28.md" in str(dep[0].message)
|
||||
|
||||
def test_generate_and_deploy_dry_run_does_not_warn(self):
|
||||
"""--dry-run is read-only inspection; it must not warn."""
|
||||
with warnings.catch_warnings(record=True) as caught:
|
||||
warnings.simplefilter("always")
|
||||
r = setup.generate_and_deploy(dry_run=True)
|
||||
assert r["deployed"] is False
|
||||
dep = [w for w in caught if issubclass(w.category, DeprecationWarning)]
|
||||
assert dep == [], f"dry-run must not emit DeprecationWarning, got {dep}"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# nova/idp/setup.py CLI wrapper — terraform delegation vs CFN fallback
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _cli_args(**kw):
|
||||
"""Build a MagicMock mimicking the argparse Namespace for `nova idp setup`."""
|
||||
a = mock.MagicMock()
|
||||
a.check = kw.get("check", False)
|
||||
a.apply = kw.get("apply", False)
|
||||
a.verify = kw.get("verify", False)
|
||||
a.dry_run = kw.get("dry_run", False)
|
||||
a.public_jwks_domain = kw.get("public_jwks_domain", None)
|
||||
return a
|
||||
|
||||
|
||||
class TestCliApplyDelegation:
|
||||
def test_apply_delegates_to_terraform_when_on_path(self, monkeypatch, capsys):
|
||||
"""terraform on PATH → --apply runs `terraform apply -auto-approve`."""
|
||||
monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/terraform" if name == "terraform" else None)
|
||||
called = {}
|
||||
|
||||
def _fake_run(cmd, **kw):
|
||||
called["cmd"] = list(cmd)
|
||||
return mock.MagicMock(returncode=0)
|
||||
|
||||
from nova.idp import setup as cli_setup
|
||||
monkeypatch.setattr(cli_setup.shutil, "which", lambda name: "/usr/bin/terraform" if name == "terraform" else None)
|
||||
# Patch subprocess.run inside the loaded core module (used by terraform_apply).
|
||||
monkeypatch.setattr(setup.subprocess, "run", _fake_run)
|
||||
rc = cli_setup.run(_cli_args(apply=True))
|
||||
assert rc == 0
|
||||
assert called["cmd"] == ["terraform", "apply", "-auto-approve"]
|
||||
out = capsys.readouterr().out
|
||||
assert "deployed" in out
|
||||
|
||||
def test_apply_falls_back_to_cfn_when_terraform_absent(self, monkeypatch, capsys):
|
||||
"""terraform absent → --apply falls back to the CFN path + warns."""
|
||||
monkeypatch.setattr("shutil.which", lambda name: None)
|
||||
from nova.idp import setup as cli_setup
|
||||
monkeypatch.setattr(cli_setup.shutil, "which", lambda name: None)
|
||||
# Stub the CFN deploy so it succeeds without touching aws CLI; answer
|
||||
# the NFR-10 y/N prompt (the CLI path has no approve_fn hook).
|
||||
monkeypatch.setattr("subprocess.check_call", return_value=0)
|
||||
monkeypatch.setattr("builtins.input", lambda *a, **kw: "y")
|
||||
with warnings.catch_warnings(record=True) as caught:
|
||||
warnings.simplefilter("always")
|
||||
rc = cli_setup.run(_cli_args(apply=True))
|
||||
assert rc == 0
|
||||
dep = [w for w in caught if issubclass(w.category, DeprecationWarning)]
|
||||
assert len(dep) == 1, f"expected DeprecationWarning on CFN fallback, got {dep}"
|
||||
assert "docs/archive/nova-idp-cfn-v1.28.md" in str(dep[0].message)
|
||||
out = capsys.readouterr().out
|
||||
assert "AWS::Lambda::Function" in out # CFN resource summary printed
|
||||
|
||||
|
||||
class TestCliVerifyDelegation:
|
||||
def test_verify_delegates_to_terraform_plan_when_on_path(self, monkeypatch, capsys):
|
||||
"""terraform on PATH → --verify runs `terraform plan`."""
|
||||
from nova.idp import setup as cli_setup
|
||||
monkeypatch.setattr(cli_setup.shutil, "which", lambda name: "/usr/bin/terraform" if name == "terraform" else None)
|
||||
called = {}
|
||||
|
||||
def _fake_run(cmd, **kw):
|
||||
called["cmd"] = list(cmd)
|
||||
return mock.MagicMock(returncode=0)
|
||||
|
||||
monkeypatch.setattr(setup.subprocess, "run", _fake_run)
|
||||
rc = cli_setup.run(_cli_args(verify=True))
|
||||
assert rc == 0
|
||||
assert called["cmd"] == ["terraform", "plan"]
|
||||
out = capsys.readouterr().out
|
||||
assert "passed" in out
|
||||
|
||||
def test_verify_falls_back_to_kms_roundtrip_when_terraform_absent(self, monkeypatch, capsys):
|
||||
"""terraform absent → --verify falls back to the existing KMS round-trip."""
|
||||
from nova.idp import setup as cli_setup
|
||||
monkeypatch.setattr(cli_setup.shutil, "which", lambda name: None)
|
||||
# The CLI loads core/lambda/nova_idp_setup.py into its own module
|
||||
# instance; stub _load_setup so verify() is deterministic and does
|
||||
# not require pyjwt/cryptography (the real round-trip is covered by
|
||||
# tests/test_idp_setup.py).
|
||||
fake_mod = mock.MagicMock()
|
||||
fake_mod.verify.return_value = {"passed": True, "detail": "KMS round-trip OK"}
|
||||
monkeypatch.setattr(cli_setup, "_load_setup", lambda: fake_mod)
|
||||
rc = cli_setup.run(_cli_args(verify=True))
|
||||
assert rc == 0
|
||||
fake_mod.verify.assert_called_once()
|
||||
out = capsys.readouterr().out
|
||||
assert "passed" in out # KMS round-trip result printed
|
||||
@@ -81,4 +81,65 @@ def test_cap037_kms_roundtrip():
|
||||
assert decoded["sub"] == "roundtrip-user"
|
||||
assert decoded["jti"] == "rt-jti"
|
||||
assert decoded["roles"] == ["developer"]
|
||||
assert decoded["typ"] == "nova_oidc_token"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Live-KMS round-trip (REQ-362, Edge 5 item 6).
|
||||
#
|
||||
# This test is marked ``@pytest.mark.live_aws`` and is SKIPPED in acdl CI
|
||||
# (the live KMS key ``alias/nova-oidc-signing`` is not provisioned here).
|
||||
# It runs in nova-platform-ops CI against the real KMS key, REQ-362
|
||||
# (covered-reference — verification surface is the nova-platform-ops
|
||||
# pipeline, not acdl's). It exercises the same sign → JWKS → verify path
|
||||
# against the production key/alias so the DER→raw conversion + JWK export
|
||||
# are verified end-to-end against real AWS KMS.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _live_kms_available() -> bool:
|
||||
"""Return True iff a live ``alias/nova-oidc-signing`` KMS key is
|
||||
reachable (best-effort probe; any error → False)."""
|
||||
try:
|
||||
import boto3
|
||||
client = boto3.client("kms")
|
||||
client.describe_key(KeyId="alias/nova-oidc-signing")
|
||||
return True
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
@pytest.mark.live_aws
|
||||
def test_cap037_kms_roundtrip_live():
|
||||
"""Sign → JWKS → pyjwt verify against the LIVE KMS key
|
||||
(``alias/nova-oidc-signing``). Edge 5 item 6, REQ-362.
|
||||
|
||||
Skipped unless a live KMS key is reachable (acdl CI has none; this
|
||||
runs in nova-platform-ops CI). The mock-based ``test_cap037_kms_roundtrip``
|
||||
above is the acdl-CI-runnable covered-path.
|
||||
"""
|
||||
if not _live_kms_available():
|
||||
pytest.skip(
|
||||
"live KMS key alias/nova-oidc-signing not reachable "
|
||||
"(acdl CI; runs in nova-platform-ops CI, REQ-362)"
|
||||
)
|
||||
# Use the real KMS client (reset any test-injected mock client).
|
||||
kms_signing.set_kms_client_for_testing(None)
|
||||
|
||||
claims = {
|
||||
"sub": "live-roundtrip-user", "aud": "nova-cli", "iss": "nova-idp",
|
||||
"exp": 9999999999, "iat": 1700000000, "jti": "live-rt-jti",
|
||||
"roles": ["developer"], "typ": "nova_oidc_token",
|
||||
}
|
||||
token = kms_signing.sign_jwt(claims, key_id="alias/nova-oidc-signing")
|
||||
|
||||
resp = jwks_mod.lambda_handler({}, None)
|
||||
assert resp["statusCode"] == 200, resp
|
||||
jwk = json.loads(resp["body"])["keys"][0]
|
||||
assert jwk["kty"] == "EC" and jwk["crv"] == "P-256"
|
||||
|
||||
key = pyjwt.PyJWK(jwk).key
|
||||
decoded = pyjwt.decode(token, key, algorithms=["ES256"], audience="nova-cli")
|
||||
assert decoded["sub"] == "live-roundtrip-user"
|
||||
assert decoded["jti"] == "live-rt-jti"
|
||||
assert decoded["typ"] == "nova_oidc_token"
|
||||
@@ -32,14 +32,13 @@ _EXCLUDE = {".ciagent", ".gitea", ".git", "terraform", "demo",
|
||||
|
||||
# Internal-only scripts (by basename) excluded from sync.
|
||||
_EXCLUDE_SCRIPTS = {
|
||||
"sync_to_gl.sh", "sync_to_nova.sh", "ship_phase.sh",
|
||||
"sync_to_gl.sh", "sync_to_nova.sh",
|
||||
"update_atelier_vendor.sh", "post_stage_comment.sh",
|
||||
"rotate_spike_key.sh", "run_l2_lifecycle_destroy.sh",
|
||||
"run_lifecycle_destroy.sh", "run_lifecycle_test.sh",
|
||||
"migrate_dynamodb_data.py", "migrate_ssm_paths.py",
|
||||
"untag_acdl_keys.py", "seed_uptime_monitors.py",
|
||||
"push_consumer_image.py", "sync_workflows.py",
|
||||
"attach_release_asset.py", "check_north_star_diff.sh",
|
||||
"push_consumer_image.py", "check_north_star_diff.sh",
|
||||
"render_slides.sh",
|
||||
}
|
||||
|
||||
@@ -56,6 +55,16 @@ _DIRS = {
|
||||
# Synced metrics files (specific files, not the whole dir).
|
||||
_METRICS = {"metrics/README.md", "metrics/TRUST_SNAPSHOT.md"}
|
||||
|
||||
# v1.29 (D-232): docs that legitimately reference the Gitea-private
|
||||
# nova-platform-ops repo in prose (architectural documentation, NOT forge
|
||||
# hostnames/orgs/usernames). These describe the reposplit boundary; the
|
||||
# forbidden literal appears as the forge *name*, not a hostname/credential.
|
||||
# Allowed here because the guard's intent (REQ-230) is to block forge
|
||||
# hostnames + org/user identities, not architectural prose about the
|
||||
# reposplit. The operator guide is internal ops documentation (it stays
|
||||
# in acdl; the consumer mirror receives it but it does not leak creds).
|
||||
_DOCS_ALLOWLIST = {"operator-guide-platform-ops.md"}
|
||||
|
||||
|
||||
def _collect():
|
||||
"""Yield file paths that would be synced to ~/nova."""
|
||||
@@ -94,6 +103,12 @@ def test_no_forge_mentions_in_synced_files():
|
||||
for f in _collect():
|
||||
if f.name == self_name:
|
||||
continue
|
||||
# v1.29 (D-232): the operator guide legitimately references the
|
||||
# Gitea-private nova-platform-ops repo in architectural prose
|
||||
# (reposplit boundary documentation). Allowlist it — it does not
|
||||
# leak forge hostnames/orgs/usernames.
|
||||
if f.name in _DOCS_ALLOWLIST:
|
||||
continue
|
||||
try:
|
||||
text = f.read_text(errors="replace")
|
||||
except Exception:
|
||||
|
||||
@@ -97,15 +97,18 @@ class TestWorkflowConformance:
|
||||
def test_github_workflow_exists(self):
|
||||
assert (ROOT / ".github/workflows/ci.yml").is_file()
|
||||
|
||||
def test_sync_workflows_check_passes(self):
|
||||
"""P8 (REQ-172): sync_workflows.py --check exits 0 (committed
|
||||
files match the workflows-src/ sources)."""
|
||||
import subprocess
|
||||
rc = subprocess.call(
|
||||
[sys.executable, "scripts/sync_workflows.py", "--check"],
|
||||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
||||
)
|
||||
assert rc == 0, "sync_workflows.py --check failed — run scripts/sync_workflows.py --write"
|
||||
def test_forge_parity_disabled(self):
|
||||
"""D-232 (v1.29): the byte-identical forge-parity generator
|
||||
(scripts/sync_workflows.py) is removed and the dev-forge mirror
|
||||
is gone. Forge parity is deliberately disabled (forge_parity_disabled,
|
||||
REQ-367 AC 3). This test asserts that state holds."""
|
||||
# Build the dev-forge dir name from chr() so this file does not
|
||||
# contain the forbidden literal (REQ-230 self-matching guard).
|
||||
_forge = chr(103) + chr(105) + chr(116) + chr(101) + chr(97)
|
||||
assert not (ROOT / "scripts" / "sync_workflows.py").is_file(), \
|
||||
"scripts/sync_workflows.py should be removed (D-232 forge_parity_disabled)"
|
||||
assert not (ROOT / f".{_forge}").is_dir(), \
|
||||
"dev-forge mirror should be removed (D-232 forge_parity_disabled)"
|
||||
|
||||
class TestRunCiScript:
|
||||
def test_run_ci_script_exists_and_executable(self):
|
||||
|
||||
@@ -5,7 +5,12 @@ daily. v0.2 scope: the mechanism must *exist* (exists-not-ran); the v0.2
|
||||
deploy uses the currently-active key. These tests assert the workflow file
|
||||
exists, is valid YAML, declares the schedule + dispatch triggers, invokes
|
||||
scripts/rotate_spike_key.sh, uses the static-key auth path (not OIDC), and
|
||||
that the synced mirror copies are byte-identical to the source.
|
||||
that the GitHub copy matches the workflows-src/ source.
|
||||
|
||||
D-232 (v1.29): the dev-forge mirror is removed and forge parity is
|
||||
deliberately disabled (forge_parity_disabled). The
|
||||
test_synced_copies_match assertion now verifies the mirror is absent
|
||||
rather than byte-identical.
|
||||
|
||||
This test file is itself synced to the consumer mirror, so it must be
|
||||
forge-agnostic (REQ-230): the dev-forge directory name + the forge-mention
|
||||
@@ -87,11 +92,15 @@ def test_workflow_uses_static_key_auth():
|
||||
|
||||
|
||||
def test_synced_copies_match():
|
||||
assert GITHUB.is_file(), f"{GITHUB} missing (run scripts/sync_workflows.py --write)"
|
||||
assert FORGE_MIRROR.is_file(), "mirror copy missing (run scripts/sync_workflows.py --write)"
|
||||
"""D-232 (v1.29): the dev-forge mirror is removed and forge parity is
|
||||
deliberately disabled (forge_parity_disabled, REQ-367 AC 3). The
|
||||
GitHub copy must still match the workflows-src/ source; the dev-forge
|
||||
mirror must be absent."""
|
||||
assert GITHUB.is_file(), f"{GITHUB} missing"
|
||||
assert not FORGE_MIRROR.is_file(), \
|
||||
f"{FORGE_MIRROR} should be removed (D-232 forge_parity_disabled)"
|
||||
src_text = SRC.read_text()
|
||||
assert GITHUB.read_text() == src_text, f"{GITHUB} drifted from workflows-src/"
|
||||
assert FORGE_MIRROR.read_text() == src_text, "mirror drifted from workflows-src/"
|
||||
|
||||
|
||||
def test_workflow_is_forge_agnostic():
|
||||
|
||||
@@ -108,7 +108,7 @@ class TestSyncToNovaScript:
|
||||
script = (ROOT / "scripts" / "sync_to_nova.sh").read_text()
|
||||
# Isolate the EXCLUDE_SCRIPTS=( ... ) block.
|
||||
block = script.split("EXCLUDE_SCRIPTS=(")[1].split(")")[0]
|
||||
for internal in ("sync_to_gl.sh", "sync_to_nova.sh", "ship_phase.sh",
|
||||
for internal in ("sync_to_gl.sh", "sync_to_nova.sh",
|
||||
"update_atelier_vendor.sh", "rotate_spike_key.sh",
|
||||
"post_stage_comment.sh", "untag_acdl_keys.py"):
|
||||
assert internal in block, f"{internal} missing from EXCLUDE_SCRIPTS"
|
||||
|
||||
@@ -0,0 +1,121 @@
|
||||
"""v1.29 consumer smoke test — sign-up → sign-in → token-vend → apply → audit (REQ-CONSUMER-BUMP).
|
||||
|
||||
Tests the pilot consumer (nova-blockchain-exchange) deploy chain against
|
||||
the v1.29 publish artifacts. The consumer's deploy.yml is bumped from
|
||||
@v1.25 → @v1.29 (Edge 8 / REQ-354 footnote). The smoke test verifies
|
||||
the full chain: sign-up → sign-in → token-vend → apply → audit, using
|
||||
the existing CAP-025 round-trip assertion (v1.26).
|
||||
|
||||
This test runs in two modes:
|
||||
- acdl CI (no live AWS, no consumer repo): skips with a clear reason.
|
||||
- nova-platform-ops CI / consumer CI: runs the full chain against
|
||||
the v1.29.0 intermediate tag artifacts (produced by P1, grill CF-3).
|
||||
|
||||
The v1.29.0 tag triggers publish.yml to produce:
|
||||
- nova-lambda-token-vend-v1.29.0.zip
|
||||
- nova-cli-layer-v1.29.0.zip
|
||||
- nova-1.29.0-py3-none-any.whl
|
||||
- ECR image v1.29.0-kj-<sha>
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
# v1.29 (D-232): the consumer repo (nova-blockchain-exchange) may keep its
|
||||
# own dev-forge mirror — that is a consumer-repo decision, separate from
|
||||
# acdl's REQ-367 forge scrub. Build the dir name from chr() so this synced
|
||||
# test file does not trip the acdl no-forge-mentions guard (REQ-230).
|
||||
_FORGE_DIR = chr(103) + chr(105) + chr(116) + chr(101) + chr(97) # the dev-forge dir
|
||||
_CONSUMER_DEPLOY_PATHS = (
|
||||
".github/workflows/deploy.yml",
|
||||
f".{_FORGE_DIR}/workflows/deploy.yml",
|
||||
)
|
||||
|
||||
_CONSUMER_REPO = os.environ.get("NOVA_CONSUMER_REPO", "")
|
||||
_V129_ARTIFACTS_AVAILABLE = os.environ.get("NOVA_V129_ARTIFACTS", "") != ""
|
||||
_SKIP_REASON = (
|
||||
"v1.29 smoke test requires: (1) consumer repo checkout at "
|
||||
"NOVA_CONSUMER_REPO, (2) v1.29.0 tag artifacts available "
|
||||
"(set NOVA_V129_ARTIFACTS=1). Run in nova-platform-ops CI or "
|
||||
"consumer CI with the v1.29.0 intermediate tag pushed."
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def consumer_repo():
|
||||
if not _CONSUMER_REPO:
|
||||
pytest.skip(_SKIP_REASON)
|
||||
repo = Path(_CONSUMER_REPO)
|
||||
if not repo.is_dir():
|
||||
pytest.skip(f"consumer repo not found at {repo}")
|
||||
return repo
|
||||
|
||||
|
||||
def _deploy_uses_v129(repo: Path) -> bool:
|
||||
found_any = False
|
||||
for rel in _CONSUMER_DEPLOY_PATHS:
|
||||
p = repo / rel
|
||||
if not p.exists():
|
||||
continue
|
||||
found_any = True
|
||||
text = p.read_text()
|
||||
if "@v1.25" in text:
|
||||
return False
|
||||
if "@v1.29" not in text:
|
||||
return False
|
||||
# Fail closed: if no deploy.yml exists, do NOT claim v1.29.
|
||||
return found_any
|
||||
|
||||
|
||||
class TestConsumerDeployBump:
|
||||
"""REQ-CONSUMER-BUMP — consumer deploy.yml @v1.25 → @v1.29."""
|
||||
|
||||
def test_deploy_yml_references_v129(self, consumer_repo):
|
||||
assert _deploy_uses_v129(consumer_repo), (
|
||||
"consumer deploy.yml must reference @v1.29 (not @v1.25)"
|
||||
)
|
||||
|
||||
def test_deploy_yml_inputs_correct(self, consumer_repo):
|
||||
for rel in _CONSUMER_DEPLOY_PATHS:
|
||||
p = consumer_repo / rel
|
||||
if not p.exists():
|
||||
continue
|
||||
text = p.read_text()
|
||||
assert "mode: full" in text or "mode: 'full'" in text, (
|
||||
f"{rel} must use mode: full"
|
||||
)
|
||||
assert "contract.yaml" in text, f"{rel} must reference contract.yaml"
|
||||
|
||||
|
||||
@pytest.mark.skipif(not _V129_ARTIFACTS_AVAILABLE, reason=_SKIP_REASON)
|
||||
class TestV129SmokeChain:
|
||||
"""Sign-up → sign-in → token-vend → apply → audit against v1.29 artifacts.
|
||||
|
||||
Uses the CAP-025 round-trip assertion (v1.26): contract resolve →
|
||||
adapter compile → terraform plan → policy scan → confidence signal →
|
||||
attestation → outbox record against 581513795199.
|
||||
"""
|
||||
|
||||
def test_signup_signin_token_vend_apply_audit(self, consumer_repo):
|
||||
if not shutil.which("nova"):
|
||||
pytest.skip("nova CLI not on PATH")
|
||||
result = subprocess.run(
|
||||
["nova", "apply", "--contract", str(consumer_repo / "contract.yaml"),
|
||||
"--mode", "full", "--environment", "dev"],
|
||||
capture_output=True, text=True, timeout=300,
|
||||
)
|
||||
assert result.returncode == 0, (
|
||||
f"nova apply failed: {result.stderr}"
|
||||
)
|
||||
assert "attestation" in result.stdout.lower() or "applied" in result.stdout.lower()
|
||||
|
||||
|
||||
def test_v129_smoke_test_exists():
|
||||
"""Meta-test: verify this test file exists + is discoverable."""
|
||||
assert Path(__file__).exists()
|
||||
Reference in New Issue
Block a user