Jon Chery
1e4133e11a
---ci---
project: acdl
phase: 30
milestone: v1.8
status: execute
---/ci---
P1-8: run_platform.sh now emits adapter output to $WORK/tf (per-run temp
dir), not the committed terraform/spike/ directory. The committed
terraform/spike/*.tf files are removed — they were scratch artifacts.
Deploy workflow artifact upload path updated to /tmp/acdl_platform_run_v18/tf/.
P1-9: contract_ingestor.py now reads GITHUB_API_BASE env for forge-agnostic
API URLs. _forge_type() detects GitHub vs Gitea. Search URL is branched
(GitHub uses /search/issues, Gitea uses /repos/{owner}/{repo}/issues).
S1: Deploy workflow configure-aws-credentials step restructured as a single
conditional step. OIDC when no static key (role-to-assume), static-key
when ACDL_AWS_ACCESS_KEY_ID present (access-key-id/secret-access-key inputs).
Both deploy workflows remain byte-identical.
Tests: +8 (292 -> 300). All pass. run_platform.sh --check-only green.
420 lines
18 KiB
Python
420 lines
18 KiB
Python
"""Unit tests for core/lambda/contract_ingestor.py.
|
|
|
|
The source file lives at ``core/lambda/contract_ingestor.py`` for repo
|
|
organization, but ``lambda`` is a Python reserved word — so the package
|
|
path ``core.lambda`` cannot be imported with normal ``import`` syntax.
|
|
The Lambda runtime packages the handler as a top-level
|
|
``contract_ingestor.py`` (handler ``contract_ingestor.lambda_handler``),
|
|
which is the name the Terraform ``handler`` attribute uses. The tests
|
|
mirror that by loading the module from its file path under the name
|
|
``contract_ingestor``.
|
|
|
|
Uses moto (already a test dependency — see requirements-test.txt) to mock
|
|
DynamoDB, mirroring the pattern in tests/test_outbox_writer.py.
|
|
"""
|
|
|
|
import datetime
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
# Load core/lambda/contract_ingestor.py as a top-level module named
|
|
# `contract_ingestor` (the name the Lambda runtime uses).
|
|
_SOURCE_PATH = Path(__file__).resolve().parent.parent / "core" / "lambda" / "contract_ingestor.py"
|
|
_spec = importlib.util.spec_from_file_location("contract_ingestor", _SOURCE_PATH)
|
|
ingestor = importlib.util.module_from_spec(_spec)
|
|
_spec.loader.exec_module(ingestor)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.fixture
|
|
def sample_payload():
|
|
return {
|
|
"consumerRepo": "acdl/consumer-a",
|
|
"contractId": "contract-001",
|
|
"contract": {"stack": "s3", "environment": "dev"},
|
|
"environment": "dev",
|
|
"action": "submit_contract",
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def function_url_event(sample_payload):
|
|
return {"body": json.dumps(sample_payload)}
|
|
|
|
|
|
@pytest.fixture
|
|
def moto_contracts_table(monkeypatch):
|
|
"""Spin up a moto-backed DynamoDB and point the ingestor at it."""
|
|
from moto import mock_aws
|
|
import boto3
|
|
|
|
monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-1")
|
|
monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing")
|
|
monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing")
|
|
|
|
with mock_aws():
|
|
dyn = boto3.client("dynamodb", region_name="us-east-1")
|
|
dyn.create_table(
|
|
TableName="acdl-contracts",
|
|
KeySchema=[
|
|
{"AttributeName": "consumerRepo", "KeyType": "HASH"},
|
|
{"AttributeName": "contractId#submittedAt", "KeyType": "RANGE"},
|
|
],
|
|
AttributeDefinitions=[
|
|
{"AttributeName": "consumerRepo", "AttributeType": "S"},
|
|
{"AttributeName": "contractId#submittedAt", "AttributeType": "S"},
|
|
],
|
|
BillingMode="PAY_PER_REQUEST",
|
|
)
|
|
|
|
# Reset the cached boto3 clients so the ingestor picks up the moto
|
|
# session, then yield with moto active.
|
|
saved_dynamodb = ingestor._dynamodb
|
|
saved_secrets = ingestor._secrets_client
|
|
ingestor._dynamodb = None
|
|
ingestor._secrets_client = None
|
|
monkeypatch.setattr(ingestor, "TABLE_NAME", "acdl-contracts")
|
|
|
|
yield dyn
|
|
|
|
ingestor._dynamodb = saved_dynamodb
|
|
ingestor._secrets_client = saved_secrets
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# submit_contract
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestSubmitContract:
|
|
def test_submit_contract_writes_correct_pk_sk_attributes(self, moto_contracts_table, sample_payload):
|
|
result = ingestor._submit_contract(sample_payload)
|
|
|
|
assert result["status"] == "ok"
|
|
assert result["contractId"] == "contract-001"
|
|
assert result["action"] == "submit_contract"
|
|
assert "submittedAt" in result
|
|
|
|
# Verify what landed in DynamoDB.
|
|
sk = f"contract-001#{result['submittedAt']}"
|
|
resp = moto_contracts_table.get_item(
|
|
TableName="acdl-contracts",
|
|
Key={
|
|
"consumerRepo": {"S": "acdl/consumer-a"},
|
|
"contractId#submittedAt": {"S": sk},
|
|
},
|
|
)
|
|
assert "Item" in resp
|
|
item = resp["Item"]
|
|
assert item["consumerRepo"]["S"] == "acdl/consumer-a"
|
|
assert item["contractId"]["S"] == "contract-001"
|
|
assert item["status"]["S"] == "submitted"
|
|
assert item["environment"]["S"] == "dev"
|
|
assert item["submittedAt"]["S"] == result["submittedAt"]
|
|
# The contract attribute holds the full contract object. boto3's
|
|
# resource API serializes a dict as a DynamoDB Map (type "M"); each
|
|
# leaf scalar is wrapped in its own type tag.
|
|
expected_contract = sample_payload["contract"]
|
|
actual_contract = item["contract"]
|
|
# The resource API stores scalars inside the map with their own type
|
|
# tags (e.g. {"S": ...}); unwrap one level for the two known leaves.
|
|
unwrapped = {
|
|
k: list(v.values())[0] if isinstance(v, dict) and len(v) == 1 else v
|
|
for k, v in actual_contract["M"].items()
|
|
}
|
|
assert unwrapped == expected_contract
|
|
|
|
def test_submit_contract_sk_contains_contract_id_and_timestamp(self, moto_contracts_table, sample_payload):
|
|
result = ingestor._submit_contract(sample_payload)
|
|
sk = f"contract-001#{result['submittedAt']}"
|
|
# SK format is contractId#ISO8601
|
|
assert sk.split("#")[0] == "contract-001"
|
|
# timestamp parses as ISO 8601 with a Z suffix.
|
|
ts = sk.split("#", 1)[1]
|
|
datetime.datetime.strptime(ts, "%Y-%m-%dT%H:%M:%SZ")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# report_error (D-055) — GitHub issue creation via the GitHub API
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestReportError:
|
|
"""The report_error action creates a GitHub issue on the platform repo.
|
|
|
|
These tests mock the GitHub API (urllib.request.urlopen) and Secrets
|
|
Manager (get_secret_value) so they run fully offline.
|
|
"""
|
|
|
|
@pytest.fixture
|
|
def error_payload(self):
|
|
return {
|
|
"consumerRepo": "acdl/consumer-a",
|
|
"contractId": "contract-001",
|
|
"error": "deploy failed",
|
|
"runUrl": "https://github.com/acdl/consumer-a/actions/runs/1",
|
|
"environment": "dev",
|
|
"action": "report_error",
|
|
}
|
|
|
|
@pytest.fixture
|
|
def patched_secrets(self, monkeypatch):
|
|
"""Patch the Secrets Manager client to return a fake token."""
|
|
def fake_get_secret_value(SecretId):
|
|
return {"SecretString": "fake-github-token-1234"}
|
|
monkeypatch.setattr(
|
|
ingestor, "_get_secrets_client",
|
|
lambda: type("FakeSecrets", (), {"get_secret_value": staticmethod(fake_get_secret_value)})()
|
|
)
|
|
|
|
def _mock_urlopen(self, monkeypatch, responses):
|
|
"""Patch urllib.request.urlopen to return queued responses.
|
|
|
|
``responses`` is a list of (status_code, json_body) tuples. Each call
|
|
to urlopen pops the next response. The returned mock object supports
|
|
context-manager use (``with urlopen(...) as resp:``) and direct call.
|
|
"""
|
|
import io
|
|
call_log = []
|
|
|
|
class FakeResp:
|
|
def __init__(self, body):
|
|
self._buf = io.BytesIO(body.encode() if isinstance(body, str) else body)
|
|
def read(self):
|
|
return self._buf.read()
|
|
def __enter__(self):
|
|
return self
|
|
def __exit__(self, *a):
|
|
return False
|
|
|
|
queue = list(responses)
|
|
|
|
def fake_urlopen(req, timeout=None):
|
|
call_log.append(req)
|
|
if queue:
|
|
status, body = queue.pop(0)
|
|
return FakeResp(body)
|
|
# Default: empty 200
|
|
return FakeResp("{}")
|
|
|
|
monkeypatch.setattr("urllib.request.urlopen", fake_urlopen)
|
|
return call_log
|
|
|
|
def test_report_error_creates_new_issue(self, monkeypatch, error_payload, patched_secrets):
|
|
# Search returns no items → create a new issue.
|
|
calls = self._mock_urlopen(monkeypatch, [
|
|
(200, json.dumps({"items": []})), # search
|
|
(201, json.dumps({"number": 42, "html_url": "https://github.com/acdl/acdl/issues/42"})), # create
|
|
])
|
|
result = ingestor._report_error(error_payload)
|
|
assert result["status"] == "issue_created"
|
|
assert result["issueNumber"] == 42
|
|
assert result["issueUrl"] == "https://github.com/acdl/acdl/issues/42"
|
|
assert result["contractId"] == "contract-001"
|
|
assert result["action"] == "report_error"
|
|
# Two API calls: search + create
|
|
assert len(calls) == 2
|
|
# The create call must be a POST to the issues endpoint
|
|
create_req = calls[1]
|
|
assert create_req.method == "POST"
|
|
assert "/issues" in create_req.full_url
|
|
|
|
def test_report_error_comments_on_existing_issue(self, monkeypatch, error_payload, patched_secrets):
|
|
# Search returns an existing open issue → comment on it (idempotency).
|
|
calls = self._mock_urlopen(monkeypatch, [
|
|
(200, json.dumps({"items": [{"number": 99}]})), # search (found)
|
|
(201, json.dumps({"id": 123, "issue_url": "https://github.com/acdl/acdl/issues/99"})), # comment
|
|
])
|
|
result = ingestor._report_error(error_payload)
|
|
assert result["status"] == "commented_on_existing"
|
|
assert result["issueNumber"] == 99
|
|
assert result["contractId"] == "contract-001"
|
|
assert result["action"] == "report_error"
|
|
# Two API calls: search + comment (no create)
|
|
assert len(calls) == 2
|
|
# The comment call is a POST to the comments endpoint
|
|
comment_req = calls[1]
|
|
assert comment_req.method == "POST"
|
|
assert "/comments" in comment_req.full_url
|
|
|
|
def test_report_error_missing_field_raises(self):
|
|
payload = {"consumerRepo": "acdl/consumer-a"} # missing contractId, error
|
|
with pytest.raises(ValueError):
|
|
ingestor._report_error(payload)
|
|
|
|
def test_report_error_secrets_manager_failure_raises(self, monkeypatch, error_payload):
|
|
# If Secrets Manager fails to return a token, the action should raise
|
|
# a RuntimeError (caught by the top-level lambda_handler → 500).
|
|
def failing_secrets():
|
|
class FailingClient:
|
|
def get_secret_value(self, SecretId):
|
|
raise Exception("secret not found")
|
|
return FailingClient()
|
|
monkeypatch.setattr(ingestor, "_get_secrets_client", failing_secrets)
|
|
with pytest.raises(RuntimeError, match="failed to read GitHub token"):
|
|
ingestor._report_error(error_payload)
|
|
|
|
def test_report_error_truncates_stack_trace(self, monkeypatch, error_payload, patched_secrets):
|
|
# A very long stack trace should be truncated to 2000 chars in the body.
|
|
error_payload["stackTrace"] = "x" * 5000
|
|
calls = self._mock_urlopen(monkeypatch, [
|
|
(200, json.dumps({"items": []})),
|
|
(201, json.dumps({"number": 1, "html_url": "u"})),
|
|
])
|
|
result = ingestor._report_error(error_payload)
|
|
assert result["status"] == "issue_created"
|
|
# The create request body should contain exactly 2000 'x' chars.
|
|
create_req = calls[1]
|
|
body = json.loads(create_req.data.decode())
|
|
# The body markdown contains the (truncated) stack trace.
|
|
assert "x" * 2000 in body["body"]
|
|
assert "x" * 2001 not in body["body"]
|
|
|
|
def test_lambda_handler_routes_report_error(self, monkeypatch, error_payload, patched_secrets):
|
|
# End-to-end via lambda_handler: action=report_error → 200.
|
|
self._mock_urlopen(monkeypatch, [
|
|
(200, json.dumps({"items": []})),
|
|
(201, json.dumps({"number": 7, "html_url": "https://github.com/acdl/acdl/issues/7"})),
|
|
])
|
|
event = {"body": json.dumps(error_payload)}
|
|
resp = ingestor.lambda_handler(event, None)
|
|
assert resp["statusCode"] == 200
|
|
body = json.loads(resp["body"])
|
|
assert body["status"] == "issue_created"
|
|
assert body["action"] == "report_error"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# lambda_handler wrapper (Function URL event)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestLambdaHandler:
|
|
def test_submit_contract_event_returns_200(self, moto_contracts_table, function_url_event):
|
|
resp = ingestor.lambda_handler(function_url_event, None)
|
|
assert resp["statusCode"] == 200
|
|
body = json.loads(resp["body"])
|
|
assert body["status"] == "ok"
|
|
assert body["contractId"] == "contract-001"
|
|
assert body["action"] == "submit_contract"
|
|
|
|
def test_body_can_be_dict_not_string(self, moto_contracts_table, sample_payload):
|
|
# Some test harnesses pass body as a dict already.
|
|
resp = ingestor.lambda_handler({"body": sample_payload}, None)
|
|
assert resp["statusCode"] == 200
|
|
|
|
def test_missing_field_returns_400(self, moto_contracts_table):
|
|
payload = {
|
|
"consumerRepo": "acdl/consumer-a",
|
|
# missing contractId, contract, environment
|
|
}
|
|
resp = ingestor.lambda_handler({"body": json.dumps(payload)}, None)
|
|
assert resp["statusCode"] == 400
|
|
body = json.loads(resp["body"])
|
|
assert "missing field" in body["error"]
|
|
|
|
def test_missing_required_field_contract(self, moto_contracts_table, sample_payload):
|
|
del sample_payload["contract"]
|
|
resp = ingestor.lambda_handler({"body": json.dumps(sample_payload)}, None)
|
|
assert resp["statusCode"] == 400
|
|
assert "contract" in json.loads(resp["body"])["error"]
|
|
|
|
def test_unknown_action_returns_400(self, moto_contracts_table):
|
|
payload = {
|
|
"consumerRepo": "acdl/consumer-a",
|
|
"contractId": "contract-001",
|
|
"contract": {},
|
|
"environment": "dev",
|
|
"action": "do_something_else",
|
|
}
|
|
resp = ingestor.lambda_handler({"body": json.dumps(payload)}, None)
|
|
assert resp["statusCode"] == 400
|
|
body = json.loads(resp["body"])
|
|
assert "unknown action" in body["error"]
|
|
|
|
def test_default_action_is_submit_contract(self, moto_contracts_table, sample_payload):
|
|
del sample_payload["action"]
|
|
resp = ingestor.lambda_handler({"body": json.dumps(sample_payload)}, None)
|
|
assert resp["statusCode"] == 200
|
|
body = json.loads(resp["body"])
|
|
assert body["action"] == "submit_contract"
|
|
|
|
def test_internal_error_returns_500(self, moto_contracts_table, sample_payload):
|
|
# Force _submit_contract to blow up after passing validation.
|
|
with mock.patch.object(ingestor, "_submit_contract", side_effect=RuntimeError("boom")):
|
|
resp = ingestor.lambda_handler(
|
|
{"body": json.dumps(sample_payload)}, None
|
|
)
|
|
assert resp["statusCode"] == 500
|
|
body = json.loads(resp["body"])
|
|
assert body["error"] == "boom"
|
|
|
|
|
|
class TestCallerIdentityValidation:
|
|
"""P1-2: the Lambda validates consumerRepo against the invoking principal."""
|
|
|
|
def test_no_identity_skips_check(self, moto_contracts_table, function_url_event):
|
|
# No requestContext.identity in the event — check is skipped (relies on IAM ABAC).
|
|
resp = ingestor.lambda_handler(function_url_event, None)
|
|
assert resp["statusCode"] == 200
|
|
|
|
def test_invalid_consumer_repo_format_rejected(self, moto_contracts_table, sample_payload):
|
|
# A consumerRepo without "/" is invalid (not org/repo format).
|
|
sample_payload["consumerRepo"] = "not-a-repo-format"
|
|
event = {"body": json.dumps(sample_payload), "requestContext": {"identity": {"userArn": "arn:aws:sts::000:assumed-role/acdl-deploy/session"}}}
|
|
resp = ingestor.lambda_handler(event, None)
|
|
assert resp["statusCode"] == 400
|
|
assert "invalid consumerRepo" in json.loads(resp["body"])["error"]
|
|
|
|
def test_valid_consumer_repo_with_identity_passes(self, moto_contracts_table, sample_payload):
|
|
# A valid org/repo consumerRepo with an identity present — passes.
|
|
event = {
|
|
"body": json.dumps(sample_payload),
|
|
"requestContext": {"identity": {"userArn": "arn:aws:sts::000:assumed-role/acdl-deploy/acdl-consumer-a"}},
|
|
}
|
|
resp = ingestor.lambda_handler(event, None)
|
|
assert resp["statusCode"] == 200
|
|
|
|
|
|
class TestForgeAgnosticApiUrls:
|
|
"""P1-9: contract_ingestor uses GITHUB_API_BASE for forge-agnostic URLs."""
|
|
|
|
def test_default_api_base_is_github(self):
|
|
assert ingestor.GITHUB_API_BASE == "https://api.github.com"
|
|
|
|
def test_forge_type_detects_gitea(self, monkeypatch):
|
|
monkeypatch.setattr(ingestor, "GITHUB_API_BASE", "https://git.cloudinit.dev/api/v1")
|
|
assert ingestor._forge_type() == "gitea"
|
|
|
|
def test_forge_type_detects_github(self):
|
|
assert ingestor._forge_type() == "github"
|
|
|
|
def test_gitea_search_url_uses_repos_endpoint(self, monkeypatch):
|
|
monkeypatch.setattr(ingestor, "GITHUB_API_BASE", "https://git.cloudinit.dev/api/v1")
|
|
url = ingestor._issues_search_url("acdl", "acdl", "contract-123")
|
|
assert "git.cloudinit.dev/api/v1" in url
|
|
assert "/repos/acdl/acdl/issues" in url
|
|
assert "/search/issues" not in url
|
|
|
|
def test_github_search_url_uses_search_endpoint(self):
|
|
url = ingestor._issues_search_url("acdl", "acdl", "contract-123")
|
|
assert "api.github.com/search/issues" in url
|
|
assert "repo:acdl/acdl" in url
|
|
|
|
def test_create_url_uses_api_base(self, monkeypatch):
|
|
monkeypatch.setattr(ingestor, "GITHUB_API_BASE", "https://git.cloudinit.dev/api/v1")
|
|
url = ingestor._issues_create_url("acdl", "acdl")
|
|
assert url == "https://git.cloudinit.dev/api/v1/repos/acdl/acdl/issues"
|
|
|
|
def test_comments_url_uses_api_base(self, monkeypatch):
|
|
monkeypatch.setattr(ingestor, "GITHUB_API_BASE", "https://git.cloudinit.dev/api/v1")
|
|
url = ingestor._issue_comments_url("acdl", "acdl", 42)
|
|
assert url == "https://git.cloudinit.dev/api/v1/repos/acdl/acdl/issues/42/comments" |