MeshChatX/tests/backend/test_api_json_contracts.py

# SPDX-License-Identifier: 0BSD

"""JSON Schema contract tests for core HTTP API responses."""

from __future__ import annotations

import json
import shutil
import tempfile
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
import RNS

from meshchatx.meshchat import ReticulumMeshChat
from tests.backend.api_json_contract_schemas import (
    API_V1_APP_INFO_ENVELOPE_SCHEMA,
    API_V1_STATUS_SCHEMA,
    AUTH_STATUS_SCHEMA,
    assert_matches_schema,
)


@pytest.fixture(autouse=True)
def _stub_threads_for_http_contract_tests():
    with patch("threading.Thread"):
        yield


@pytest.fixture
def temp_dir():
    dir_path = tempfile.mkdtemp()
    yield dir_path
    shutil.rmtree(dir_path)


@pytest.fixture
def mock_rns_minimal():
    with (
        patch("RNS.Reticulum") as mock_rns,
        patch("RNS.Transport"),
        patch("LXMF.LXMRouter"),
        patch("meshchatx.meshchat.get_file_path", return_value="/tmp/mock_path"),
    ):
        mock_rns_instance = mock_rns.return_value
        mock_rns_instance.configpath = "/tmp/mock_config"
        mock_rns_instance.is_connected_to_shared_instance = False
        mock_rns_instance.transport_enabled.return_value = True

        mock_id = MagicMock(spec=RNS.Identity)
        mock_id.hash = b"test_hash_32_bytes_long_01234567"
        mock_id.hexhash = mock_id.hash.hex()
        mock_id.get_private_key.return_value = b"test_private_key"
        yield mock_id


def _route_handler(app: ReticulumMeshChat, path: str, method: str):
    for route in app.get_routes():
        if route.path == path and route.method == method:
            return route.handler
    return None


@pytest.mark.asyncio
async def test_api_v1_status_json_contract(mock_rns_minimal, temp_dir):
    with patch("meshchatx.meshchat.generate_ssl_certificate"):
        app_instance = ReticulumMeshChat(
            identity=mock_rns_minimal,
            storage_dir=temp_dir,
            reticulum_config_dir=temp_dir,
        )
        handler = _route_handler(app_instance, "/api/v1/status", "GET")
        assert handler is not None
        request = MagicMock()
        response = await handler(request)
        data = json.loads(response.body)
        assert_matches_schema(data, API_V1_STATUS_SCHEMA)


@pytest.mark.asyncio
async def test_api_v1_app_info_json_contract(mock_rns_minimal, temp_dir):
    with (
        patch("meshchatx.meshchat.generate_ssl_certificate"),
        patch("psutil.Process") as mock_process,
        patch("psutil.net_io_counters") as mock_net_io,
        patch("importlib.metadata.version", return_value="1.2.3"),
        patch("meshchatx.meshchat.LXST") as mock_lxst,
    ):
        mock_lxst.__version__ = "1.2.3"

        mock_proc_instance = mock_process.return_value
        mock_proc_instance.memory_info.return_value.rss = 1024
        mock_proc_instance.memory_info.return_value.vms = 2048
        mock_proc_instance.net_connections.return_value = []

        mock_net_instance = mock_net_io.return_value
        mock_net_instance.bytes_sent = 0
        mock_net_instance.bytes_recv = 0
        mock_net_instance.packets_sent = 0
        mock_net_instance.packets_recv = 0

        app_instance = ReticulumMeshChat(
            identity=mock_rns_minimal,
            storage_dir=temp_dir,
            reticulum_config_dir=temp_dir,
        )

        handler = _route_handler(app_instance, "/api/v1/app/info", "GET")
        assert handler is not None
        request = MagicMock()
        response = await handler(request)
        data = json.loads(response.body)
        assert_matches_schema(data, API_V1_APP_INFO_ENVELOPE_SCHEMA)


@pytest.mark.asyncio
async def test_api_v1_auth_status_json_contract(mock_rns_minimal, temp_dir):
    with (
        patch("meshchatx.meshchat.generate_ssl_certificate"),
        patch("meshchatx.meshchat.get_session", new_callable=AsyncMock) as mock_session,
    ):
        mock_session.return_value = {
            "authenticated": False,
            "identity_hash": None,
        }

        app_instance = ReticulumMeshChat(
            identity=mock_rns_minimal,
            storage_dir=temp_dir,
            reticulum_config_dir=temp_dir,
        )

        handler = _route_handler(app_instance, "/api/v1/auth/status", "GET")
        assert handler is not None
        request = MagicMock()
        response = await handler(request)
        data = json.loads(response.body)
        assert_matches_schema(data, AUTH_STATUS_SCHEMA)


def test_auth_status_schema_accepts_error_envelope():
    sample = {
        "auth_enabled": True,
        "password_set": True,
        "authenticated": False,
        "error": "decryption failed",
    }
    assert_matches_schema(sample, AUTH_STATUS_SCHEMA)


def test_auth_status_schema_accepts_starting_envelope():
    sample = {
        "auth_enabled": False,
        "password_set": False,
        "authenticated": False,
        "network_ready": False,
        "status": "starting",
        "stage": "rns",
    }
    assert_matches_schema(sample, AUTH_STATUS_SCHEMA)


def test_status_schema_accepts_starting_and_failed_envelopes():
    starting = {
        "status": "starting",
        "stage": "identity",
        "network_ready": False,
        "listen_host": "127.0.0.1",
        "listen_port": 9337,
        "https_enabled": True,
        "is_loopback_bind": True,
        "plugins_enabled": True,
        "landlock_kernel_supported": False,
        "landlock_requested": False,
        "landlock_auto_enabled": False,
        "landlock_disabled_by_env": False,
        "landlock_active": False,
        "appcontainer_supported": False,
        "appcontainer_requested": False,
        "appcontainer_auto_enabled": False,
        "appcontainer_disabled_by_env": False,
        "appcontainer_active": False,
        "fs_sandbox_active": False,
        "seccomp_kernel_supported": False,
        "seccomp_requested": False,
        "seccomp_auto_enabled": False,
        "seccomp_disabled_by_env": False,
        "seccomp_active": False,
    }
    assert_matches_schema(starting, API_V1_STATUS_SCHEMA)
    failed = {
        **starting,
        "status": "failed",
        "stage": "failed",
        "error": "RNS init failed",
    }
    assert_matches_schema(failed, API_V1_STATUS_SCHEMA)