test_ohm.py

#!/usr/bin/env python3
"""
Ohm Truth & Integrity Test Suite
=================================
Validates the correctness, deterministic nature, and security integrity
of the Ohm Truth module, including row-checksum validation, local
memory lookups, zero-dependency search scrapers, and plug-in decorators.
"""

import os
import sys
import json
import time
import asyncio
import unittest
import sqlite3
from unittest.mock import MagicMock, patch

# Ensure import works
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from verity import (
    OhmMemory,
    OhmWebClient,
    OhmTruthEngine,
    ohm_truth_shield,
    TruthError
)

class TestOhmMemory(unittest.TestCase):
    """Validates the Memory Entropy Level 0 database and cryptographic row-sealing."""

    def setUp(self):
        self.db_path = "logs/test_ohm_library.db"
        # Ensure clean slate
        if os.path.exists(self.db_path):
            os.remove(self.db_path)
        self.memory = OhmMemory(db_path=self.db_path)

    def tearDown(self):
        if os.path.exists(self.db_path):
            try:
                os.remove(self.db_path)
            except Exception:
                pass

    def test_save_and_retrieve_fact(self):
        """Verifies saving a fact, hashing it, and retrieving it works perfectly."""
        saved = self.memory.save_fact("Guido van Rossum", "created", "Python", "https://python.org")
        self.assertTrue(saved)

        # Retrieve
        fact = self.memory.get_fact("Guido van Rossum", "created")
        self.assertIsNotNone(fact)
        self.assertEqual(fact["object"], "Python")
        self.assertEqual(fact["source_url"], "https://python.org")
        self.assertEqual(fact["integrity"], "verified_level_0")

    def test_integrity_tampering_detection(self):
        """
        Simulates manual database corruption (entropy drift) to verify
        that the system flags corrupted facts and raises a ValueError.
        """
        # Save a valid fact
        self.memory.save_fact("Boiling point of water", "is", "100 degrees Celsius", "https://wikipedia.org")

        # Now, simulate a direct bit-flip/tampering inside the database
        # We manually update the object of the fact without updating the row_hash!
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute("UPDATE ohm_facts SET object = '400 degrees Celsius' WHERE subject = 'Boiling point of water'")
        conn.commit()
        conn.close()

        # Reading the corrupted fact must throw an error due to row checksum mismatch!
        # This guarantees memory entropy level 0.
        with self.assertRaises(ValueError) as context:
            self.memory.get_fact("Boiling point of water", "is")
        
        self.assertIn("Memory Entropy Drift detected", str(context.exception))


class TestOhmWebClient(unittest.TestCase):
    """Validates the standard-library based Wikipedia and DuckDuckGo parsers."""

    @patch("urllib.request.urlopen")
    def test_wikipedia_parser(self, mock_urlopen):
        """Mocks MediaWiki APIs and verifies article summary extraction."""
        client = OhmWebClient()

        # Mock the search call first, then the extract call
        mock_search_json = json.dumps({
            "query": {
                "search": [
                    {"title": "Python (programming language)"}
                ]
            }
        }).encode('utf-8')

        mock_extract_json = json.dumps({
            "query": {
                "pages": {
                    "12345": {
                        "extract": "Python is a high-level programming language created by Guido van Rossum."
                    }
                }
            }
        }).encode('utf-8')

        # Setup mock responses
        mock_resp_search = MagicMock()
        mock_resp_search.__enter__.return_value = mock_resp_search
        mock_resp_search.read.return_value = mock_search_json
        
        mock_resp_extract = MagicMock()
        mock_resp_extract.__enter__.return_value = mock_resp_extract
        mock_resp_extract.read.return_value = mock_extract_json

        mock_urlopen.side_effect = [mock_resp_search, mock_resp_extract]

        results = client.query_wikipedia("Guido van Rossum")
        self.assertEqual(len(results), 1)
        self.assertEqual(results[0]["title"], "Python (programming language)")
        self.assertIn("Guido van Rossum", results[0]["summary"])
        self.assertEqual(results[0]["url"], "https://en.wikipedia.org/wiki/Python+%28programming+language%29")

    @patch("urllib.request.urlopen")
    def test_duckduckgo_parser(self, mock_urlopen):
        """Mocks DDG HTML scrapers and verifies snippet extracts."""
        client = OhmWebClient()

        # Mock Instant Answer API empty, then HTML search results
        mock_html = """
        <html>
        <body>
            <div class="links">
                <a class="result__a" href="https://example.com/l/?kh=-1&uddg=https%3A%2F%2Fpython.org%2Fabout">Python Official Site</a>
                <a class="result__snippet" href="#">Python is an amazing programming language first released in 1991.</a>
            </div>
        </body>
        </html>
        """.encode('utf-8')

        mock_resp_api = MagicMock()
        mock_resp_api.__enter__.return_value = mock_resp_api
        mock_resp_api.read.return_value = b"{}"  # Empty Instant Answer

        mock_resp_html = MagicMock()
        mock_resp_html.__enter__.return_value = mock_resp_html
        mock_resp_html.read.return_value = mock_html

        mock_urlopen.side_effect = [mock_resp_api, mock_resp_html]

        results = client.query_duckduckgo("Python released in 1991")
        self.assertEqual(len(results), 1)
        self.assertEqual(results[0]["title"], "Python Official Site")
        self.assertEqual(results[0]["url"], "https://python.org/about")
        self.assertIn("released in 1991", results[0]["snippet"])


