'''
Bindspace declaration, identity and live-capability contracts.

'''
from __future__ import annotations

from pathlib import Path
import os
import sys
from typing import BinaryIO

import msgspec
import pytest
import trio

from tractor.discovery import (
    BindspaceHandle,
    BindspaceIdentity,
    BindspaceOwnership,
    BindspaceSpec,
    CURRENT_NETNS,
    attach_netns,
)
from tractor.discovery import _bindspace
from tractor.msg import ProcessLocal


def test_bindspace_declarations_roundtrip() -> None:
    '''
    Spawn configuration and realized identity must cross actor IPC.

    Encode both frozen structs through msgpack and decode with their
    concrete types, proving names and stable inode identity survive
    without carrying any process-local capability state.

    '''
    values: tuple[
        BindspaceSpec|BindspaceIdentity,
        ...,
    ] = (
        BindspaceSpec(kind='netns', key='tractor-wg0'),
        BindspaceIdentity(
            kind='netns',
            key='tractor-wg0',
            inode=1234,
        ),
    )
    value: BindspaceSpec|BindspaceIdentity
    for value in values:
        encoded: bytes = msgspec.msgpack.encode(value)
        decoded: BindspaceSpec|BindspaceIdentity = (
            msgspec.msgpack.decode(
                encoded,
                type=type(value),
            )
        )
        assert decoded == value


def test_bindspace_handle_pins_local_capability(
    tmp_path: Path,
) -> None:
    '''
    A live handle pins one exact FD and realized identity.

    Open a stand-in platform handle, record its inode in the realized
    identity and construct an owned capability. Prove the generic
    msgspec struct retains that exact local state. Its ability to
    encode ordinary fields is not authority to transfer the handle.

    '''
    token_path: Path = tmp_path / 'bindspace'
    token_path.touch()
    namespace_file: BinaryIO
    with token_path.open('rb') as namespace_file:
        namespace_fd: int = namespace_file.fileno()
        inode: int = token_path.stat().st_ino
        spec: BindspaceSpec = BindspaceSpec(
            kind='netns',
            key='tractor-wg0',
        )
        identity: BindspaceIdentity = BindspaceIdentity(
            kind='netns',
            key='tractor-wg0',
            inode=inode,
        )
        handle: BindspaceHandle = BindspaceHandle(
            spec=spec,
            identity=identity,
            namespace_fd=namespace_fd,
            ownership='owned',
        )

        assert handle.spec is spec
        assert handle.identity is identity
        assert handle.namespace_fd == namespace_file.fileno()
        assert handle.ownership == 'owned'
        assert isinstance(handle, msgspec.Struct)
        assert isinstance(handle, ProcessLocal)
        with pytest.raises(
            TypeError,
            match='_ProcessLocalToken.*unsupported',
        ):
            msgspec.msgpack.encode(handle)


def test_bindspace_handle_rejects_mismatched_identity(
    tmp_path: Path,
) -> None:
    '''
    A name or inode mismatch would make a handle stale authority.

    Construct a requested named spec, then prove both a different
    realized name and an inode not belonging to the supplied FD are
    rejected before either can become a live capability.

    '''
    token_path: Path = tmp_path / 'bindspace'
    token_path.touch()
    spec: BindspaceSpec = BindspaceSpec(
        kind='netns',
        key='tractor-wg0',
    )
    # Keep ownership and FD fixed so only identity changes below.
    ownership: BindspaceOwnership = 'borrowed'
    namespace_file: BinaryIO
    with token_path.open('rb') as namespace_file:
        namespace_fd: int = namespace_file.fileno()
        wrong_name: BindspaceIdentity = BindspaceIdentity(
            kind='netns',
            key='other-wg',
            inode=token_path.stat().st_ino,
        )
        with pytest.raises(
            ValueError,
            match='Spec.key.*Identity.key',
        ):
            BindspaceHandle(
                spec=spec,
                identity=wrong_name,
                namespace_fd=namespace_fd,
                ownership=ownership,
            )

        wrong_inode: BindspaceIdentity = BindspaceIdentity(
            kind='netns',
            key='tractor-wg0',
            inode=token_path.stat().st_ino + 1,
        )
        with pytest.raises(
            ValueError,
            match='FD inode.*identity inode',
        ):
            BindspaceHandle(
                spec=spec,
                identity=wrong_inode,
                namespace_fd=namespace_fd,
                ownership=ownership,
            )


