Jon Chery
b758a7c242
---ci--- project: acdl phase: 21 milestone: v1.6 status: execute ---/ci--- Rename the acdl_platform/ package to core/ across the directory, all imports in tests/scripts/pipelines/workflows, and doc references. The package is imported as core.confidence_signal / core.contract_resolver / core.outbox_writer. The deploy workflow's platform-repo checkout dir is renamed acdl-platform/ -> platform/ (workspace path, not the python package). Both .gitea + .github workflows stay byte-identical. Note: the original target name 'platform/' shadows Python's stdlib platform module (pytest's import uuid -> platform.system() fails when the repo root is on sys.path, which every test does). 'core/' avoids the clash while honoring the intent (drop the verbose acdl_platform). Tests: 154 pass. run_ci.sh green.
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 core.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": "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" |