Jon Chery
d5bae868a4
core/env.py dual-read helper (D-108); 21 ACDL_*→NOVA_* env vars migrated across core/scripts/adapters/tests/workflows + .env/.env.secrets (key rename, values stay). G-106 binding: run_platform.sh:288-289 + regression_verify.py:309-312 dual-read (NOVA first, ACDL fallback). G-108 binding: Gitea NOVA_* secrets created via API + workflow secrets: refs updated (deploy.yml + modules-lifecycle.yml, .gitea + .github). acdl_tagging.py→nova_tagging.py (D-109 warn mode, nova:* enforced). .acdl/→.nova/ consumer path (resolver + deploy workflow + schema + tests + docs). Test fixtures updated; pytest + run_ci.sh PASS. ---ci--- project: acdl phase: 2 milestone: v1.15 status: execute ---/ci---
507 lines
20 KiB
Python
507 lines
20 KiB
Python
"""Local emulating adapters (D-092, REQ-113).
|
|
|
|
The platform must be fully locally testable without cloud credentials.
|
|
These adapters emulate the four cloud-backed interactions the platform
|
|
uses, so the headline E2E (contract submission -> service live ->
|
|
evidence event) runs end-to-end against the local tier with no AWS:
|
|
|
|
1. FlatFileOutbox - emulates the DynamoDB outbox (core/outbox_writer.py)
|
|
2. LocalEcsEmulator - emulates an ECS Fargate service returning HTTP 200
|
|
3. LocalS3StateBackend - rewrites the terraform S3 backend to a local backend
|
|
4. LocalLambdaStub - invokes the contract_ingestor handler in-process
|
|
|
|
Each adapter exposes the same interface as the live counterpart so the
|
|
caller code path is unchanged; only the I/O target swaps. Selection is
|
|
gated on the NOVA_LOCAL_TIER env var (set by run_platform.sh --local).
|
|
Dual-read via core/env.py: NOVA_* preferred, ACDL_* fallback until P5.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime
|
|
import hashlib
|
|
import http.server
|
|
import json
|
|
import os
|
|
import socket
|
|
import socketserver
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
# Repo root on sys.path so `from core import env` resolves to THIS package
|
|
# when run as a script (avoids editable-installed third-party `core` shadow).
|
|
_REPO_ROOT = str(Path(__file__).resolve().parent.parent)
|
|
if _REPO_ROOT not in sys.path:
|
|
sys.path.insert(0, _REPO_ROOT)
|
|
|
|
from core import env
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
def is_local_tier() -> bool:
|
|
"""True when the local emulating tier is active."""
|
|
return env.get_env("LOCAL_TIER", "") == "1"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 1. Flat-file DynamoDB outbox emulator
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@dataclass
|
|
class FlatFileOutbox:
|
|
"""Emulates the DynamoDB outbox with flat files in a temp folder.
|
|
|
|
Same write/read interface contract as core.outbox_writer.write_event:
|
|
accepts an event dict, returns the item dict (with a hash-chained
|
|
`hash` field). The item is appended to a JSONL file
|
|
`<dir>/outbox.jsonl` so the chain is reconstructable.
|
|
"""
|
|
|
|
dir: Path
|
|
_chain_tail_hash: str = "GENESIS"
|
|
|
|
@classmethod
|
|
def create(cls, dir: Optional[Path] = None) -> "FlatFileOutbox":
|
|
d = Path(dir) if dir else Path(tempfile.mkdtemp(prefix="acdl_outbox_"))
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
out = cls(dir=d)
|
|
# Re-read the chain tail if the file already exists.
|
|
jl = d / "outbox.jsonl"
|
|
if jl.exists():
|
|
tail = None
|
|
for line in jl.read_text().splitlines():
|
|
if line.strip():
|
|
tail = json.loads(line)
|
|
if tail:
|
|
out._chain_tail_hash = tail["hash"]
|
|
return out
|
|
|
|
def _canonical_hash(self, event: Dict) -> str:
|
|
canonical = json.dumps(event, sort_keys=True, separators=(",", ":"))
|
|
return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
|
|
|
|
def write_event(self, event: Dict[str, Any],
|
|
outbox_table: str = "acdl-outbox-local",
|
|
region: str = "local") -> Dict[str, Any]:
|
|
"""Write an evidence event to the flat-file outbox.
|
|
|
|
Mirrors core.outbox_writer.write_event signature. Returns the
|
|
item dict (single-valued, not DynamoDB-typed) so the caller can
|
|
inspect it without unwrapping."""
|
|
contract_id = event["contractId"]
|
|
event_type = event.get("eventType", "CONFIDENCE_COMPUTED")
|
|
event_ts = event.get("ts") or datetime.datetime.now(
|
|
datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
sk = f"{event_type}#{event_ts}"
|
|
prev_hash = event.get("prev_event_hash", self._chain_tail_hash)
|
|
event_hash = self._canonical_hash(event)
|
|
item = {
|
|
"contractId": contract_id,
|
|
"eventType#eventTs": sk,
|
|
"payload": event,
|
|
"prev_event_hash": prev_hash,
|
|
"hash": event_hash,
|
|
"environment": str(event.get("environment", "")),
|
|
"stack": str(event.get("stack", "")),
|
|
"score": event.get("score", 0),
|
|
"band": str(event.get("band", "")),
|
|
"expire_at": int((datetime.datetime.now(datetime.timezone.utc)
|
|
+ datetime.timedelta(days=365)).timestamp()),
|
|
}
|
|
jl = self.dir / "outbox.jsonl"
|
|
with jl.open("a") as f:
|
|
f.write(json.dumps(item, sort_keys=True) + "\n")
|
|
self._chain_tail_hash = event_hash
|
|
return item
|
|
|
|
def read_all(self) -> List[Dict[str, Any]]:
|
|
"""Read every event in the flat-file outbox (for verification)."""
|
|
jl = self.dir / "outbox.jsonl"
|
|
if not jl.exists():
|
|
return []
|
|
return [json.loads(line) for line in jl.read_text().splitlines()
|
|
if line.strip()]
|
|
|
|
def verify_chain(self) -> bool:
|
|
"""Verify the hash chain is intact (each prev_event_hash matches
|
|
the prior event's hash; the first event's prev is GENESIS)."""
|
|
events = self.read_all()
|
|
prev = "GENESIS"
|
|
for ev in events:
|
|
if ev["prev_event_hash"] != prev:
|
|
return False
|
|
# Recompute the hash and confirm it matches.
|
|
recomputed = self._canonical_hash(ev["payload"])
|
|
if recomputed != ev["hash"]:
|
|
return False
|
|
prev = ev["hash"]
|
|
return True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 2. Local ECS Fargate emulator
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@dataclass
|
|
class LocalEcsEmulator:
|
|
"""Emulates an ECS Fargate service by serving HTTP 200 from a local
|
|
shell process.
|
|
|
|
Records the service definition (so the caller can inspect what would
|
|
have been deployed) and starts a tiny HTTP server on a free port that
|
|
returns 200 OK for any path. The caller can then curl the endpoint to
|
|
confirm the service is "live" in the local tier.
|
|
"""
|
|
|
|
service_name: str
|
|
service_definition: Dict[str, Any]
|
|
_server: Optional[socketserver.TCPServer] = None
|
|
_thread: Optional[threading.Thread] = None
|
|
_port: int = 0
|
|
|
|
def deploy(self) -> Dict[str, Any]:
|
|
"""Start the local HTTP server; return the endpoint metadata."""
|
|
service_name = self.service_name # capture for the handler closure
|
|
|
|
class Handler(http.server.BaseHTTPRequestHandler):
|
|
def do_GET(self, *a, **k):
|
|
body = json.dumps({
|
|
"service": service_name,
|
|
"status": "RUNNING",
|
|
"tier": "local-emulator",
|
|
"path": self.path,
|
|
}).encode()
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def log_message(self, *a, **k):
|
|
pass # silence
|
|
|
|
# Bind directly to port 0 (the OS assigns a free port atomically).
|
|
# The prior approach (open a socket, read the port, close, then
|
|
# bind TCPServer) was a TOCTOU race: another process could grab
|
|
# the port between close and bind. Binding to port 0 avoids the
|
|
# race entirely.
|
|
self._server = socketserver.TCPServer(
|
|
("127.0.0.1", 0), Handler)
|
|
self._server.allow_reuse_address = True
|
|
self._port = self._server.server_address[1]
|
|
self._thread = threading.Thread(
|
|
target=self._server.serve_forever, daemon=True)
|
|
self._thread.start()
|
|
return {
|
|
"service_arn": f"arn:local:ecs:us-east-1:000000000000:service/{self.service_name}",
|
|
"endpoint": f"http://127.0.0.1:{self._port}",
|
|
"status": "RUNNING",
|
|
"tier": "local-emulator",
|
|
"desired_count": self.service_definition.get("desired_count", 1),
|
|
"running_count": self.service_definition.get("desired_count", 1),
|
|
}
|
|
|
|
def health_check(self, endpoint: str, timeout_s: float = 5.0) -> Tuple[bool, int]:
|
|
"""curl the endpoint; return (ok, status_code)."""
|
|
import urllib.request
|
|
url = endpoint if endpoint.startswith("http") else f"http://{endpoint}"
|
|
t0 = time.monotonic()
|
|
while time.monotonic() - t0 < timeout_s:
|
|
try:
|
|
with urllib.request.urlopen(url, timeout=1.0) as r:
|
|
return (r.status == 200, r.status)
|
|
except Exception:
|
|
time.sleep(0.1)
|
|
return (False, 0)
|
|
|
|
def destroy(self):
|
|
"""Stop the local HTTP server."""
|
|
if self._server is not None:
|
|
self._server.shutdown()
|
|
self._server.server_close()
|
|
self._server = None
|
|
if self._thread is not None:
|
|
self._thread.join(timeout=2.0)
|
|
self._thread = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 3. Local S3 state backend (terraform backend rewrite)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@dataclass
|
|
class LocalS3StateBackend:
|
|
"""Replaces the terraform S3 backend with a local backend.
|
|
|
|
The adapter emits a `backend "s3" { ... }` block. In the local tier
|
|
we rewrite it to `backend "local" { path = "<temp>/terraform.tfstate" }`
|
|
so `terraform init/plan` runs without S3. The rewrite is applied to
|
|
the emitted terraform.tf file before terraform is invoked.
|
|
"""
|
|
|
|
state_dir: Path
|
|
|
|
@classmethod
|
|
def create(cls, dir: Optional[Path] = None) -> "LocalS3StateBackend":
|
|
d = Path(dir) if dir else Path(tempfile.mkdtemp(prefix="acdl_tfstate_"))
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
return cls(state_dir=d)
|
|
|
|
def state_path(self, stack_name: str) -> Path:
|
|
return self.state_dir / f"{stack_name}.tfstate"
|
|
|
|
def rewrite_terraform_tf(self, tf_path: Path, stack_name: str) -> str:
|
|
"""Rewrite the backend block in a terraform.tf file to local.
|
|
|
|
Returns the new content (also written to disk)."""
|
|
import re
|
|
content = Path(tf_path).read_text()
|
|
# Replace the `backend "s3" { ... }` block with a local backend.
|
|
new_content = re.sub(
|
|
r'backend "s3" \{[^}]*\}',
|
|
f'backend "local" {{\n path = "{self.state_path(stack_name)}"\n }}',
|
|
content,
|
|
count=1,
|
|
flags=re.DOTALL,
|
|
)
|
|
Path(tf_path).write_text(new_content)
|
|
return new_content
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 4. Local Lambda stub (in-process handler invocation)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@dataclass
|
|
class LocalLambdaStub:
|
|
"""Invokes the contract_ingestor handler in-process.
|
|
|
|
Instead of calling AWS Lambda via boto3, this stub imports
|
|
core.lambda.contract_ingestor.lambda_handler and invokes it with a
|
|
synthesized Function-URL-style event. The DynamoDB write inside the
|
|
handler is redirected to a FlatFileOutbox so no AWS is required.
|
|
"""
|
|
|
|
outbox: FlatFileOutbox
|
|
|
|
def invoke(self, payload: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Invoke the contract_ingestor handler in-process.
|
|
|
|
Returns the handler's response dict
|
|
({statusCode, body}). The handler's DynamoDB calls are
|
|
intercepted via the NOVA_LOCAL_TIER env var (the handler checks
|
|
_get_dynamodb(); under local tier it would need patching - we
|
|
patch the module's _get_dynamodb to return a local stub)."""
|
|
# Import the handler module (the dir is named `lambda`, a Python
|
|
# keyword, so use importlib instead of a dotted import).
|
|
import importlib
|
|
ci = importlib.import_module("core.lambda.contract_ingestor")
|
|
|
|
# Patch the handler's DynamoDB resource with a local stub that
|
|
# writes to the flat-file outbox. The handler uses _get_dynamodb()
|
|
# which returns a boto3 resource; we replace it with a minimal
|
|
# object exposing .Table(name) with .put_item(Item=...).
|
|
original_get = ci._get_dynamodb
|
|
|
|
class _LocalTable:
|
|
def __init__(self, name, outbox):
|
|
self.name = name
|
|
self.outbox = outbox
|
|
|
|
def put_item(self, *, TableName=None, Item=None, **kwargs):
|
|
# The handler calls put_item(TableName=..., Item=...).
|
|
# DynamoDB-typed items ({'S': ...}, {'N': ...}) are
|
|
# flattened for the flat-file outbox.
|
|
Item = Item or {}
|
|
flat = {}
|
|
for k, v in Item.items():
|
|
if isinstance(v, dict):
|
|
if "S" in v:
|
|
flat[k] = v["S"]
|
|
elif "N" in v:
|
|
flat[k] = v["N"]
|
|
else:
|
|
flat[k] = v
|
|
else:
|
|
flat[k] = v
|
|
self.outbox.write_event({
|
|
"contractId": flat.get("contractId", "local"),
|
|
"eventType": f"LAMBDA_{self.name}",
|
|
"ts": datetime.datetime.now(datetime.timezone.utc)
|
|
.strftime("%Y-%m-%dT%H:%M:%SZ"),
|
|
"environment": flat.get("environment", "local"),
|
|
"stack": self.name,
|
|
"score": 0,
|
|
"band": "local",
|
|
"prev_event_hash": "GENESIS",
|
|
})
|
|
return {}
|
|
|
|
class _LocalDynamoResource:
|
|
def __init__(self, outbox):
|
|
self.outbox = outbox
|
|
|
|
def Table(self, name):
|
|
return _LocalTable(name, self.outbox)
|
|
|
|
class _LocalSecretsClient:
|
|
def get_secret_value(self, SecretId):
|
|
return {"SecretString": json.dumps({"token": "local-stub"})}
|
|
|
|
ci._get_dynamodb = lambda: _LocalDynamoResource(self.outbox)
|
|
ci._get_secrets_client = lambda: _LocalSecretsClient()
|
|
# Stub the urllib GitHub API call so report_error doesn't hit the network.
|
|
original_urlopen = None
|
|
try:
|
|
import urllib.request
|
|
original_urlopen = urllib.request.urlopen
|
|
|
|
class _FakeResponse:
|
|
def __init__(self, body=b"{}", status=200):
|
|
self._body = body
|
|
self.status = status
|
|
|
|
def read(self):
|
|
return self._body
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *a):
|
|
return False
|
|
|
|
def _fake_urlopen(url, *a, **k):
|
|
return _FakeResponse(
|
|
json.dumps([{"number": 1, "title": "stub"}]).encode())
|
|
urllib.request.urlopen = _fake_urlopen
|
|
except (AttributeError, TypeError) as e:
|
|
import sys
|
|
print(f"WARNING: could not patch urlopen for local Lambda stub: {e}", file=sys.stderr)
|
|
|
|
try:
|
|
event = {
|
|
"body": json.dumps(payload),
|
|
"requestContext": {
|
|
"httpContext": {"authorizer": {"iam": {"userId": "local-stub"}}}
|
|
},
|
|
}
|
|
result = ci.lambda_handler(event, None)
|
|
finally:
|
|
ci._get_dynamodb = original_get
|
|
if original_urlopen is not None:
|
|
import urllib.request
|
|
urllib.request.urlopen = original_urlopen
|
|
return result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Convenience: run the headline E2E against the local tier
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def run_local_e2e(contract_path: str, repo_root: Optional[Path] = None) -> Dict[str, Any]:
|
|
"""Run the headline E2E against the local emulating tier.
|
|
|
|
Steps:
|
|
1. Resolve the contract -> Target Stack.
|
|
2. Adapter compiles the stack -> terraform files (structure validated).
|
|
3. LocalS3StateBackend rewrites the backend to local.
|
|
4. LocalEcsEmulator deploys a synthetic HTTP 200 service (if the
|
|
stack has an ECS service) and confirms health.
|
|
5. FlatFileOutbox writes a CONFIDENCE_COMPUTED event; chain verified.
|
|
6. LocalLambdaStub invokes the contract_ingestor handler in-process.
|
|
|
|
Returns a dict of results. Raises AssertionError on any failure.
|
|
"""
|
|
root = Path(repo_root) if repo_root else ROOT
|
|
prior_cwd = os.getcwd()
|
|
os.chdir(str(root))
|
|
try:
|
|
sys.path.insert(0, str(root))
|
|
from core.contract_resolver import resolve
|
|
import adapters.terraform.adapter as adapter
|
|
|
|
stack = resolve(contract_path, str(root))
|
|
stack_name = stack["stack"]["name"]
|
|
work = Path(tempfile.mkdtemp(prefix="acdl_local_e2e_"))
|
|
tf_dir = work / "tf"
|
|
tf_dir.mkdir(exist_ok=True)
|
|
adapter.adapt(stack, str(tf_dir))
|
|
|
|
# 3. Local S3 state backend rewrite.
|
|
backend = LocalS3StateBackend.create(dir=work / "tfstate")
|
|
tf_tf = tf_dir / "terraform.tf"
|
|
backend.rewrite_terraform_tf(tf_tf, stack_name)
|
|
assert "backend \"local\"" in tf_tf.read_text(), "backend not rewritten"
|
|
|
|
# 4. Local ECS emulator (only if the stack has an ECS service).
|
|
ecs_result = None
|
|
has_ecs = any(r["type"] == "aws:ecs:service" for r in stack["resources"])
|
|
if has_ecs:
|
|
ecs = LocalEcsEmulator(
|
|
service_name=stack_name,
|
|
service_definition={"desired_count": 1},
|
|
)
|
|
deploy_meta = ecs.deploy()
|
|
ok, status = ecs.health_check(deploy_meta["endpoint"])
|
|
assert ok, f"ECS emulator health check failed: status={status}"
|
|
ecs_result = deploy_meta
|
|
ecs.destroy()
|
|
|
|
# 5. Flat-file outbox: write a CONFIDENCE_COMPUTED event + verify chain.
|
|
outbox = FlatFileOutbox.create(dir=work / "outbox")
|
|
event = {
|
|
"contractId": "local-e2e-test",
|
|
"eventType": "CONFIDENCE_COMPUTED",
|
|
"ts": datetime.datetime.now(datetime.timezone.utc)
|
|
.strftime("%Y-%m-%dT%H:%M:%SZ"),
|
|
"environment": "dev",
|
|
"stack": stack_name,
|
|
"score": 0.9,
|
|
"band": "pass",
|
|
"prev_event_hash": "GENESIS",
|
|
}
|
|
item = outbox.write_event(event)
|
|
assert item["hash"], "outbox item missing hash"
|
|
assert outbox.verify_chain(), "outbox hash chain broken"
|
|
|
|
# 6. Local Lambda stub: invoke the contract_ingestor handler.
|
|
lambda_stub = LocalLambdaStub(outbox=outbox)
|
|
lambda_result = lambda_stub.invoke({
|
|
"action": "submit_contract",
|
|
"consumerRepo": "local-test/consumer",
|
|
"contractId": "local-e2e-test",
|
|
"contract": {"module": stack_name, "environment": "dev"},
|
|
"environment": "dev",
|
|
})
|
|
assert lambda_result["statusCode"] == 200, (
|
|
f"lambda stub returned {lambda_result['statusCode']}: {lambda_result.get('body')}")
|
|
|
|
return {
|
|
"stack_name": stack_name,
|
|
"tier": "local-emulator",
|
|
"tf_dir": str(tf_dir),
|
|
"backend": "local",
|
|
"ecs": ecs_result,
|
|
"outbox_dir": str(outbox.dir),
|
|
"outbox_events": len(outbox.read_all()),
|
|
"outbox_chain_verified": True,
|
|
"lambda_status": lambda_result["statusCode"],
|
|
}
|
|
finally:
|
|
os.chdir(prior_cwd)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
contract = sys.argv[1] if len(sys.argv) > 1 else "contracts/microservice.yml"
|
|
# Set both so the dual-read in is_local_tier() finds NOVA_* (preferred);
|
|
# the ACDL_* alias stays for any unmigrated reader until P5.
|
|
os.environ["NOVA_LOCAL_TIER"] = "1"
|
|
os.environ["ACDL_LOCAL_TIER"] = "1" # legacy alias (dual-read fallback), removed in P5
|
|
result = run_local_e2e(contract)
|
|
print(json.dumps(result, indent=2)) |