feat(P18): testing + CI/CD pipelines - pytest suite, check-only mode, Gitea + GitHub workflows (v1.3.2)
90 offline tests covering adapter, confidence_signal, checkov_adapter, outbox_writer, and pipeline integration. Identical CI/CD workflows for Gitea Actions (dev) and GitHub Actions (production). New --check-only mode for run_platform.sh (offline, no AWS). ---ci--- project: acdl phase: 18 milestone: v1.3 status: verify ---/ci---
This commit is contained in:
Jon Chery
@@ -98,6 +98,17 @@
|
||||
- **REQ-37:** A `modules-ir/README-TEMPLATE.md` exists that works for both L1 and L2 modules, written in plain language (no jargon), with sections for Overview, Resources, Inputs, Outputs, Usage, Compliance extension points, and Versioning.
|
||||
- **REQ-38:** Every module has a `README.md`: the 7 L1 modules have full READMEs with Resources/Inputs/Outputs/Usage/Compliance-extension-points/Versioning sections derived from their `interface.json`; the 2 L2 modules have placeholder READMEs noting the composition is under redesign. A `modules-ir/README.md` catalog index lists all modules with one-line descriptions and links.
|
||||
|
||||
### Category: Testing
|
||||
- **REQ-39:** A pytest test suite exists under `tests/` covering the platform components offline (no AWS, no Checkov, no DynamoDB): the Terraform adapter (`adapters/terraform/adapter.py`), the confidence signal (`acdl_platform/confidence_signal.py`), the Checkov adapter (`adapters/terraform/policy/checkov_adapter.py`), and the outbox writer (`acdl_platform/outbox_writer.py`). The suite validates the IR schema, registry, spike_instance, and adapter output structure. `pyproject.toml` + `requirements-test.txt` pin test dependencies (pytest, jsonschema, pyyaml, boto3-stubs or moto for outbox mocking).
|
||||
|
||||
### Category: Shell Reproducibility
|
||||
- **REQ-40:** `scripts/run_platform.sh` has a `--check-only` mode that runs offline: loads the pre-existing IR instance, runs the adapter to emit Terraform, validates the JSON structure — without AWS credentials, Checkov, or DynamoDB. The existing `--plan-only` and full modes continue to require AWS. The `--check-only` mode is what CI pipelines run.
|
||||
|
||||
### Category: CI/CD Pipelines
|
||||
- **REQ-41:** Identical CI/CD pipelines exist for both Gitea Actions (`.gitea/workflows/ci.yml`, dev environment) and GitHub Actions (`.github/workflows/ci.yml`, production). Both run the same three stages: (1) lint — `py_compile` all Python files, (2) test — `pytest`, (3) check-only — `bash scripts/run_platform.sh --check-only`. Both trigger on push to main + pull request. Both use `ubuntu-latest`. Identical outcomes — the only difference is the runner environment.
|
||||
|
||||
- **REQ-42:** `pyproject.toml` exists at the repo root with pytest configuration (testpaths, markers) and the project metadata. `requirements-test.txt` pins test-only dependencies separate from runtime dependencies.
|
||||
|
||||
## Out of Scope (v1.2)
|
||||
|
||||
| REQ | Original criterion | Clarified criterion (effective) | Decision |
|
||||
@@ -174,4 +185,8 @@
|
||||
|-------------|-------|--------|
|
||||
| REQ-36 | 17 | complete (v1.3.1) |
|
||||
| REQ-37 | 17 | complete (v1.3.1) |
|
||||
| REQ-38 | 17 | complete (v1.3.1) |
|
||||
| REQ-38 | 17 | complete (v1.3.1) |
|
||||
| REQ-39 | 18 | complete (v1.3.2) |
|
||||
| REQ-40 | 18 | complete (v1.3.2) |
|
||||
| REQ-41 | 18 | complete (v1.3.2) |
|
||||
| REQ-42 | 18 | complete (v1.3.2) |
|
||||
+12
-1
@@ -238,4 +238,15 @@ L2 composition mechanism will be redesigned in a later phase.
|
||||
- A README-TEMPLATE.md exists for both L1 and L2 modules.
|
||||
- Every L1 module has a README.md with Resources/Inputs/Outputs/Usage/Compliance-extension-points/Versioning.
|
||||
- Every L2 module has a placeholder README.md noting the composition is under redesign.
|
||||
- A modules-ir/README.md catalog index exists.
|
||||
- A modules-ir/README.md catalog index exists.
|
||||
|
||||
### Phase 18 — testing-and-cicd-pipelines
|
||||
- **Description:** Create a pytest test suite that reproduces the platform pipeline offline (adapter, confidence_signal, checkov_adapter, outbox_writer). Add an offline `--check-only` mode to `run_platform.sh` that runs the pipeline up to adapter emission without AWS/Checkov/outbox. Create identical CI/CD pipelines for both Gitea Actions (`.gitea/workflows/ci.yml`, dev environment) and GitHub Actions (`.github/workflows/ci.yml`, production) that run: lint, pytest, `run_platform.sh --check-only`. Add `pyproject.toml` + `requirements-test.txt` for dependency pinning.
|
||||
- **Status:** complete (v1.3.2)
|
||||
- **Depends on:** [17]
|
||||
- **Requirements:** REQ-39, REQ-40, REQ-41, REQ-42
|
||||
- **Success Criteria:**
|
||||
- `pytest` runs and passes offline (no AWS, no Checkov, no DynamoDB).
|
||||
- `run_platform.sh --check-only` runs offline and exits 0.
|
||||
- `.gitea/workflows/ci.yml` and `.github/workflows/ci.yml` exist with identical job stages (lint, test, check-only).
|
||||
- `pyproject.toml` + `requirements-test.txt` pin test dependencies.
|
||||
+29
-31
@@ -1,47 +1,45 @@
|
||||
# Phase 17 — Verify (v1.3.1)
|
||||
# Phase 18 — Verify (v1.3.2)
|
||||
|
||||
## Structural
|
||||
|
||||
All 6 deleted files confirmed gone. All 7 kept files confirmed present.
|
||||
All 11 new files confirmed present. Registry has 7 L1 entries, 0 L2
|
||||
entries. `contracts/` directory deleted. L2 directories kept as
|
||||
placeholders with READMEs. **PASS.**
|
||||
All 11 new files confirmed present: pyproject.toml, requirements-test.txt,
|
||||
tests/__init__.py, tests/conftest.py, tests/test_adapter.py,
|
||||
tests/test_confidence_signal.py, tests/test_checkov_adapter.py,
|
||||
tests/test_outbox_writer.py, tests/test_pipeline.py,
|
||||
.gitea/workflows/ci.yml, .github/workflows/ci.yml. **PASS.**
|
||||
|
||||
## Behavioral
|
||||
|
||||
- All remaining Python files compile (`py_compile` OK).
|
||||
- `scripts/run_platform.sh` syntax valid; loads pre-existing IR instance
|
||||
(`modules-ir/l1/l1-s3/spike_instance.json`); no code calls to
|
||||
`contract_resolver.py` (only a documentation comment noting removal).
|
||||
- Adapter successfully compiles the pre-existing IR instance to
|
||||
Terraform (`main.tf` + `terraform.tf` + `providers.tf` emitted).
|
||||
- `registry.json` is valid JSON with only L1 entries.
|
||||
- `schemas/ir.schema.json` is valid JSON.
|
||||
- `spike_instance.json` validates against `ir.schema.json`.
|
||||
- No dangling references to deleted files in active code (only the
|
||||
documentation comment in `run_platform.sh` and historical
|
||||
`.ciagent/` + `verify_phaseNN.sh` files, which are expected).
|
||||
- `py_compile` passes on all Python files. **PASS.**
|
||||
- `pytest` — 90 tests, all passing, all offline (moto for DynamoDB
|
||||
mocking). **PASS.**
|
||||
- `run_platform.sh --check-only` — exits 0, outputs
|
||||
"PLATFORM CHECK OK", requires no AWS credentials. **PASS.**
|
||||
- `run_platform.sh --plan-only` — syntax valid (unchanged from phase 17).
|
||||
**PASS.**
|
||||
- Both workflow YAMLs are valid YAML, parseable. **PASS.**
|
||||
- Workflows are byte-identical (diff confirms). **PASS.**
|
||||
|
||||
## Security
|
||||
|
||||
- No secrets in new files. **PASS.**
|
||||
- P1 (deferred): `modules-ir/l1/l1-ecs-service/README.md` usage example
|
||||
contains the AWS account ID `581513795199` (same as the existing
|
||||
`terraform/microservice/main.tf`). Not a new leak — the account ID was
|
||||
already in the repo. Recommend replacing with a placeholder in a
|
||||
future docs pass.
|
||||
- No secrets in any new file (tests, workflows, pyproject, requirements).
|
||||
**PASS.**
|
||||
- CI pipelines do not use any AWS credentials — `--check-only` is fully
|
||||
offline. **PASS.**
|
||||
|
||||
## Quality
|
||||
|
||||
- README template has all required sections (Overview, Resources,
|
||||
Inputs, Outputs, Usage, Compliance extension points, Versioning)
|
||||
after adding the `## Overview` header. **PASS.**
|
||||
- All 7 L1 READMEs have all required sections. **PASS.**
|
||||
- Both L2 READMEs have placeholder notes mentioning redesign. **PASS.**
|
||||
- Catalog index lists all 7 L1s + 2 L2s. **PASS.**
|
||||
- pyproject.toml has pytest config (testpaths, markers, addopts).
|
||||
**PASS.**
|
||||
- requirements-test.txt pins all test deps. **PASS.**
|
||||
- Test suite covers all 4 platform components (adapter, confidence
|
||||
signal, checkov adapter, outbox writer) + pipeline integration.
|
||||
**PASS.**
|
||||
- Both workflows run 3 stages: lint, test, check-only. **PASS.**
|
||||
- README updated with "Test the platform" section + CI/CD documentation.
|
||||
**PASS.**
|
||||
|
||||
## Verdict
|
||||
|
||||
**VERIFY PASS** — all four layers pass. One P1 (account ID in usage
|
||||
example) deferred to post-hoc review.
|
||||
**VERIFY PASS** — all four layers pass. 90 offline tests, no AWS
|
||||
required for CI.
|
||||
@@ -0,0 +1,59 @@
|
||||
name: acdl-ci
|
||||
|
||||
on:
|
||||
push:
|
||||
branches: [main]
|
||||
pull_request:
|
||||
branches: [main]
|
||||
|
||||
jobs:
|
||||
lint:
|
||||
name: Lint
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
|
||||
- uses: actions/setup-python@v5
|
||||
with:
|
||||
python-version: "3.12"
|
||||
|
||||
- name: Compile all Python files
|
||||
run: |
|
||||
python3 -m py_compile \
|
||||
acdl_platform/confidence_signal.py \
|
||||
acdl_platform/outbox_writer.py \
|
||||
adapters/terraform/adapter.py \
|
||||
adapters/terraform/policy/checkov_adapter.py \
|
||||
scripts/push_consumer_image.py
|
||||
|
||||
test:
|
||||
name: Test
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
|
||||
- uses: actions/setup-python@v5
|
||||
with:
|
||||
python-version: "3.12"
|
||||
|
||||
- name: Install test dependencies
|
||||
run: pip install -r requirements-test.txt
|
||||
|
||||
- name: Run pytest
|
||||
run: python3 -m pytest tests/ -v --tb=short
|
||||
|
||||
check-only:
|
||||
name: Platform check-only (offline)
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
|
||||
- uses: actions/setup-python@v5
|
||||
with:
|
||||
python-version: "3.12"
|
||||
|
||||
- name: Install runtime dependencies
|
||||
run: pip install jsonschema pyyaml boto3
|
||||
|
||||
- name: Run platform check-only
|
||||
run: bash scripts/run_platform.sh --check-only
|
||||
@@ -0,0 +1,59 @@
|
||||
name: acdl-ci
|
||||
|
||||
on:
|
||||
push:
|
||||
branches: [main]
|
||||
pull_request:
|
||||
branches: [main]
|
||||
|
||||
jobs:
|
||||
lint:
|
||||
name: Lint
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
|
||||
- uses: actions/setup-python@v5
|
||||
with:
|
||||
python-version: "3.12"
|
||||
|
||||
- name: Compile all Python files
|
||||
run: |
|
||||
python3 -m py_compile \
|
||||
acdl_platform/confidence_signal.py \
|
||||
acdl_platform/outbox_writer.py \
|
||||
adapters/terraform/adapter.py \
|
||||
adapters/terraform/policy/checkov_adapter.py \
|
||||
scripts/push_consumer_image.py
|
||||
|
||||
test:
|
||||
name: Test
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
|
||||
- uses: actions/setup-python@v5
|
||||
with:
|
||||
python-version: "3.12"
|
||||
|
||||
- name: Install test dependencies
|
||||
run: pip install -r requirements-test.txt
|
||||
|
||||
- name: Run pytest
|
||||
run: python3 -m pytest tests/ -v --tb=short
|
||||
|
||||
check-only:
|
||||
name: Platform check-only (offline)
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
|
||||
- uses: actions/setup-python@v5
|
||||
with:
|
||||
python-version: "3.12"
|
||||
|
||||
- name: Install runtime dependencies
|
||||
run: pip install jsonschema pyyaml boto3
|
||||
|
||||
- name: Run platform check-only
|
||||
run: bash scripts/run_platform.sh --check-only
|
||||
@@ -110,15 +110,41 @@ ACDL_BOOTSTRAP_AWS_ACCESS_KEY_ID=... ACDL_BOOTSTRAP_AWS_SECRET_ACCESS_KEY=... \
|
||||
ACDL_BOOTSTRAP_AWS_ACCESS_KEY_ID=... ACDL_BOOTSTRAP_AWS_SECRET_ACCESS_KEY=... \
|
||||
bash scripts/rotate_spike_key.sh
|
||||
|
||||
# 3. Run the full platform pipeline (contract -> IR -> plan -> Checkov ->
|
||||
# 3. Run the full platform pipeline (IR -> adapter -> plan -> Checkov ->
|
||||
# confidence -> outbox)
|
||||
bash scripts/run_platform.sh
|
||||
# Expected: "=== PLATFORM E2E OK ==="
|
||||
|
||||
# Or plan-only (contract -> IR -> terraform plan; no Checkov/outbox):
|
||||
# Or plan-only (IR -> adapter -> terraform plan; no Checkov/outbox):
|
||||
bash scripts/run_platform.sh --plan-only
|
||||
```
|
||||
|
||||
### Test the platform (offline, no AWS required)
|
||||
|
||||
```bash
|
||||
# Install test dependencies
|
||||
pip install -r requirements-test.txt
|
||||
|
||||
# Run the test suite (90 tests, all offline — uses moto for DynamoDB mocking)
|
||||
python3 -m pytest tests/ -v
|
||||
|
||||
# Run the platform in check-only mode (offline — no AWS, no Checkov, no outbox)
|
||||
bash scripts/run_platform.sh --check-only
|
||||
# Expected: "=== PLATFORM CHECK OK ==="
|
||||
```
|
||||
|
||||
### CI/CD pipelines
|
||||
|
||||
Identical pipelines run on both Gitea Actions (dev) and GitHub Actions
|
||||
(production):
|
||||
|
||||
- `.gitea/workflows/ci.yml` — Gitea Actions (dev environment)
|
||||
- `.github/workflows/ci.yml` — GitHub Actions (production)
|
||||
|
||||
Both run three stages: **lint** (py_compile), **test** (pytest), and
|
||||
**check-only** (`run_platform.sh --check-only`). Both trigger on push to
|
||||
`main` and on pull requests.
|
||||
|
||||
### Re-run the archived v1.0 demo (stubs only, no AWS)
|
||||
|
||||
```bash
|
||||
|
||||
@@ -0,0 +1,31 @@
|
||||
[project]
|
||||
name = "acdl"
|
||||
version = "1.3.0"
|
||||
description = "Agentic Cloud Delivery Platform — consumers declare intent; the platform delivers safe production deployment."
|
||||
requires-python = ">=3.10"
|
||||
dependencies = [
|
||||
"boto3>=1.34",
|
||||
"jsonschema>=4.20",
|
||||
"pyyaml>=6.0",
|
||||
]
|
||||
|
||||
[project.optional-dependencies]
|
||||
test = [
|
||||
"pytest>=8.0",
|
||||
"pytest-cov>=4.0",
|
||||
"moto[dynamodb]>=5.0",
|
||||
]
|
||||
|
||||
[tool.pytest.ini_options]
|
||||
testpaths = ["tests"]
|
||||
markers = [
|
||||
"offline: tests that run without AWS/Checkov/DynamoDB",
|
||||
]
|
||||
addopts = "-v --tb=short"
|
||||
|
||||
[tool.coverage]
|
||||
run.source = ["acdl_platform", "adapters"]
|
||||
|
||||
[build-system]
|
||||
requires = ["setuptools>=68"]
|
||||
build-backend = "setuptools.backends._legacy:_Backend"
|
||||
@@ -0,0 +1,6 @@
|
||||
pytest>=8.0
|
||||
pytest-cov>=4.0
|
||||
moto[dynamodb]>=5.0
|
||||
jsonschema>=4.20
|
||||
pyyaml>=6.0
|
||||
boto3>=1.34
|
||||
+47
-14
@@ -1,11 +1,13 @@
|
||||
#!/usr/bin/env bash
|
||||
# scripts/run_platform.sh - the ACDL platform pipeline.
|
||||
#
|
||||
# Default: full end-to-end pipeline (load pre-existing IR instance ->
|
||||
# adapter -> terraform plan (real AWS) -> Checkov -> PolicyCheckResult ->
|
||||
# confidence signal -> evidence event to DynamoDB outbox).
|
||||
# --plan-only: load IR + adapter + terraform init/validate/plan (steps
|
||||
# 1-4), then exit.
|
||||
# Modes:
|
||||
# --check-only (offline, no AWS/Checkov/DynamoDB — for CI)
|
||||
# load IR -> adapter -> validate output structure -> exit 0
|
||||
# --plan-only (requires AWS creds, no Checkov/outbox)
|
||||
# load IR -> adapter -> terraform init/validate/plan -> exit 0
|
||||
# (default) (requires AWS creds + Checkov + DynamoDB)
|
||||
# load IR -> adapter -> terraform plan -> Checkov -> confidence -> outbox
|
||||
#
|
||||
# NOTE: contract resolution (contract_resolver.py) was removed when the
|
||||
# thin-composition layer was taken out. The pipeline now starts from a
|
||||
@@ -18,9 +20,11 @@ set -euo pipefail
|
||||
ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
|
||||
cd "$ROOT"
|
||||
|
||||
CHECK_ONLY=0
|
||||
PLAN_ONLY=0
|
||||
for arg in "$@"; do
|
||||
case "$arg" in
|
||||
--check-only) CHECK_ONLY=1 ;;
|
||||
--plan-only) PLAN_ONLY=1 ;;
|
||||
*) echo "FAIL: unknown argument: $arg" >&2; exit 1 ;;
|
||||
esac
|
||||
@@ -28,15 +32,6 @@ done
|
||||
|
||||
fail() { echo "FAIL: $*" >&2; exit 1; }
|
||||
|
||||
ENV_FILE="$ROOT/.env.secrets"
|
||||
[ -f "$ENV_FILE" ] || fail ".env.secrets missing (run scripts/rotate_spike_key.sh)"
|
||||
set -a
|
||||
. "$ENV_FILE"
|
||||
set +a
|
||||
export AWS_ACCESS_KEY_ID="$ACDL_AWS_ACCESS_KEY_ID"
|
||||
export AWS_SECRET_ACCESS_KEY="$ACDL_AWS_SECRET_ACCESS_KEY"
|
||||
export AWS_DEFAULT_REGION="$AWS_DEFAULT_REGION"
|
||||
|
||||
CONTRACT_ID="11111111-1111-1111-1111-111111111111" # spike fixed UUID
|
||||
WORK="/tmp/spike_e2e"
|
||||
rm -rf "$WORK"; mkdir -p "$WORK"
|
||||
@@ -51,6 +46,44 @@ echo "=== Step 3: adapter compiles IR -> terraform/spike/*.tf (regenerate) ==="
|
||||
python3 adapters/terraform/adapter.py "$WORK/spike_ir.json" terraform/spike || fail "adapter failed"
|
||||
echo "adapter: emitted terraform/spike/{main.tf,terraform.tf,providers.tf}"
|
||||
|
||||
if [ "$CHECK_ONLY" = "1" ]; then
|
||||
echo ""
|
||||
echo "=== Step 3b: validate adapter output structure (offline) ==="
|
||||
python3 -c "
|
||||
import json, os
|
||||
d = json.load(open('$WORK/spike_ir.json'))
|
||||
assert d['stack']['name'] == 'l1-s3'
|
||||
assert len(d['resources']) == 1
|
||||
tf_dir = 'terraform/spike'
|
||||
for f in ('main.tf', 'terraform.tf', 'providers.tf'):
|
||||
assert os.path.isfile(os.path.join(tf_dir, f)), f'{f} missing'
|
||||
main = open(os.path.join(tf_dir, 'main.tf')).read()
|
||||
assert 'aws_s3_bucket' in main
|
||||
assert 'acdl-spike-bucket' in main
|
||||
assert 'versioning' in main
|
||||
tf = open(os.path.join(tf_dir, 'terraform.tf')).read()
|
||||
assert 'backend' in tf
|
||||
assert 'required_version' in tf
|
||||
prov = open(os.path.join(tf_dir, 'providers.tf')).read()
|
||||
assert 'provider \"aws\"' in prov
|
||||
print('adapter output: OK')
|
||||
"
|
||||
echo ""
|
||||
echo "=== PLATFORM CHECK OK ==="
|
||||
echo "IR instance -> adapter -> structure validated (offline, no AWS)"
|
||||
exit 0
|
||||
fi
|
||||
|
||||
echo "=== Loading AWS credentials (not needed for --check-only) ==="
|
||||
ENV_FILE="$ROOT/.env.secrets"
|
||||
[ -f "$ENV_FILE" ] || fail ".env.secrets missing (run scripts/rotate_spike_key.sh)"
|
||||
set -a
|
||||
. "$ENV_FILE"
|
||||
set +a
|
||||
export AWS_ACCESS_KEY_ID="$ACDL_AWS_ACCESS_KEY_ID"
|
||||
export AWS_SECRET_ACCESS_KEY="$ACDL_AWS_SECRET_ACCESS_KEY"
|
||||
export AWS_DEFAULT_REGION="$AWS_DEFAULT_REGION"
|
||||
|
||||
echo "=== Step 4: terraform init + validate + plan -lock=false (real AWS) ==="
|
||||
cd terraform/spike
|
||||
terraform init -reconfigure -lock=false -input=false >> "$WORK/tf.log" 2>&1 || fail "terraform init failed"
|
||||
|
||||
@@ -8,7 +8,7 @@ terraform {
|
||||
}
|
||||
backend "s3" {
|
||||
bucket = "acdl-tfstate-581513795199-us-east-1"
|
||||
key = "spike/l2-static-asset/terraform.tfstate"
|
||||
key = "spike/l1-s3/terraform.tfstate"
|
||||
region = "us-east-1"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,34 @@
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def repo_root():
|
||||
return str(ROOT)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def spike_ir():
|
||||
return json.load(open(ROOT / "modules-ir/l1/l1-s3/spike_instance.json"))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def ir_schema():
|
||||
return json.load(open(ROOT / "schemas/ir.schema.json"))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def registry():
|
||||
return json.load(open(ROOT / "modules-ir/registry.json"))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def policy_check_result_schema():
|
||||
return json.load(open(ROOT / "schemas/policy_check_result.schema.json"))
|
||||
@@ -0,0 +1,174 @@
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import jsonschema
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from adapters.terraform.adapter import (
|
||||
TYPE_MAP, INPUT_MAP, OUTPUT_MAP, adapt, _tf_value, _ref_expr,
|
||||
)
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
|
||||
|
||||
class TestSpikeInstance:
|
||||
def test_spike_instance_validates_against_ir_schema(self, spike_ir, ir_schema):
|
||||
jsonschema.validate(spike_ir, ir_schema)
|
||||
|
||||
def test_spike_instance_has_one_resource(self, spike_ir):
|
||||
assert len(spike_ir["resources"]) == 1
|
||||
r = spike_ir["resources"][0]
|
||||
assert r["id"] == "s3"
|
||||
assert r["type"] == "aws:s3:bucket"
|
||||
|
||||
def test_spike_instance_stack_is_l1_s3(self, spike_ir):
|
||||
assert spike_ir["stack"]["name"] == "l1-s3"
|
||||
assert spike_ir["stack"]["kind"] == "l1"
|
||||
|
||||
|
||||
class TestRegistry:
|
||||
def test_registry_has_7_l1_entries(self, registry):
|
||||
assert len(registry) == 7
|
||||
for key in registry:
|
||||
assert key.startswith("l1-")
|
||||
|
||||
def test_registry_has_no_l2_entries(self, registry):
|
||||
l2 = [k for k in registry if k.startswith("l2")]
|
||||
assert l2 == []
|
||||
|
||||
def test_all_l1_interfaces_exist(self, registry, repo_root):
|
||||
for name, versions in registry.items():
|
||||
for ver, entry in versions.items():
|
||||
iface_path = os.path.join(repo_root, entry["interface"])
|
||||
assert os.path.isfile(iface_path), f"{iface_path} missing"
|
||||
iface = json.load(open(iface_path))
|
||||
assert iface["name"] == name
|
||||
|
||||
|
||||
class TestTypeMap:
|
||||
def test_s3_in_type_map(self):
|
||||
assert TYPE_MAP["aws:s3:bucket"] == "aws_s3_bucket"
|
||||
|
||||
def test_vpc_types_in_type_map(self):
|
||||
assert TYPE_MAP["aws:ec2:vpc"] == "aws_vpc"
|
||||
assert TYPE_MAP["aws:ec2:subnet"] == "aws_subnet"
|
||||
assert TYPE_MAP["aws:ec2:routetable"] == "aws_route_table"
|
||||
|
||||
def test_ecs_types_in_type_map(self):
|
||||
assert TYPE_MAP["aws:ecs:cluster"] == "aws_ecs_cluster"
|
||||
assert TYPE_MAP["aws:ecs:task_definition"] == "aws_ecs_task_definition"
|
||||
assert TYPE_MAP["aws:ecs:service"] == "aws_ecs_service"
|
||||
|
||||
def test_alb_types_in_type_map(self):
|
||||
assert TYPE_MAP["aws:elbv2:loadbalancer"] == "aws_lb"
|
||||
assert TYPE_MAP["aws:elbv2:listener"] == "aws_lb_listener"
|
||||
assert TYPE_MAP["aws:elbv2:targetgroup"] == "aws_lb_target_group"
|
||||
|
||||
def test_iam_and_ecr_in_type_map(self):
|
||||
assert TYPE_MAP["aws:iam:role"] == "aws_iam_role"
|
||||
assert TYPE_MAP["aws:ecr:repository"] == "aws_ecr_repository"
|
||||
|
||||
|
||||
class TestTfValue:
|
||||
def test_string_quoted(self):
|
||||
assert _tf_value("hello") == '"hello"'
|
||||
|
||||
def test_bool_true(self):
|
||||
assert _tf_value(True) == "true"
|
||||
|
||||
def test_bool_false(self):
|
||||
assert _tf_value(False) == "false"
|
||||
|
||||
def test_int(self):
|
||||
assert _tf_value(42) == "42"
|
||||
|
||||
def test_float(self):
|
||||
assert _tf_value(3.14) == "3.14"
|
||||
|
||||
def test_dict_jsonencoded(self):
|
||||
result = _tf_value({"key": "val"})
|
||||
assert "jsonencode" in result
|
||||
assert '"key"' in result
|
||||
|
||||
def test_list_jsonencoded(self):
|
||||
result = _tf_value([1, 2])
|
||||
assert "jsonencode" in result
|
||||
|
||||
def test_json_string_jsonencoded(self):
|
||||
result = _tf_value('{"k":"v"}')
|
||||
assert "jsonencode" in result
|
||||
|
||||
def test_ref_raises(self):
|
||||
with pytest.raises(ValueError, match="ref: values"):
|
||||
_tf_value("ref:s3.bucket_arn")
|
||||
|
||||
|
||||
class TestRefExpr:
|
||||
def test_basic_ref(self):
|
||||
type_by_id = {"s3": "aws:s3:bucket"}
|
||||
result = _ref_expr("ref:s3.bucket_arn", type_by_id)
|
||||
assert result == "aws_s3_bucket.s3.arn"
|
||||
|
||||
def test_vpc_ref(self):
|
||||
type_by_id = {"vpc": "aws:ec2:vpc"}
|
||||
result = _ref_expr("ref:vpc.vpc_id", type_by_id)
|
||||
assert result == "aws_vpc.vpc.id"
|
||||
|
||||
def test_unknown_id_raises(self):
|
||||
with pytest.raises(ValueError, match="unknown IR resource id"):
|
||||
_ref_expr("ref:nonexistent.output", {"s3": "aws:s3:bucket"})
|
||||
|
||||
|
||||
class TestAdapt:
|
||||
def test_adapt_emits_three_files(self, spike_ir, tmp_path):
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(spike_ir, out_dir)
|
||||
assert os.path.isfile(os.path.join(out_dir, "main.tf"))
|
||||
assert os.path.isfile(os.path.join(out_dir, "terraform.tf"))
|
||||
assert os.path.isfile(os.path.join(out_dir, "providers.tf"))
|
||||
|
||||
def test_main_tf_has_s3_bucket(self, spike_ir, tmp_path):
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(spike_ir, out_dir)
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert 'resource "aws_s3_bucket" "s3"' in main_tf
|
||||
assert 'bucket = "acdl-spike-bucket"' in main_tf
|
||||
|
||||
def test_main_tf_has_versioning(self, spike_ir, tmp_path):
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(spike_ir, out_dir)
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert "versioning" in main_tf
|
||||
assert "enabled = true" in main_tf
|
||||
|
||||
def test_main_tf_has_outputs(self, spike_ir, tmp_path):
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(spike_ir, out_dir)
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert 'output "bucket_arn"' in main_tf
|
||||
assert 'output "bucket_name"' in main_tf
|
||||
|
||||
def test_terraform_tf_has_backend(self, spike_ir, tmp_path):
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(spike_ir, out_dir)
|
||||
terraform_tf = open(os.path.join(out_dir, "terraform.tf")).read()
|
||||
assert 'backend "s3"' in terraform_tf
|
||||
assert 'required_version' in terraform_tf
|
||||
assert ">= 1.9" in terraform_tf
|
||||
|
||||
def test_providers_tf_has_aws(self, spike_ir, tmp_path):
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(spike_ir, out_dir)
|
||||
providers_tf = open(os.path.join(out_dir, "providers.tf")).read()
|
||||
assert 'provider "aws"' in providers_tf
|
||||
assert "us-east-1" in providers_tf
|
||||
|
||||
def test_backend_key_uses_stack_name(self, spike_ir, tmp_path):
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(spike_ir, out_dir)
|
||||
terraform_tf = open(os.path.join(out_dir, "terraform.tf")).read()
|
||||
assert "spike/l1-s3/terraform.tfstate" in terraform_tf
|
||||
@@ -0,0 +1,126 @@
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import jsonschema
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from adapters.terraform.policy.checkov_adapter import (
|
||||
RULE_MAP, _to_pcr, _emit_tag_naming_skipped, adapt,
|
||||
)
|
||||
|
||||
|
||||
class TestRuleMap:
|
||||
def test_secrets_rules(self):
|
||||
assert RULE_MAP["CKV_AWS_41"][0] == "secrets-in-plaintext"
|
||||
assert RULE_MAP["CKV_AWS_45"][0] == "secrets-in-plaintext"
|
||||
|
||||
def test_public_ingress_rules(self):
|
||||
assert RULE_MAP["CKV_AWS_20"][0] == "public-ingress"
|
||||
assert RULE_MAP["CKV_AWS_57"][0] == "public-ingress"
|
||||
|
||||
def test_iam_wildcard(self):
|
||||
assert RULE_MAP["CKV_AWS_1"][0] == "iam-wildcard"
|
||||
|
||||
def test_kms(self):
|
||||
assert RULE_MAP["CKV_AWS_7"][0] == "kms-key-reference"
|
||||
|
||||
def test_all_have_severities(self):
|
||||
for rule_id, (cat, sev) in RULE_MAP.items():
|
||||
assert sev in ("high", "medium", "low", "info"), f"{rule_id} has bad severity {sev}"
|
||||
|
||||
|
||||
class TestToPcr:
|
||||
def test_passed_result(self):
|
||||
rec = {"check_id": "CKV_AWS_20", "check_name": "No public ingress", "file_path": "main.tf"}
|
||||
pcr = _to_pcr(rec, "contract-123", "PASSED")
|
||||
assert pcr["result"] == "pass"
|
||||
assert pcr["contractId"] == "contract-123"
|
||||
assert pcr["engine"] == "checkov"
|
||||
assert pcr["ruleId"] == "CKV_AWS_20"
|
||||
assert pcr["severity"] == "high"
|
||||
|
||||
def test_failed_result(self):
|
||||
rec = {"check_id": "CKV_AWS_1", "check_name": "No wildcard IAM"}
|
||||
pcr = _to_pcr(rec, "c-1", "FAILED")
|
||||
assert pcr["result"] == "fail"
|
||||
assert pcr["severity"] == "high"
|
||||
|
||||
def test_skipped_result(self):
|
||||
rec = {"check_id": "UNKNOWN_RULE", "check_name": "some check"}
|
||||
pcr = _to_pcr(rec, "c-1", "SKIPPED")
|
||||
assert pcr["result"] == "skipped"
|
||||
assert pcr["severity"] == "info"
|
||||
|
||||
def test_unknown_rule_defaults_to_info(self):
|
||||
rec = {"check_id": "UNKNOWN_RULE", "check_name": "unknown"}
|
||||
pcr = _to_pcr(rec, "c-1", "FAILED")
|
||||
assert pcr["severity"] == "info"
|
||||
|
||||
def test_pcr_validates_against_schema(self, policy_check_result_schema):
|
||||
rec = {"check_id": "CKV_AWS_20", "check_name": "test", "file_path": "main.tf",
|
||||
"resource": "aws_s3_bucket.s3", "resource_address": "aws_s3_bucket.s3"}
|
||||
pcr = _to_pcr(rec, "c-1", "FAILED")
|
||||
jsonschema.validate(pcr, policy_check_result_schema)
|
||||
|
||||
|
||||
class TestTagNamingSkipped:
|
||||
def test_skipped_pcr(self):
|
||||
pcr = _emit_tag_naming_skipped("c-1")
|
||||
assert pcr["result"] == "skipped"
|
||||
assert pcr["ruleId"] == "ACDL_TAG_NAMING"
|
||||
assert pcr["severity"] == "info"
|
||||
|
||||
|
||||
class TestAdapt:
|
||||
def _sample_checkov_json(self):
|
||||
return {
|
||||
"terraform_plan": {
|
||||
"results": {
|
||||
"passed_checks": [
|
||||
{"check_id": "CKV_AWS_20", "check_name": "no public ingress",
|
||||
"file_path": "main.tf", "resource": "aws_vpc.vpc"}
|
||||
],
|
||||
"failed_checks": [
|
||||
{"check_id": "CKV_AWS_1", "check_name": "no wildcard iam",
|
||||
"file_path": "main.tf", "resource": "aws_iam_role.r"}
|
||||
],
|
||||
"skipped_checks": []
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
def test_adapt_returns_list(self, tmp_path):
|
||||
data = self._sample_checkov_json()
|
||||
f = tmp_path / "checkov.json"
|
||||
f.write_text(json.dumps(data))
|
||||
results = adapt(str(f), "c-1")
|
||||
assert isinstance(results, list)
|
||||
|
||||
def test_adapt_includes_tag_naming(self, tmp_path):
|
||||
data = self._sample_checkov_json()
|
||||
f = tmp_path / "checkov.json"
|
||||
f.write_text(json.dumps(data))
|
||||
results = adapt(str(f), "c-1")
|
||||
tag = [r for r in results if r["ruleId"] == "ACDL_TAG_NAMING"]
|
||||
assert len(tag) == 1
|
||||
assert tag[0]["result"] == "skipped"
|
||||
|
||||
def test_adapt_has_passed_and_failed(self, tmp_path):
|
||||
data = self._sample_checkov_json()
|
||||
f = tmp_path / "checkov.json"
|
||||
f.write_text(json.dumps(data))
|
||||
results = adapt(str(f), "c-1")
|
||||
passed = [r for r in results if r["result"] == "pass"]
|
||||
failed = [r for r in results if r["result"] == "fail"]
|
||||
assert len(passed) >= 1
|
||||
assert len(failed) >= 1
|
||||
|
||||
def test_adapt_empty_input(self, tmp_path):
|
||||
data = {"terraform_plan": {"results": {"passed_checks": [], "failed_checks": [], "skipped_checks": []}}}
|
||||
f = tmp_path / "checkov.json"
|
||||
f.write_text(json.dumps(data))
|
||||
results = adapt(str(f), "c-1")
|
||||
assert len(results) == 1 # just the tag naming skipped
|
||||
@@ -0,0 +1,181 @@
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from acdl_platform.confidence_signal import (
|
||||
WEIGHTS, PENALTY, THRESHOLDS, compute, Signal, _per_input_score,
|
||||
)
|
||||
|
||||
|
||||
class TestWeights:
|
||||
def test_weights_sum_to_one(self):
|
||||
assert sum(WEIGHTS.values()) == pytest.approx(1.0)
|
||||
|
||||
def test_policy_weight_highest(self):
|
||||
assert WEIGHTS["policy"] == 0.30
|
||||
|
||||
def test_validation_weight(self):
|
||||
assert WEIGHTS["validation"] == 0.25
|
||||
|
||||
|
||||
class TestThresholds:
|
||||
def test_dev_threshold(self):
|
||||
assert THRESHOLDS["dev"] == 0.50
|
||||
|
||||
def test_qa_threshold(self):
|
||||
assert THRESHOLDS["qa"] == 0.75
|
||||
|
||||
def test_prod_threshold(self):
|
||||
assert THRESHOLDS["prod"] == 0.90
|
||||
|
||||
def test_dr_threshold(self):
|
||||
assert THRESHOLDS["dr"] == 0.95
|
||||
|
||||
|
||||
class TestPenalty:
|
||||
def test_critical_is_none(self):
|
||||
assert PENALTY["critical"] is None
|
||||
|
||||
def test_high_penalty(self):
|
||||
assert PENALTY["high"] == 0.20
|
||||
|
||||
def test_medium_penalty(self):
|
||||
assert PENALTY["medium"] == 0.05
|
||||
|
||||
def test_low_penalty(self):
|
||||
assert PENALTY["low"] == 0.01
|
||||
|
||||
def test_info_no_penalty(self):
|
||||
assert PENALTY["info"] == 0.0
|
||||
|
||||
|
||||
class TestPerInputScore:
|
||||
def test_missing_input_returns_half(self):
|
||||
score, reasons = _per_input_score("policy", None)
|
||||
assert score == 0.5
|
||||
assert "INPUT_MISSING:policy" in reasons
|
||||
|
||||
def test_empty_policy_list(self):
|
||||
score, reasons = _per_input_score("policy", [])
|
||||
assert score == 0.5
|
||||
assert reasons == []
|
||||
|
||||
def test_all_pass_policy(self):
|
||||
pcrs = [{"result": "pass"}, {"result": "pass"}]
|
||||
score, reasons = _per_input_score("policy", pcrs)
|
||||
assert score == 1.0
|
||||
assert reasons == []
|
||||
|
||||
def test_mixed_policy(self):
|
||||
pcrs = [{"result": "pass"}, {"result": "fail"}]
|
||||
score, reasons = _per_input_score("policy", pcrs)
|
||||
assert score == 0.5
|
||||
|
||||
def test_skipped_counts_as_pass(self):
|
||||
pcrs = [{"result": "skipped"}]
|
||||
score, reasons = _per_input_score("policy", pcrs)
|
||||
assert score == 1.0
|
||||
|
||||
def test_validation_all_true(self):
|
||||
score, reasons = _per_input_score("validation", {
|
||||
"schema": True, "ir_resolved": True,
|
||||
"tf_validated": True, "tf_planned": True
|
||||
})
|
||||
assert score == 1.0
|
||||
|
||||
def test_validation_partial(self):
|
||||
score, reasons = _per_input_score("validation", {
|
||||
"schema": True, "ir_resolved": True,
|
||||
"tf_validated": False, "tf_planned": False
|
||||
})
|
||||
assert score == 0.5
|
||||
|
||||
def test_freshness_fresh(self):
|
||||
score, _ = _per_input_score("freshness", {"age_days": 0, "max_age_days": 7})
|
||||
assert score == 1.0
|
||||
|
||||
def test_freshness_stale(self):
|
||||
score, _ = _per_input_score("freshness", {"age_days": 7, "max_age_days": 7})
|
||||
assert score == pytest.approx(0.0)
|
||||
|
||||
def test_source_complete(self):
|
||||
score, _ = _per_input_score("source", {"submitter": "dev", "commit_sha": "abc"})
|
||||
assert score == 1.0
|
||||
|
||||
def test_source_partial(self):
|
||||
score, _ = _per_input_score("source", {"submitter": "dev"})
|
||||
assert score == 0.5
|
||||
|
||||
def test_history_clean(self):
|
||||
score, _ = _per_input_score("history", {"prior_rollbacks": 0, "prior_policy_fails": 0})
|
||||
assert score == 1.0
|
||||
|
||||
def test_history_with_failures(self):
|
||||
score, _ = _per_input_score("history", {"prior_rollbacks": 2, "prior_policy_fails": 3})
|
||||
assert score == pytest.approx(0.3)
|
||||
|
||||
def test_nfrs_none(self):
|
||||
score, _ = _per_input_score("nfrs", {"conformance": None})
|
||||
assert score == 0.5
|
||||
|
||||
def test_nfrs_full(self):
|
||||
score, _ = _per_input_score("nfrs", {"conformance": 0.95})
|
||||
assert score == 0.95
|
||||
|
||||
|
||||
class TestCompute:
|
||||
def _base_inputs(self):
|
||||
return {
|
||||
"policy": [{"result": "pass"}],
|
||||
"validation": {"schema": True, "ir_resolved": True,
|
||||
"tf_validated": True, "tf_planned": True},
|
||||
"freshness": {"age_days": 0, "max_age_days": 7},
|
||||
"source": {"submitter": "dev", "commit_sha": "abc"},
|
||||
"history": {"prior_rollbacks": 0, "prior_policy_fails": 0},
|
||||
"nfrs": {"conformance": None},
|
||||
}
|
||||
|
||||
def test_dev_pass(self):
|
||||
sig = compute("test-001", "dev", self._base_inputs())
|
||||
assert sig.band == "pass"
|
||||
assert sig.score >= 0.50
|
||||
|
||||
def test_missing_input_blocks(self):
|
||||
inputs = self._base_inputs()
|
||||
del inputs["policy"]
|
||||
sig = compute("test-002", "dev", inputs)
|
||||
assert sig.band == "block"
|
||||
assert sig.score == 0.0
|
||||
assert any("INPUT_MISSING" in r for r in sig.reasonCodes)
|
||||
|
||||
def test_critical_policy_blocks(self):
|
||||
inputs = self._base_inputs()
|
||||
inputs["policy"] = [{"result": "fail", "severity": "critical", "ruleId": "CKV_X"}]
|
||||
sig = compute("test-003", "dev", inputs)
|
||||
assert sig.band == "block"
|
||||
assert sig.score == 0.0
|
||||
assert any("CRITICAL_OVERRIDE" in r for r in sig.reasonCodes)
|
||||
|
||||
def test_high_policy_lowers_score(self):
|
||||
inputs = self._base_inputs()
|
||||
inputs["policy"] = [{"result": "fail", "severity": "high", "ruleId": "CKV_Y"}]
|
||||
sig = compute("test-004", "dev", inputs)
|
||||
assert sig.score < 1.0
|
||||
|
||||
def test_dev_warn_becomes_block(self):
|
||||
sig = compute("test-005", "dev", self._base_inputs())
|
||||
assert sig.band != "warn"
|
||||
|
||||
def test_signal_has_per_input(self):
|
||||
sig = compute("test-006", "dev", self._base_inputs())
|
||||
assert "policy" in sig.perInput
|
||||
assert "validation" in sig.perInput
|
||||
assert "nfrs" in sig.perInput
|
||||
|
||||
def test_all_six_inputs_present(self):
|
||||
sig = compute("test-007", "dev", self._base_inputs())
|
||||
assert len(sig.perInput) == 6
|
||||
@@ -0,0 +1,135 @@
|
||||
import datetime
|
||||
import hashlib
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from acdl_platform.outbox_writer import _canonical_hash, write_event
|
||||
|
||||
|
||||
class TestCanonicalHash:
|
||||
def test_deterministic(self):
|
||||
event = {"b": 2, "a": 1}
|
||||
h1 = _canonical_hash(event)
|
||||
h2 = _canonical_hash(event)
|
||||
assert h1 == h2
|
||||
|
||||
def test_order_independent(self):
|
||||
h1 = _canonical_hash({"a": 1, "b": 2})
|
||||
h2 = _canonical_hash({"b": 2, "a": 1})
|
||||
assert h1 == h2
|
||||
|
||||
def test_is_sha256_hex(self):
|
||||
h = _canonical_hash({"key": "val"})
|
||||
assert len(h) == 64
|
||||
assert all(c in "0123456789abcdef" for c in h)
|
||||
|
||||
def test_different_events_different_hash(self):
|
||||
h1 = _canonical_hash({"a": 1})
|
||||
h2 = _canonical_hash({"a": 2})
|
||||
assert h1 != h2
|
||||
|
||||
|
||||
class TestWriteEvent:
|
||||
def _sample_event(self):
|
||||
return {
|
||||
"contractId": "test-contract-001",
|
||||
"eventType": "CONFIDENCE_COMPUTED",
|
||||
"ts": "2026-07-22T00:00:00Z",
|
||||
"environment": "dev",
|
||||
"stack": "l1-s3",
|
||||
"score": 0.85,
|
||||
"band": "pass",
|
||||
"prev_event_hash": "GENESIS",
|
||||
}
|
||||
|
||||
def test_write_event_with_mock_dynamodb(self):
|
||||
from moto import mock_aws
|
||||
|
||||
import boto3
|
||||
|
||||
with mock_aws():
|
||||
dyn = boto3.client("dynamodb", region_name="us-east-1")
|
||||
dyn.create_table(
|
||||
TableName="acdl-outbox",
|
||||
KeySchema=[
|
||||
{"AttributeName": "contractId", "KeyType": "HASH"},
|
||||
{"AttributeName": "eventType#eventTs", "KeyType": "RANGE"},
|
||||
],
|
||||
AttributeDefinitions=[
|
||||
{"AttributeName": "contractId", "AttributeType": "S"},
|
||||
{"AttributeName": "eventType#eventTs", "AttributeType": "S"},
|
||||
],
|
||||
BillingMode="PAY_PER_REQUEST",
|
||||
)
|
||||
|
||||
event = self._sample_event()
|
||||
item = write_event(event, outbox_table="acdl-outbox", region="us-east-1")
|
||||
|
||||
assert item["contractId"]["S"] == "test-contract-001"
|
||||
assert item["prev_event_hash"]["S"] == "GENESIS"
|
||||
assert "hash" in item
|
||||
assert len(item["hash"]["S"]) == 64
|
||||
assert "expire_at" in item
|
||||
|
||||
def test_write_event_hash_matches_canonical(self):
|
||||
from moto import mock_aws
|
||||
|
||||
import boto3
|
||||
|
||||
with mock_aws():
|
||||
dyn = boto3.client("dynamodb", region_name="us-east-1")
|
||||
dyn.create_table(
|
||||
TableName="acdl-outbox",
|
||||
KeySchema=[
|
||||
{"AttributeName": "contractId", "KeyType": "HASH"},
|
||||
{"AttributeName": "eventType#eventTs", "KeyType": "RANGE"},
|
||||
],
|
||||
AttributeDefinitions=[
|
||||
{"AttributeName": "contractId", "AttributeType": "S"},
|
||||
{"AttributeName": "eventType#eventTs", "AttributeType": "S"},
|
||||
],
|
||||
BillingMode="PAY_PER_REQUEST",
|
||||
)
|
||||
|
||||
event = self._sample_event()
|
||||
item = write_event(event, outbox_table="acdl-outbox", region="us-east-1")
|
||||
expected_hash = _canonical_hash(event)
|
||||
assert item["hash"]["S"] == expected_hash
|
||||
|
||||
def test_write_event_persists_to_dynamodb(self):
|
||||
from moto import mock_aws
|
||||
|
||||
import boto3
|
||||
|
||||
with mock_aws():
|
||||
dyn = boto3.client("dynamodb", region_name="us-east-1")
|
||||
dyn.create_table(
|
||||
TableName="acdl-outbox",
|
||||
KeySchema=[
|
||||
{"AttributeName": "contractId", "KeyType": "HASH"},
|
||||
{"AttributeName": "eventType#eventTs", "KeyType": "RANGE"},
|
||||
],
|
||||
AttributeDefinitions=[
|
||||
{"AttributeName": "contractId", "AttributeType": "S"},
|
||||
{"AttributeName": "eventType#eventTs", "AttributeType": "S"},
|
||||
],
|
||||
BillingMode="PAY_PER_REQUEST",
|
||||
)
|
||||
|
||||
event = self._sample_event()
|
||||
write_event(event, outbox_table="acdl-outbox", region="us-east-1")
|
||||
|
||||
resp = dyn.get_item(
|
||||
TableName="acdl-outbox",
|
||||
Key={
|
||||
"contractId": {"S": "test-contract-001"},
|
||||
"eventType#eventTs": {"S": "CONFIDENCE_COMPUTED#2026-07-22T00:00:00Z"},
|
||||
},
|
||||
)
|
||||
assert "Item" in resp
|
||||
assert resp["Item"]["band"]["S"] == "pass"
|
||||
@@ -0,0 +1,67 @@
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
|
||||
|
||||
class TestPipelineIntegration:
|
||||
def test_load_ir_and_adapt_offline(self, tmp_path):
|
||||
ir = json.load(open(ROOT / "modules-ir/l1/l1-s3/spike_instance.json"))
|
||||
assert ir["stack"]["name"] == "l1-s3"
|
||||
|
||||
sys.path.insert(0, str(ROOT))
|
||||
from adapters.terraform.adapter import adapt
|
||||
out_dir = str(tmp_path / "tf")
|
||||
adapt(ir, out_dir)
|
||||
|
||||
assert os.path.isfile(os.path.join(out_dir, "main.tf"))
|
||||
assert os.path.isfile(os.path.join(out_dir, "terraform.tf"))
|
||||
assert os.path.isfile(os.path.join(out_dir, "providers.tf"))
|
||||
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert "aws_s3_bucket" in main_tf
|
||||
assert "acdl-spike-bucket" in main_tf
|
||||
|
||||
def test_confidence_signal_with_adapted_tf(self):
|
||||
sys.path.insert(0, str(ROOT))
|
||||
from acdl_platform.confidence_signal import compute
|
||||
|
||||
inputs = {
|
||||
"policy": [{"result": "pass"}],
|
||||
"validation": {"schema": True, "ir_resolved": True,
|
||||
"tf_validated": True, "tf_planned": True},
|
||||
"freshness": {"age_days": 0, "max_age_days": 7},
|
||||
"source": {"submitter": "test", "commit_sha": "test-sha"},
|
||||
"history": {"prior_rollbacks": 0, "prior_policy_fails": 0},
|
||||
"nfrs": {"conformance": None},
|
||||
}
|
||||
sig = compute("integration-test", "dev", inputs)
|
||||
assert sig.band == "pass"
|
||||
assert sig.score >= 0.50
|
||||
|
||||
def test_run_platform_check_only(self):
|
||||
result = subprocess.run(
|
||||
["bash", str(ROOT / "scripts/run_platform.sh"), "--check-only"],
|
||||
capture_output=True, text=True, cwd=str(ROOT),
|
||||
timeout=30,
|
||||
)
|
||||
assert result.returncode == 0, f"stdout: {result.stdout}\nstderr: {result.stderr}"
|
||||
assert "PLATFORM CHECK OK" in result.stdout
|
||||
|
||||
def test_run_platform_check_only_no_aws_creds(self):
|
||||
env = os.environ.copy()
|
||||
env.pop("AWS_ACCESS_KEY_ID", None)
|
||||
env.pop("AWS_SECRET_ACCESS_KEY", None)
|
||||
env.pop("AWS_DEFAULT_REGION", None)
|
||||
result = subprocess.run(
|
||||
["bash", str(ROOT / "scripts/run_platform.sh"), "--check-only"],
|
||||
capture_output=True, text=True, cwd=str(ROOT), env=env,
|
||||
timeout=30,
|
||||
)
|
||||
assert result.returncode == 0
|
||||
assert "PLATFORM CHECK OK" in result.stdout
|
||||
Reference in New Issue
Block a user