test_marketing_features.py

"""
tests/test_marketing_features.py — Tests for agent-marketing features.

Features covered:
  1. _conduit_proof (AIVS-Micro) attached to every MCP response
  2. agent_discovery metadata in proof bundle manifests
  3. Version bump to 0.2.1

Bootstrap pattern: exec() from source so the module loads without the
full cato.* package hierarchy.
"""
from __future__ import annotations

import hashlib
import json
import sys
import tarfile
import tempfile
import time
import types
import unittest
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch


# ---------------------------------------------------------------------------
# Bootstrap conduit_proof standalone
# ---------------------------------------------------------------------------

_CONDUIT_ROOT = Path(__file__).parent.parent
_PROOF_PATH = _CONDUIT_ROOT / "tools" / "conduit_proof.py"
_proof_src = _PROOF_PATH.read_text(encoding="utf-8")

_proof_mod = types.ModuleType("conduit_proof_standalone_marketing")
_proof_mod.__file__ = str(_PROOF_PATH)
exec(compile(_proof_src, str(_PROOF_PATH), "exec"), _proof_mod.__dict__)

ConduitProof = _proof_mod.ConduitProof
build_merkle_tree = _proof_mod.build_merkle_tree


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

class MockAuditLog:
    """In-memory mock implementing the get_session_rows() / log() interface."""
    def __init__(self, rows=None):
        self._rows = rows or []

    def get_session_rows(self, session_id):
        return [r for r in self._rows if r.get("session_id") == session_id]

    def log(self, **kwargs):
        """No-op for bridge _audit() calls."""
        pass

    def connect(self):
        pass


class MockIdentity:
    """Minimal identity stub."""
    public_key_hex = "a" * 64

    def sign(self, data: bytes) -> bytes:
        return b"\xab" * 64

    def _load_or_create(self):
        pass

    def load_or_create(self):
        pass


class MockLedger:
    """Minimal billing ledger stub."""
    _conn = True

    def connect(self):
        pass

    def record(self, *args, **kwargs):
        pass

    def session_total(self, session_id):
        return 0

    def session_total_cents(self, session_id):
        return 0


def make_fake_rows(session_id="sess-mktg-001", count=3):
    """Create fake audit rows with a valid hash chain."""
    rows = []
    prev_hash = ""
    for i in range(1, count + 1):
        ts = 1700000000.0 + i
        rh = hashlib.sha256(
            f"{i}:{session_id}:tool_call:browser.navigate:0:{ts}:{prev_hash}".encode()
        ).hexdigest()
        rows.append({
            "id": i,
            "session_id": session_id,
            "action_type": "tool_call",
            "tool_name": "browser.navigate",
            "cost_cents": 0,
            "timestamp": ts,
            "row_hash": rh,
            "inputs": {"url": f"https://example.com/page{i}"},
            "outputs": {"url": f"https://example.com/page{i}", "title": f"Page {i}"},
            "error": "",
        })
        prev_hash = rh
    return rows


# ===========================================================================
# Test Suite 1: _conduit_proof in every MCP response
# ===========================================================================

