test_dynamo_store.py

"""DynamoDB-backed Store adapter (M4 infra follow-up, ADR-0023)."""
from __future__ import annotations

import boto3
import pytest
from moto import mock_aws

from openjobradar.tenancy.context import TenantContext
from openjobradar.tenancy.dynamo_store import DynamoStore

ALICE = TenantContext.from_sub("alice-sub-0000001")
BOB = TenantContext.from_sub("bob-sub-00000002")
TABLE_NAME = "test-tenant-table"


@pytest.fixture
def dynamo_table():
    with mock_aws():
        resource = boto3.resource("dynamodb", region_name="us-east-1")
        table = resource.create_table(
            TableName=TABLE_NAME,
            KeySchema=[
                {"AttributeName": "userId", "KeyType": "HASH"},
                {"AttributeName": "sk", "KeyType": "RANGE"},
            ],
            AttributeDefinitions=[
                {"AttributeName": "userId", "AttributeType": "S"},
                {"AttributeName": "sk", "AttributeType": "S"},
            ],
            BillingMode="PAY_PER_REQUEST",
        )
        table.wait_until_exists()
        yield table


def test_put_get_roundtrip_preserves_python_types(dynamo_table) -> None:
    store = DynamoStore(dynamo_table)
    body = {
        "spent_usd": 4.5,
        "events": 3,
        "active": True,
        "tags": ["a", "b"],
        "nested": {"rate": 0.125, "count": 7},
    }
    store.put(ALICE, "budget_daily", "2026-08-23", body)
    item = store.get(ALICE, "budget_daily", "2026-08-23")

    assert item["spent_usd"] == 4.5 and isinstance(item["spent_usd"], float)
    assert item["events"] == 3 and isinstance(item["events"], int)
    assert item["active"] is True
    assert item["tags"] == ["a", "b"]
    assert item["nested"] == {"rate": 0.125, "count": 7}
    assert item["userId"] == ALICE.user_id
    assert item["sk"] == "budget_daily:2026-08-23"


def test_get_missing_item_returns_none(dynamo_table) -> None:
    store = DynamoStore(dynamo_table)
    assert store.get(ALICE, "settings", "active") is None


def test_delete_reports_whether_an_item_existed(dynamo_table) -> None:
    store = DynamoStore(dynamo_table)
    store.put(ALICE, "settings", "active", {"v": 1})
    assert store.delete(ALICE, "settings", "active") is True
    assert store.delete(ALICE, "settings", "active") is False
    assert store.get(ALICE, "settings", "active") is None


def test_put_rejects_reserved_key_fields(dynamo_table) -> None:
    store = DynamoStore(dynamo_table)
    with pytest.raises(ValueError, match="reserved key field"):
        store.put(ALICE, "postings", "x1", {"userId": "someone-else"})


def test_list_scopes_by_entity_and_prefix_and_sorts_by_sk(dynamo_table) -> None:
    store = DynamoStore(dynamo_table)
    for sk in ("alpha", "beta-42", "gamma"):
        store.put(ALICE, "seen", sk, {"v": sk})
    store.put(ALICE, "postings", "alpha", {"v": "different-entity"})

    all_seen = store.list(ALICE, "seen")
    assert [item["sk"] for item in all_seen] == ["seen:alpha", "seen:beta-42", "seen:gamma"]

    prefixed = store.list(ALICE, "seen", "beta")
    assert [item["v"] for item in prefixed] == ["beta-42"]


def test_list_is_isolated_per_tenant(dynamo_table) -> None:
    store = DynamoStore(dynamo_table)
    store.put(ALICE, "orgs", "acme", {"v": "alice"})
    store.put(BOB, "orgs", "acme", {"v": "bob"})

    assert [item["v"] for item in store.list(ALICE, "orgs")] == ["alice"]
    assert [item["v"] for item in store.list(BOB, "orgs")] == ["bob"]


class _FakePaginatingTable:
    """Deterministic two-page double, so pagination is tested without relying on real
    DynamoDB's ~1MB page-size threshold (impractical to trigger with small test items)."""

    def __init__(self) -> None:
        self.calls: list[dict] = []

    def query(self, **kwargs):
        self.calls.append(kwargs)
        if "ExclusiveStartKey" not in kwargs:
            return {
                "Items": [{"userId": "u", "sk": "e:1", "v": 1}],
                "LastEvaluatedKey": {"userId": "u", "sk": "e:1"},
            }
        return {"Items": [{"userId": "u", "sk": "e:2", "v": 2}]}


def test_list_paginates_across_multiple_query_pages() -> None:
    fake_table = _FakePaginatingTable()
    store = DynamoStore(fake_table)
    items = store.list(ALICE, "e")
    assert [item["sk"] for item in items] == ["e:1", "e:2"]
    assert len(fake_table.calls) == 2
    assert "ExclusiveStartKey" in fake_table.calls[1]