class TestOhmTruthEngine(unittest.TestCase):
    """Validates logic parsing, cross-referencing, and factual auditing."""

    def setUp(self):
        self.db_path = "logs/test_ohm_engine.db"
        if os.path.exists(self.db_path):
            os.remove(self.db_path)
        
        self.memory = OhmMemory(db_path=self.db_path)
        self.web_client = MagicMock()
        self.engine = OhmTruthEngine(memory=self.memory, client=self.web_client)

    def tearDown(self):
        if os.path.exists(self.db_path):
            try:
                os.remove(self.db_path)
            except Exception:
                pass

    def test_assertion_extraction(self):
        """Verifies parsing heuristic extracts subject/predicate/object correctly."""
        text = "Guido van Rossum created Python in the past. The boiling point of water is 100 Celsius."
        assertions = self.engine.extract_assertions(text)
        
        # Verify Guido creation
        self.assertTrue(any(a[0] == "Guido van Rossum" and a[1] == "created" and "Python" in a[2] for a in assertions))
        # Verify boiling point
        self.assertTrue(any("boiling point" in a[0] and a[1] == "is" and "100" in a[2] for a in assertions))

    def test_verify_assertion_via_cache(self):
        """Verifies lookups hit the local database first and succeed."""
        self.memory.save_fact("Kyoto population", "is", "1.46 million", "https://kyoto.gov")
        
        is_verified, source, explanation = self.engine.verify_assertion("Kyoto population", "is", "1.46 million")
        self.assertTrue(is_verified)
        self.assertEqual(source, "https://kyoto.gov")
        self.assertIn("local Ohm Library", explanation)

    def test_verify_assertion_via_web_success(self):
        """Verifies that search results are cross-referenced and written to cache on success."""
        # Setup mock web sources confirming that Anders Celsius established the Celsius scale in 1742
        self.web_client.query_wikipedia.return_value = [
            {"title": "Celsius Scale", "summary": "Anders Celsius established the Celsius scale in 1742.", "url": "https://wikipedia.org/celsius"}
        ]
        self.web_client.query_duckduckgo.return_value = [
            {"title": "Anders Celsius", "snippet": "In 1742, Anders Celsius created the Celsius temperature scale.", "url": "https://celsius.org"}
        ]

        is_verified, source, explanation = self.engine.verify_assertion("Anders Celsius", "established", "Celsius scale in 1742")
        self.assertTrue(is_verified)
        self.assertEqual(source, "https://wikipedia.org/celsius")

        # Must have been written to local memory
        cached = self.memory.get_fact("Anders Celsius", "established")
        self.assertIsNotNone(cached)
        self.assertEqual(cached["object"], "Celsius scale in 1742")