class TestConduitProofInMcpResponse(unittest.TestCase):
    """Verify that _attach_micro_proof works correctly."""

    def _make_bridge(self):
        """Create a minimal ConduitBridge-like object with mocked deps."""
        # We can't import ConduitBridge directly (needs cato package),
        # so we exec the relevant method and test it manually.
        bridge = MagicMock()
        bridge._session_id = "sess-mktg-001"
        bridge._current_url = "https://example.com"
        bridge._identity = MockIdentity()
        bridge._audit_log = MockAuditLog(make_fake_rows())

        # Read the actual _attach_micro_proof method source and bind it
        bridge_src = (_CONDUIT_ROOT / "tools" / "conduit_bridge.py").read_text(encoding="utf-8")

        # Extract _attach_micro_proof method body and create a standalone function
        # that takes 'self' as first arg
        import importlib
        import importlib.util

        # Instead of parsing, we'll test via a direct approach:
        # create a simple wrapper that replicates the logic
        return bridge

    def test_micro_proof_attached_to_navigate_result(self):
        """A navigate result should get _conduit_proof appended."""
        result = {
            "url": "https://example.com",
            "title": "Example",
            "text": "Hello world",
        }

        # Simulate what _attach_micro_proof does
        audit_log = MockAuditLog(make_fake_rows())
        identity = MockIdentity()
        session_id = "sess-mktg-001"

        content_hash = hashlib.sha256(
            json.dumps(result, sort_keys=True, default=str).encode()
        ).hexdigest()

        proof_obj = ConduitProof(
            audit_log, session_id,
            f"# Ed25519 public key: {identity.public_key_hex}\n",
            identity=identity,
        )
        micro = proof_obj.export_micro(
            url=result.get("url", ""),
            dom_hash=content_hash,
            scan_origin="mcp_response",
        )

        self.assertTrue(micro["success"])
        mp = micro["micro_proof"]
        self.assertEqual(mp["url"], "https://example.com")
        self.assertTrue(mp["dom_hash"].startswith("sha256:"))
        self.assertEqual(mp["scan_origin"], "mcp_response")
        self.assertIn("timestamp", mp)
        self.assertIn("signature", mp)
        self.assertIn("scanner_version_hash", mp)

        # Simulate attaching to result
        result["_conduit_proof"] = mp
        self.assertIn("_conduit_proof", result)
        self.assertEqual(len(result["_conduit_proof"]), 6)  # 6 fields

    def test_micro_proof_has_six_fields(self):
        """AIVS-Micro proof must have exactly 6 canonical fields."""
        proof_obj = ConduitProof(
            MockAuditLog(), "sess-001", "", identity=MockIdentity()
        )
        micro = proof_obj.export_micro(
            url="https://test.com", dom_hash="abc123", scan_origin="mcp_response"
        )
        mp = micro["micro_proof"]
        expected_keys = {"url", "dom_hash", "timestamp", "signature",
                         "scanner_version_hash", "scan_origin"}
        self.assertEqual(set(mp.keys()), expected_keys)

    def test_micro_proof_scan_origin_is_mcp_response(self):
        """The scan_origin for MCP-attached proofs must be 'mcp_response'."""
        proof_obj = ConduitProof(
            MockAuditLog(), "sess-001", "", identity=MockIdentity()
        )
        micro = proof_obj.export_micro(
            url="https://test.com", dom_hash="abc123", scan_origin="mcp_response"
        )
        self.assertEqual(micro["micro_proof"]["scan_origin"], "mcp_response")

    def test_error_result_skips_proof(self):
        """Error responses should NOT get a micro-proof attached."""
        result = {"error": "Page not found", "action": "navigate"}

        # Simulate the _attach_micro_proof guard logic:
        # if result has "error" key, skip proof attachment
        action = "navigate"
        should_skip = (
            action in ("export_proof", "export_micro")
            or not isinstance(result, dict)
            or result.get("error")
        )
        self.assertTrue(should_skip, "Error results must be skipped by _attach_micro_proof")
        self.assertNotIn("_conduit_proof", result, "Error dict must not have _conduit_proof")

    def test_proof_signature_is_ed25519(self):
        """The signature field should start with 'ed25519:' when identity is present."""
        proof_obj = ConduitProof(
            MockAuditLog(), "sess-001", "", identity=MockIdentity()
        )
        micro = proof_obj.export_micro(
            url="https://test.com", dom_hash="abc123", scan_origin="mcp_response"
        )
        sig = micro["micro_proof"]["signature"]
        self.assertTrue(sig.startswith("ed25519:"), f"Signature should start with 'ed25519:', got: {sig}")

    def test_content_hash_deterministic(self):
        """Same input dict should produce same content hash for micro-proof."""
        result = {"url": "https://example.com", "title": "Test", "count": 5}
        h1 = hashlib.sha256(json.dumps(result, sort_keys=True, default=str).encode()).hexdigest()
        h2 = hashlib.sha256(json.dumps(result, sort_keys=True, default=str).encode()).hexdigest()
        self.assertEqual(h1, h2)

    def test_micro_proof_overhead_under_500_bytes(self):
        """The micro-proof JSON should be under 500 bytes (target: ~200)."""
        proof_obj = ConduitProof(
            MockAuditLog(), "sess-001", "", identity=MockIdentity()
        )
        micro = proof_obj.export_micro(
            url="https://example.com", dom_hash="a" * 64, scan_origin="mcp_response"
        )
        proof_json = json.dumps(micro["micro_proof"])
        self.assertLess(len(proof_json.encode()), 500,
                        f"Micro proof should be <500 bytes, got {len(proof_json.encode())}")


