tests/test_browser.py

"""Real Chromium integration tests; goals/actions are test fixtures, not a policy."""
import json
from pathlib import Path
import tempfile
import unittest
from playwright.sync_api import sync_playwright
from baim.actions import Action, Kind, Decision
from baim.authority import Authority
from baim.browser import Browser
from baim.memory import Recorder
from baim.runtime import Runtime


class BrowserTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.pw = sync_playwright().start()
        cls.chromium = cls.pw.chromium.launch(headless=True)

    @classmethod
    def tearDownClass(cls):
        cls.chromium.close()
        cls.pw.stop()

    def setUp(self):
        self.context = self.chromium.new_context()
        self.browser = Browser(self.context)
        self.authority = Authority()
        self.authority.replace('Fill the labelled field and activate the button')
        self.browser.page.set_content('''<label>Destination<input></label>
            <button onclick="document.querySelector('output').textContent='Accepted'">Proceed</button>
            <output>Pending</output><input type=password aria-label="Secret">
            <div id=host></div><iframe srcdoc="<button>Inside</button>"></iframe>
            <script>document.querySelector('#host').attachShadow({mode:'open'}).innerHTML='<button>Shadow</button>'</script>''')
        self.browser.observe()
        self.runtime = Runtime(self.browser, self.authority, lambda *_: True,
            completion=lambda b: 'acknowledgment_visible' if b.page.locator('output').inner_text() == 'Accepted' else None)

    def tearDown(self):
        self.context.close()

    def ref(self, name):
        return next(e.ref for e in self.browser.state.elements if e.name == name)

    def execute(self, kind, *args):
        return self.runtime.execute(Decision(self.authority.ticket(self.browser.state), Action(kind, args)))

    def test_multistep_execution_and_completion(self):
        self.assertEqual(self.execute(Kind.FINISH).code, 'COMPLETION_NOT_PROVEN')
        self.assertEqual(self.execute(Kind.TYPE, self.ref('Destination'), 'Tokyo').status, 'ok')
        self.browser.observe()
        self.assertEqual(self.execute(Kind.CLICK, self.ref('Proceed')).status, 'ok')
        self.browser.observe()
        self.assertEqual(self.execute(Kind.FINISH).evidence, 'acknowledgment_visible')
        self.assertEqual(self.browser.page.locator('label input').input_value(), 'Tokyo')

    def test_frame_and_shadow_nodes(self):
        self.assertEqual(self.execute(Kind.CLICK, self.ref('Inside')).status, 'ok')
        self.browser.observe()
        self.assertEqual(self.execute(Kind.CLICK, self.ref('Shadow')).status, 'ok')

    def test_path_optimization_preserves_semantics(self):
        optimized = self.browser.state
        self.browser.path_mode = 'naive'
        naive = self.browser.observe()
        self.assertEqual(optimized.elements, naive.elements)
        self.assertEqual(optimized.state_hash, naive.state_hash)

    def test_changed_and_detached_targets(self):
        ref = self.ref('Proceed')
        self.browser.page.locator('button').first.evaluate("el => el.textContent = 'Changed'")
        self.assertEqual(self.execute(Kind.CLICK, ref).code, 'CHANGED_TARGET')
        self.browser.page.locator('button').first.evaluate('el => el.remove()')
        self.assertEqual(self.execute(Kind.CLICK, ref).code, 'DETACHED_TARGET')

    def test_old_task_cannot_execute_or_finish(self):
        decision = Decision(self.authority.ticket(self.browser.state), Action(Kind.CLICK, (self.ref('Proceed'),)))
        self.authority.replace('New task')
        self.assertEqual(self.runtime.execute(decision).code, 'STALE_AUTHORITY_OR_OBSERVATION')
        self.assertEqual(self.execute(Kind.FINISH).code, 'NO_COMPLETION_VERIFIER')
        self.assertEqual(self.browser.page.locator('output').inner_text(), 'Pending')

    def test_loops_sensitive_targets_and_url_schemes(self):
        secret = next(e.ref for e in self.browser.state.elements if e.sensitive)
        self.assertEqual(self.execute(Kind.TYPE, secret, 'dont-log-me').code, 'SENSITIVE_TARGET_REQUIRES_HOST_FLOW')
        self.assertEqual(self.execute(Kind.NAVIGATE, 'javascript:alert(1)').code, 'UNSAFE_URL')
        for _ in range(3):
            self.assertEqual(self.execute(Kind.WAIT, 0).status, 'ok')
        self.assertEqual(self.execute(Kind.WAIT, 0).code, 'REPEATED_ACTION')

    def test_recording_contains_no_raw_typed_values(self):
        with tempfile.TemporaryDirectory() as temp:
            path = Path(temp) / 'events.sqlite'
            recorder = Recorder(path)
            self.runtime.recorder = recorder
            try:
                self.assertEqual(self.execute(Kind.TYPE, self.ref('Destination'), 'sensitive-value-123').status, 'ok')
                self.assertEqual(recorder.db.execute('SELECT count(*) FROM events').fetchone()[0], 1)
            finally:
                recorder.close()
            self.assertNotIn(b'sensitive-value-123', path.read_bytes())

    def test_document_navigation_invalidates_all_actions(self):
        self.browser.page.goto('data:text/html,<p>New document</p>')
        self.assertEqual(self.execute(Kind.WAIT, 0).code, 'CHANGED_DOCUMENT')

    def test_host_policy_denies_page_instruction(self):
        self.runtime.permission = lambda *_: False
        self.assertEqual(self.execute(Kind.CLICK, self.ref('Proceed')).code, 'HOST_POLICY_DENIED')
        self.assertEqual(self.browser.page.locator('output').inner_text(), 'Pending')