# A read-only endpoint consumer must not need the host's mutation authority.
import pytest
from hearth import Scene, Box
from hearth.kernel import Capability, Contract, Port, Grant, Plan
from hearth.space import Frame
from hearth.blocks import state_of
from hearth_extensions import Endpoint, EndpointContact


class Landing:

    def capability(self):
        return Capability('test.landing', ('landing',))

    def negotiate(self, ctx, parameters):
        return Contract(Box((-1, 0, -1), (1, 0, 1)), ports=(Port('access', 'access', Frame((0, 1, 0)), Box((-1, 0, -1), (1, 2, 1)), 4, 'resurface'),), grants=(Grant('resurface', Box((-1, 0, -1), (1, 0, 1)), ('surface',), ('path',), 9),))

    def realize(self, ctx, contract):
        p = Plan()
        p.fill(contract.envelope, state_of('stone_bricks'))
        return p


def test_read_only_connection_to_mutable_port():
    s = Scene()
    s.place('landing', Landing())
    endpoint = Endpoint('/landing', 'access')
    before = dict(s.blocks)
    s.place('consumer', EndpointContact(endpoint), bindings=(endpoint.binding(),))
    assert s.blocks == before
    assert ('/consumer', 'connected_to', '/landing') in s.relations
    s.finalize()


def test_connection_does_not_acquire_write_authority():
    from hearth.kernel import ContractError

    class Bad(EndpointContact):

        def realize(self, ctx, contract):
            p = Plan()
            p.block((0, 0, 0), state_of('gold_block'))
            return p

    s = Scene()
    s.place('landing', Landing())
    endpoint = Endpoint('/landing', 'access')
    before = dict(s.blocks)
    with pytest.raises(ContractError, match='write-authority'):
        s.place('bad', Bad(endpoint), bindings=(endpoint.binding(),))
    assert s.blocks == before and '/bad' not in s.nodes


def test_atomic_multi_branch_replay_accepts_explicit_replacements():
    from hearth.components.primitives import Volume
    s = Scene()
    s.place('a', Volume(Box((0, 0, 0), (0, 0, 0)), state_of('stone')))
    s.place('b', Volume(Box((3, 0, 0), (3, 0, 0)), state_of('stone')))
    s.regenerate('/a', dependents=('/b',), replacements={'/b': Volume(Box((3, 0, 0), (4, 0, 0)), state_of('oak_planks'))})
    assert len(s.cells('/b')) == 2
    assert s.inspect((0, 0, 0))['state'] == state_of('stone')


def test_multi_branch_replay_rejects_undeclared_target_and_rolls_back():
    from hearth.components.primitives import Volume
    from hearth.kernel import ContractError
    from hearth.persistence import semantic_digest
    s = Scene()
    s.place('a', Volume(Box((0, 0, 0), (0, 0, 0)), state_of('stone')))
    before = semantic_digest(s)
    with pytest.raises(ContractError, match='replay-targets'):
        s.regenerate('/a', replacements={'/unknown': Landing()})
    assert semantic_digest(s) == before and set(s.nodes) == {'/a'}


def test_importable_generation_has_no_new_native_dependency():
    import subprocess, sys
    from pathlib import Path
    code = '''
import sys,importlib.abc
class NoNative(importlib.abc.MetaPathFinder):
 def find_spec(self, fullname, path=None, target=None):
  if fullname.split('.')[0] in ('numpy','mcio','scipy','shapely','ctypes'):
   raise RuntimeError('Unexpected native/transport dependency: '+fullname)
sys.meta_path.insert(0,NoNative())
from hearth_extensions.assessment import hall_scene
from hearth.persistence import semantic_digest
scene=hall_scene(4,1)
print(semantic_digest(scene))
'''
    result = subprocess.run([sys.executable, '-c', code], cwd=Path(__file__).parents[1], capture_output=True, text=True)
    assert result.returncode == 0, result.stderr
    assert len(result.stdout.strip()) == 64