diff --git a/.ciagent/REQUIREMENTS.md b/.ciagent/REQUIREMENTS.md index 5b81e78..95540eb 100644 --- a/.ciagent/REQUIREMENTS.md +++ b/.ciagent/REQUIREMENTS.md @@ -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) | \ No newline at end of file +| 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) | \ No newline at end of file diff --git a/.ciagent/ROADMAP.md b/.ciagent/ROADMAP.md index 12e6df9..e38989f 100644 --- a/.ciagent/ROADMAP.md +++ b/.ciagent/ROADMAP.md @@ -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. \ No newline at end of file + - 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. \ No newline at end of file diff --git a/.ciagent/VERIFY.md b/.ciagent/VERIFY.md index 7c044fb..d435b7c 100644 --- a/.ciagent/VERIFY.md +++ b/.ciagent/VERIFY.md @@ -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. \ No newline at end of file +**VERIFY PASS** — all four layers pass. 90 offline tests, no AWS +required for CI. \ No newline at end of file diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml new file mode 100644 index 0000000..d8fd80f --- /dev/null +++ b/.gitea/workflows/ci.yml @@ -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 \ No newline at end of file diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..d8fd80f --- /dev/null +++ b/.github/workflows/ci.yml @@ -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 \ No newline at end of file diff --git a/README.md b/README.md index d89d538..6af90e4 100644 --- a/README.md +++ b/README.md @@ -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 diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..fb9f7db --- /dev/null +++ b/pyproject.toml @@ -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" \ No newline at end of file diff --git a/requirements-test.txt b/requirements-test.txt new file mode 100644 index 0000000..769bff9 --- /dev/null +++ b/requirements-test.txt @@ -0,0 +1,6 @@ +pytest>=8.0 +pytest-cov>=4.0 +moto[dynamodb]>=5.0 +jsonschema>=4.20 +pyyaml>=6.0 +boto3>=1.34 \ No newline at end of file diff --git a/scripts/run_platform.sh b/scripts/run_platform.sh index c2b7ac3..daac43c 100755 --- a/scripts/run_platform.sh +++ b/scripts/run_platform.sh @@ -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" diff --git a/terraform/spike/terraform.tf b/terraform/spike/terraform.tf index f7e3b04..c1d14ab 100644 --- a/terraform/spike/terraform.tf +++ b/terraform/spike/terraform.tf @@ -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" } } diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..f89709d --- /dev/null +++ b/tests/conftest.py @@ -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")) \ No newline at end of file diff --git a/tests/test_adapter.py b/tests/test_adapter.py new file mode 100644 index 0000000..a4e7ffb --- /dev/null +++ b/tests/test_adapter.py @@ -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 \ No newline at end of file diff --git a/tests/test_checkov_adapter.py b/tests/test_checkov_adapter.py new file mode 100644 index 0000000..e479d1c --- /dev/null +++ b/tests/test_checkov_adapter.py @@ -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 \ No newline at end of file diff --git a/tests/test_confidence_signal.py b/tests/test_confidence_signal.py new file mode 100644 index 0000000..4003508 --- /dev/null +++ b/tests/test_confidence_signal.py @@ -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 \ No newline at end of file diff --git a/tests/test_outbox_writer.py b/tests/test_outbox_writer.py new file mode 100644 index 0000000..5ef0fc1 --- /dev/null +++ b/tests/test_outbox_writer.py @@ -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" \ No newline at end of file diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py new file mode 100644 index 0000000..82ec557 --- /dev/null +++ b/tests/test_pipeline.py @@ -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 \ No newline at end of file