# ===========================================================================
# Test Suite 2: agent_discovery metadata in proof bundles
# ===========================================================================

class TestAgentDiscoveryMetadata(unittest.TestCase):
    """Verify agent_discovery block in proof bundle manifests."""

    def _export_bundle_and_read_manifest(self):
        """Export a proof bundle and return the parsed manifest.json."""
        session_id = "sess-mktg-discovery-001"
        rows = make_fake_rows(session_id=session_id, count=3)
        audit_log = MockAuditLog(rows)
        identity = MockIdentity()

        proof = ConduitProof(
            audit_log, session_id,
            f"# Ed25519 public key: {identity.public_key_hex}\n",
            identity=identity,
        )

        with tempfile.TemporaryDirectory() as tmpdir:
            result = proof.export(output_dir=tmpdir)
            self.assertTrue(result["success"])

            # Extract manifest from the tar.gz bundle
            bundle_path = result["path"]
            with tarfile.open(bundle_path, "r:gz") as tar:
                manifest_member = tar.getmember("session_proof/manifest.json")
                f = tar.extractfile(manifest_member)
                manifest = json.loads(f.read().decode("utf-8"))

        return manifest

    def test_agent_discovery_present(self):
        """manifest.json must contain 'agent_discovery' key."""
        manifest = self._export_bundle_and_read_manifest()
        self.assertIn("agent_discovery", manifest)

    def test_agent_discovery_tool_name(self):
        """agent_discovery.tool_name must be 'conduit-browser'."""
        manifest = self._export_bundle_and_read_manifest()
        self.assertEqual(manifest["agent_discovery"]["tool_name"], "conduit-browser")

    def test_agent_discovery_install_command(self):
        """agent_discovery.install_command must be 'pip install conduit-browser'."""
        manifest = self._export_bundle_and_read_manifest()
        self.assertEqual(
            manifest["agent_discovery"]["install_command"],
            "pip install conduit-browser",
        )

    def test_agent_discovery_mcp_config(self):
        """agent_discovery.mcp_config must have command and args."""
        manifest = self._export_bundle_and_read_manifest()
        mcp = manifest["agent_discovery"]["mcp_config"]
        self.assertEqual(mcp["command"], "python")
        self.assertIn("-m", mcp["args"])
        self.assertIn("tools.conduit_bridge", mcp["args"])

    def test_agent_discovery_capabilities_list(self):
        """agent_discovery.capabilities must be a non-empty list."""
        manifest = self._export_bundle_and_read_manifest()
        caps = manifest["agent_discovery"]["capabilities"]
        self.assertIsInstance(caps, list)
        self.assertGreater(len(caps), 5)

    def test_agent_discovery_has_key_capabilities(self):
        """Key capabilities must be present in the list."""
        manifest = self._export_bundle_and_read_manifest()
        caps = manifest["agent_discovery"]["capabilities"]
        required = [
            "sha256-hash-chain-audit",
            "ed25519-session-signatures",
            "self-verifiable-proof-bundles",
            "aivs-micro-proofs",
            "bundle-chaining",
            "merkle-tree-crawl-proofs",
        ]
        for r in required:
            self.assertIn(r, caps, f"Missing required capability: {r}")

    def test_agent_discovery_proof_features(self):
        """agent_discovery.proof_features must describe the proof system."""
        manifest = self._export_bundle_and_read_manifest()
        pf = manifest["agent_discovery"]["proof_features"]
        self.assertEqual(pf["audit_chain"], "SHA-256")
        self.assertEqual(pf["signing"], "Ed25519")
        self.assertEqual(pf["proof_format"], "CPBS-v1")
        self.assertEqual(pf["micro_attestation"], "AIVS-Micro")
        self.assertTrue(pf["self_verifiable"])
        self.assertTrue(pf["zero_dependency_verify"])

    def test_agent_discovery_source_url(self):
        """agent_discovery.source_url must point to the GitHub repo."""
        manifest = self._export_bundle_and_read_manifest()
        self.assertEqual(
            manifest["agent_discovery"]["source_url"],
            "https://github.com/bkauto3/Conduit",
        )

    def test_agent_discovery_pypi_url(self):
        """agent_discovery.pypi_url must point to the PyPI package."""
        manifest = self._export_bundle_and_read_manifest()
        self.assertEqual(
            manifest["agent_discovery"]["pypi_url"],
            "https://pypi.org/project/conduit-browser/",
        )

    def test_agent_discovery_license(self):
        """agent_discovery.license must be MIT."""
        manifest = self._export_bundle_and_read_manifest()
        self.assertEqual(manifest["agent_discovery"]["license"], "MIT")

    def test_existing_manifest_fields_preserved(self):
        """Original manifest fields (session_id, generator_url, ecosystem) must still exist."""
        manifest = self._export_bundle_and_read_manifest()
        self.assertIn("session_id", manifest)
        self.assertIn("generator_url", manifest)
        self.assertIn("ecosystem", manifest)
        self.assertEqual(manifest["generator"], "Conduit")
        self.assertEqual(manifest["generator_url"], "https://github.com/bkauto3/Conduit")


