import pytest
from akos.composition import joins
from akos.vault.redact import redact_text, redact_obj, register_secret, forget_secret
from akos.provenance import records as R

L = [{"date": "2020", "value": 100.0, "iso": "CAN"}, {"date": "2021", "value": 110.0, "iso": "CAN"}]
Rr = [{"key": "2020-01-02", "CAD": 1.3}, {"key": "2020-01-03", "CAD": 1.5}, {"key": "2021-01-04", "CAD": 1.2}]


def test_join_refuses_granularity_and_units_and_many_to_many():
    rr = R.normalize(Rr, {"derive": [{"name": "year", "expr": "substr(key,0,4)"}]})
    bad = joins.validate(L, rr, {"granularity": "country-year", "units": {"value": "USD"}}, {"granularity": "day", "units": {"CAD": "CAD/USD"}}, {"on": [{"left": "date", "right": "year"}]})
    assert not bad["ok"] and "granularity mismatch" in bad["errors"][0]
    spec = {"on": [{"left": "date", "right": "year"}], "aggregate": {"side": "right", "by": ["year"], "agg": {"CAD": "mean"}}, "derive": [{"name": "x", "expr": "left.value + right.CAD"}]}
    bad2 = joins.validate(L, rr, {"granularity": "country-year", "units": {"value": "USD"}}, {"granularity": "day", "units": {"CAD": "CAD/USD"}}, spec)
    assert not bad2["ok"] and any("unit mismatch" in e for e in bad2["errors"])
    nm = joins.validate(L + L, Rr, {"granularity": "y", "units": {}}, {"granularity": "y", "units": {}}, {"on": [{"left": "date", "right": "key"}]})
    assert not nm["ok"]


def test_join_valid_and_deterministic():
    rr = R.normalize(Rr, {"derive": [{"name": "year", "expr": "substr(key,0,4)"}]})
    spec = {"on": [{"left": "date", "right": "year"}], "aggregate": {"side": "right", "by": ["year"], "agg": {"CAD": "mean"}}, "derive": [{"name": "value_cad", "expr": "left.value * right.CAD", "unit": "CAD"}]}
    v = joins.validate(L, rr, {"granularity": "country-year", "units": {"value": "USD"}}, {"granularity": "day", "units": {"CAD": "CAD/USD"}}, spec)
    assert v["ok"] and v["cardinality"] == "1:1" and v["matched_keys"] == 2 and v["units"]["value_cad"] == "CAD"
    out = joins.run(L, rr, spec)
    assert len(out) == 2 and abs(out[0]["value_cad"] - 140.0) < 1e-9
    assert R.rows_hash(out) == R.rows_hash(joins.run(L, rr, spec))
    unsafe = joins.validate(L, rr, {}, {}, {"on": [{"left": "date", "right": "year"}], "aggregate": {"side": "right", "by": ["year"], "agg": {"CAD": "mean"}}, "derive": [{"name": "x", "expr": "__import__('os')"}]})
    assert not unsafe["ok"] and any("unsupported expression" in e for e in unsafe["errors"])
    with pytest.raises(joins.JoinError):
        joins.run(L, rr, {"on": [{"left": "date", "right": "year"}], "derive": [{"name": "x", "expr": "open('/etc/passwd')"}]})


def test_redaction():
    register_secret("super-secret-token-123456")
    assert "super-secret-token-123456" not in redact_text("Authorization: Bearer super-secret-token-123456 ok")
    assert redact_obj({"password": "hunter2hunter2", "nested": {"api_key": "abcdefgh1234"}})["password"] == "[REDACTED]"
    assert "[REDACTED]" in redact_text("token=akos_abcdefghijklmnopqrstuvwxyz0123456789")
    forget_secret("super-secret-token-123456")


async def test_audit_chain_detects_tampering(pool):
    from akos import audit
    async with pool.acquire() as c:
        await audit.record(c, actor_type="test", actor_id="t", action="a1")
        await audit.record(c, actor_type="test", actor_id="t", action="a2", details={"k": "v"})
        assert (await audit.verify_chain(c))["ok"]
        with pytest.raises(Exception):
            await c.execute("UPDATE audit_events SET details = '{}' WHERE action = 'a2'")   # trigger: append-only
        with pytest.raises(Exception):
            await c.execute("DELETE FROM audit_events WHERE action = 'a1'")


async def test_audit_concurrent_appends_stay_linear(pool):
    import asyncio
    from akos import audit

    async def one(i):
        async with pool.acquire() as c:
            await audit.record(c, actor_type="test", actor_id=str(i), action="concurrent")
    await asyncio.gather(*(one(i) for i in range(25)))
    async with pool.acquire() as c:
        v = await audit.verify_chain(c)
    assert v["ok"] and v["linear"] and v["content_ok"], v