1import pytest2from akos.composition import joins3from akos.vault.redact import redact_text, redact_obj, register_secret, forget_secret4from akos.provenance import records as R56L = [{"date": "2020", "value": 100.0, "iso": "CAN"}, {"date": "2021", "value": 110.0, "iso": "CAN"}]7Rr = [{"key": "2020-01-02", "CAD": 1.3}, {"key": "2020-01-03", "CAD": 1.5}, {"key": "2021-01-04", "CAD": 1.2}]8910def test_join_refuses_granularity_and_units_and_many_to_many():11 rr = R.normalize(Rr, {"derive": [{"name": "year", "expr": "substr(key,0,4)"}]})12 bad = joins.validate(L, rr, {"granularity": "country-year", "units": {"value": "USD"}}, {"granularity": "day", "units": {"CAD": "CAD/USD"}}, {"on": [{"left": "date", "right": "year"}]})13 assert not bad["ok"] and "granularity mismatch" in bad["errors"][0]14 spec = {"on": [{"left": "date", "right": "year"}], "aggregate": {"side": "right", "by": ["year"], "agg": {"CAD": "mean"}}, "derive": [{"name": "x", "expr": "left.value + right.CAD"}]}15 bad2 = joins.validate(L, rr, {"granularity": "country-year", "units": {"value": "USD"}}, {"granularity": "day", "units": {"CAD": "CAD/USD"}}, spec)16 assert not bad2["ok"] and any("unit mismatch" in e for e in bad2["errors"])17 nm = joins.validate(L + L, Rr, {"granularity": "y", "units": {}}, {"granularity": "y", "units": {}}, {"on": [{"left": "date", "right": "key"}]})18 assert not nm["ok"]192021def test_join_valid_and_deterministic():22 rr = R.normalize(Rr, {"derive": [{"name": "year", "expr": "substr(key,0,4)"}]})23 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"}]}24 v = joins.validate(L, rr, {"granularity": "country-year", "units": {"value": "USD"}}, {"granularity": "day", "units": {"CAD": "CAD/USD"}}, spec)25 assert v["ok"] and v["cardinality"] == "1:1" and v["matched_keys"] == 2 and v["units"]["value_cad"] == "CAD"26 out = joins.run(L, rr, spec)27 assert len(out) == 2 and abs(out[0]["value_cad"] - 140.0) < 1e-928 assert R.rows_hash(out) == R.rows_hash(joins.run(L, rr, spec))29 unsafe = joins.validate(L, rr, {}, {}, {"on": [{"left": "date", "right": "year"}], "aggregate": {"side": "right", "by": ["year"], "agg": {"CAD": "mean"}}, "derive": [{"name": "x", "expr": "__import__('os')"}]})30 assert not unsafe["ok"] and any("unsupported expression" in e for e in unsafe["errors"])31 with pytest.raises(joins.JoinError):32 joins.run(L, rr, {"on": [{"left": "date", "right": "year"}], "derive": [{"name": "x", "expr": "open('/etc/passwd')"}]})333435def test_redaction():36 register_secret("super-secret-token-123456")37 assert "super-secret-token-123456" not in redact_text("Authorization: Bearer super-secret-token-123456 ok")38 assert redact_obj({"password": "hunter2hunter2", "nested": {"api_key": "abcdefgh1234"}})["password"] == "[REDACTED]"39 assert "[REDACTED]" in redact_text("token=akos_abcdefghijklmnopqrstuvwxyz0123456789")40 forget_secret("super-secret-token-123456")414243async def test_audit_chain_detects_tampering(pool):44 from akos import audit45 async with pool.acquire() as c:46 await audit.record(c, actor_type="test", actor_id="t", action="a1")47 await audit.record(c, actor_type="test", actor_id="t", action="a2", details={"k": "v"})48 assert (await audit.verify_chain(c))["ok"]49 with pytest.raises(Exception):50 await c.execute("UPDATE audit_events SET details = '{}' WHERE action = 'a2'") # trigger: append-only51 with pytest.raises(Exception):52 await c.execute("DELETE FROM audit_events WHERE action = 'a1'")535455async def test_audit_concurrent_appends_stay_linear(pool):56 import asyncio57 from akos import audit5859 async def one(i):60 async with pool.acquire() as c:61 await audit.record(c, actor_type="test", actor_id=str(i), action="concurrent")62 await asyncio.gather(*(one(i) for i in range(25)))63 async with pool.acquire() as c:64 v = await audit.verify_chain(c)65 assert v["ok"] and v["linear"] and v["content_ok"], v66