class TestOhmTruthShield(unittest.TestCase):
    """Validates the plug-and-play decorator wrapper functionality."""

    def setUp(self):
        self.db_path = "logs/test_ohm_shield.db"
        if os.path.exists(self.db_path):
            os.remove(self.db_path)
        
        # Seed local memory so decorator passes
        self.memory = OhmMemory(db_path=self.db_path)
        self.memory.save_fact("Guido van Rossum", "created", "Python", "https://python.org")

    def tearDown(self):
        if os.path.exists(self.db_path):
            try:
                os.remove(self.db_path)
            except Exception:
                pass

    @patch("verity.ohm.OhmMemory")
    def test_decorator_passing(self, mock_memory_class):
        """Verifies that truthful functions execute and return responses unmodified."""
        mock_memory_class.return_value = self.memory

        @ohm_truth_shield(strict=True)
        def my_ai_func():
            return "Guido van Rossum created Python."

        output = my_ai_func()
        self.assertEqual(output, "Guido van Rossum created Python.")

    @patch("verity.ohm.OhmMemory")
    @patch("verity.ohm.OhmWebClient")
    def test_decorator_refusal_strict(self, mock_client_class, mock_memory_class):
        """Verifies that fabricated facts are caught and blocked in strict mode."""
        mock_memory_class.return_value = self.memory
        
        # Mock web queries to return nothing (representing unverifiable/hallucinated claim)
        mock_client = mock_client_class.return_value
        mock_client.query_wikipedia.return_value = []
        mock_client.query_duckduckgo.return_value = []

        @ohm_truth_shield(strict=True)
        def my_fabricated_ai_func():
            return "Kyoto population is 40 million."

        output = my_fabricated_ai_func()
        self.assertIn("=== OHM INTEGRITY REFUSAL ===", output)
        self.assertIn("factual fabrication", output.lower())

    @patch("verity.ohm.OhmMemory")
    @patch("verity.ohm.OhmWebClient")
    def test_decorator_annotation_non_strict(self, mock_client_class, mock_memory_class):
        """Verifies that in non-strict mode fabrications are annotated inline."""
        mock_memory_class.return_value = self.memory
        
        mock_client = mock_client_class.return_value
        mock_client.query_wikipedia.return_value = []
        mock_client.query_duckduckgo.return_value = []

        @ohm_truth_shield(strict=False)
        def my_fabricated_ai_func():
            return "Kyoto population is 40 million."

        output = my_fabricated_ai_func()
        self.assertIn("Kyoto population is 40 million [OHM_FABRICATION_RISK", output)

    @patch("verity.ohm.OhmMemory")
    def test_decorator_async_passing(self, mock_memory_class):
        """Verifies that an asynchronous decorated function works perfectly."""
        mock_memory_class.return_value = self.memory

        @ohm_truth_shield(strict=True)
        async def my_async_ai_func():
            return "Guido van Rossum created Python."

        # Run async function using asyncio
        loop = asyncio.new_event_loop()
        try:
            output = loop.run_until_complete(my_async_ai_func())
            self.assertEqual(output, "Guido van Rossum created Python.")
        finally:
            loop.close()


class TestServerOhmIntegration(unittest.TestCase):
    """Verifies the METAPRIME server.py integration of Ohm Truth Engine."""

    def setUp(self):
        import socket
        import threading
        # Find a free port
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.bind(('127.0.0.1', 0))
        self.port = s.getsockname()[1]
        s.close()

        # Import server's Handler
        sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
        from server import Handler
        import socketserver

        # Create the TCPServer
        self.httpd = socketserver.TCPServer(("127.0.0.1", self.port), Handler)
        self.thread = threading.Thread(target=self.httpd.serve_forever, daemon=True)
        self.thread.start()
        time.sleep(0.5) # Allow server socket binding

    def tearDown(self):
        self.httpd.shutdown()
        self.httpd.server_close()
        self.thread.join(timeout=1.0)

    def test_truth_api_endpoint_post(self):
        """Verifies POST /api/ohm returns correct truth reports."""
        import urllib.request
        import json
        
        url = f"http://127.0.0.1:{self.port}/api/ohm"
        payload = json.dumps({"text": "Guido van Rossum created Python."}).encode('utf-8')
        
        req = urllib.request.Request(
            url,
            data=payload,
            headers={"Content-Type": "application/json"},
            method="POST"
        )
        
        try:
            with urllib.request.urlopen(req, timeout=5.0) as response:
                self.assertEqual(response.status, 200)
                data = json.loads(response.read().decode('utf-8'))
                self.assertIn("is_truthful", data)
                self.assertIn("assertions_audited", data)
        except Exception as e:
            self.fail(f"POST /api/ohm failed: {e}")

    def test_truth_api_endpoint_get(self):
        """Verifies GET /api/ohm returns correct truth reports."""
        import urllib.request
        import json
        import urllib.parse
        
        query = urllib.parse.quote_plus("Guido van Rossum created Python.")
        url = f"http://127.0.0.1:{self.port}/api/ohm?text={query}"
        
        try:
            with urllib.request.urlopen(url, timeout=5.0) as response:
                self.assertEqual(response.status, 200)
                data = json.loads(response.read().decode('utf-8'))
                self.assertIn("is_truthful", data)
        except Exception as e:
            self.fail(f"GET /api/ohm failed: {e}")


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