@pytest.mark.parametrize(
    ('model', 'kwargs', 'match'),
    (
        pytest.param(
            BindspaceIdentity,
            {
                'kind': 'netns',
                'key': None,
                'inode': None,
            },
            'must be a positive `int`',
            id='identity-requires-inode',
        ),
        pytest.param(
            BindspaceSpec,
            {'kind': 'vrf'},
            'Unsupported bindspace kind',
            id='spec-rejects-kind',
        ),
        pytest.param(
            BindspaceIdentity,
            {
                'kind': 'vrf',
                'key': 'blue',
                'inode': 1234,
            },
            'Unsupported bindspace kind',
            id='identity-rejects-kind',
        ),
        pytest.param(
            BindspaceSpec,
            {
                'kind': 'netns',
                'key': '../outside',
            },
            'Invalid netns name',
            id='spec-rejects-path',
        ),
        pytest.param(
            BindspaceSpec,
            {
                'kind': 'netns',
                'key': '',
            },
            'BindspaceSpec.key',
            id='spec-rejects-empty-key',
        ),
        pytest.param(
            BindspaceIdentity,
            {
                'kind': 'netns',
                'key': '',
                'inode': 1234,
            },
            'BindspaceIdentity.key',
            id='identity-rejects-empty-key',
        ),
    ),
)
def test_bindspace_models_reject_invalid_values(
    model: type[BindspaceSpec]|type[BindspaceIdentity],
    kwargs: dict[str, object],
    match: str,
) -> None:
    '''
    Direct msgspec construction does not enforce field annotations.

    Parameterize the missing stable inode and future, unimplemented
    kinds. Prove neither serializable model can carry invalid
    identity or provisioning instructions into spawn configuration.

    '''
    with pytest.raises(ValueError, match=match):
        model(**kwargs)  # type: ignore[arg-type]


@pytest.mark.skipif(
    sys.platform != 'linux',
    reason='Linux netns API',
)
def test_attach_current_netns_borrows_and_closes() -> None:
    '''
    The unnamed spec must borrow and pin the caller's current netns.

    Open `/proc/self/ns/net`, prove the yielded handle records its
    stable inode and borrowed ownership, then exit the context and
    prove the exact descriptor was closed without altering the
    namespace itself.

    '''
    async def main() -> int:
        '''
        Borrow the current netns and return its descriptor number.

        '''
        spec: BindspaceSpec = BindspaceSpec(
            kind='netns',
        )
        assert spec.key is CURRENT_NETNS
        async with attach_netns(spec) as handle:
            namespace_fd: int|None = handle.namespace_fd
            assert namespace_fd is not None
            assert handle.spec is spec
            assert handle.identity.key is None
            assert handle.identity.inode == os.fstat(
                namespace_fd
            ).st_ino
            assert handle.ownership == 'borrowed'
            return namespace_fd

    namespace_fd: int = trio.run(main)
    with pytest.raises(OSError):
        os.fstat(namespace_fd)


@pytest.mark.skipif(
    sys.platform != 'linux',
    reason='Linux netns API',
)
def test_attach_named_netns_uses_run_directory(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    '''
    A named spec must resolve only beneath the configured netns dir.

    Replace the run directory with a temporary stand-in, borrow its
    named inode, and prove the context neither deletes the existing
    resource nor leaves its descriptor open after exit.

    '''
    netns_path: Path = tmp_path / 'tractor-wg0'
    netns_path.touch()
    monkeypatch.setattr(
        _bindspace,
        '_NETNS_RUN_DIR',
        tmp_path,
    )

    async def main() -> int:
        '''
        Borrow the named stand-in and return its descriptor number.

        '''
        spec: BindspaceSpec = BindspaceSpec(
            kind='netns',
            key='tractor-wg0',
        )
        async with attach_netns(spec) as handle:
            namespace_fd: int|None = handle.namespace_fd
            assert namespace_fd is not None
            assert handle.identity.key == 'tractor-wg0'
            assert handle.identity.inode == netns_path.stat().st_ino
            assert handle.ownership == 'borrowed'
            return namespace_fd

    namespace_fd: int = trio.run(main)
    assert netns_path.exists()
    with pytest.raises(OSError):
        os.fstat(namespace_fd)


@pytest.mark.skipif(
    sys.platform != 'linux',
    reason='Linux netns API',
)
def test_attach_named_netns_never_creates(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    '''
    Borrow-only lookup must fail without creating a missing resource.

    Point the run directory at an empty location, request one named
    netns, and prove the open error propagates while no path appears.

    '''
    monkeypatch.setattr(
        _bindspace,
        '_NETNS_RUN_DIR',
        tmp_path,
    )
    missing_path: Path = tmp_path / 'missing'

    async def main() -> None:
        '''
        Attempt to borrow one absent named netns.

        '''
        spec: BindspaceSpec = BindspaceSpec(
            kind='netns',
            key='missing',
        )
        async with attach_netns(spec):
            raise AssertionError('Missing netns unexpectedly opened')

    with pytest.raises(FileNotFoundError):
        trio.run(main)
    assert not missing_path.exists()