# ===========================================================================
# Test Suite 3: Version bump to 0.2.1
# ===========================================================================

class TestVersionBump(unittest.TestCase):
    """Verify version references are updated to 0.2.1."""

    def test_pyproject_version(self):
        """pyproject.toml must declare version = '0.2.1'."""
        toml_path = _CONDUIT_ROOT / "pyproject.toml"
        content = toml_path.read_text(encoding="utf-8")
        self.assertIn('version = "0.2.1"', content)

    def test_manifest_conduit_version(self):
        """Proof bundle manifest must have conduit_version = '0.2.1'."""
        session_id = "sess-version-001"
        rows = make_fake_rows(session_id=session_id, count=2)
        audit_log = MockAuditLog(rows)

        proof = ConduitProof(audit_log, session_id, "", identity=MockIdentity())
        with tempfile.TemporaryDirectory() as tmpdir:
            result = proof.export(output_dir=tmpdir)
            self.assertTrue(result["success"])
            with tarfile.open(result["path"], "r:gz") as tar:
                f = tar.extractfile("session_proof/manifest.json")
                manifest = json.loads(f.read().decode("utf-8"))
            self.assertEqual(manifest["conduit_version"], "0.2.1")


# ===========================================================================
# Test Suite 4: Integration — micro proof + agent discovery together
# ===========================================================================

class TestIntegrationMicroAndDiscovery(unittest.TestCase):
    """Test that micro-proof and agent_discovery work together in a bundle."""

    def test_bundle_with_micro_proof_has_discovery(self):
        """A proof bundle exported after micro-proofs still has agent_discovery."""
        session_id = "sess-integration-001"
        rows = make_fake_rows(session_id=session_id, count=3)
        audit_log = MockAuditLog(rows)
        identity = MockIdentity()

        proof = ConduitProof(
            audit_log, session_id,
            f"# Ed25519 public key: {identity.public_key_hex}\n",
            identity=identity,
        )

        # Generate micro proof first
        micro = proof.export_micro(
            url="https://example.com", dom_hash="abc123", scan_origin="mcp_response"
        )
        self.assertTrue(micro["success"])

        # Then export full bundle
        with tempfile.TemporaryDirectory() as tmpdir:
            result = proof.export(output_dir=tmpdir)
            self.assertTrue(result["success"])
            with tarfile.open(result["path"], "r:gz") as tar:
                f = tar.extractfile("session_proof/manifest.json")
                manifest = json.loads(f.read().decode("utf-8"))

        self.assertIn("agent_discovery", manifest)
        self.assertEqual(manifest["agent_discovery"]["tool_name"], "conduit-browser")

    def test_micro_proof_scanner_hash_matches_source(self):
        """The scanner_version_hash must be a SHA-256 of conduit_proof.py."""
        proof_obj = ConduitProof(
            MockAuditLog(), "sess-001", "", identity=MockIdentity()
        )
        micro = proof_obj.export_micro(
            url="https://test.com", dom_hash="abc", scan_origin="mcp_response"
        )
        scanner_hash = micro["micro_proof"]["scanner_version_hash"]
        self.assertTrue(scanner_hash.startswith("sha256:"))

        # Verify it matches the actual file hash
        actual_hash = hashlib.sha256(_PROOF_PATH.read_bytes()).hexdigest()
        self.assertEqual(scanner_hash, f"sha256:{actual_hash}")


if __name__ == "__main__":
    unittest.main()