Jon Chery
1598c54a8b
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---
135 lines
4.5 KiB
Python
135 lines
4.5 KiB
Python
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" |