homelab/userbot/panel/backend/tests/test_kubernetes_service.py

from datetime import UTC, datetime
from types import SimpleNamespace
from unittest.mock import Mock

import pytest
from app.config import Settings
from app.errors import PanelError
from app.kubernetes_service import KubernetesService
from app.models import AccountBase
from kubernetes.client.exceptions import ApiException


def service() -> KubernetesService:
    core = Mock()
    apps = Mock()
    custom = Mock()
    custom.get_namespaced_custom_object.side_effect = ApiException(status=404)
    return KubernetesService(Settings(), core=core, apps=apps, custom=custom)


def account() -> AccountBase:
    return AccountBase(
        instance_id='test-account',
        display_name='Test Account',
        api_id=12345,
        api_hash='0123456789abcdef0123456789abcdef',
    )


def test_renders_managed_resources_without_leaking_credentials() -> None:
    kube = service()
    names = kube._resource_names('test-account')
    secret = kube._secret(account(), 'SESSION', names)
    pvc = kube._pvc(account(), names)
    deployment = kube._deployment(account(), names)

    assert secret.string_data == {
        'API_ID': '12345',
        'API_HASH': '0123456789abcdef0123456789abcdef',
        'STRINGSESSION': 'SESSION',
    }
    assert pvc.spec.storage_class_name == 'local-path-retain'
    assert pvc.spec.resources.requests['storage'] == '1Gi'
    assert deployment.spec.strategy.type == 'Recreate'
    assert deployment.spec.replicas == 1
    assert deployment.spec.template.spec.automount_service_account_token is False
    assert deployment.spec.template.spec.service_account_name == 'userbot-runtime'
    assert deployment.spec.template.spec.volumes[1].host_path.path.endswith('/Downloads')
    assert deployment.spec.template.spec.containers[0].resources.requests == {
        'cpu': '80m',
        'memory': '512Mi',
    }


def test_collision_is_reported_before_auth() -> None:
    kube = service()
    kube.apps.read_namespaced_deployment.return_value = object()

    with pytest.raises(PanelError) as error:
        kube.assert_available('test-account')

    assert error.value.status_code == 409
    assert 'Deployment' in error.value.detail


def test_partial_provision_rolls_back_only_created_resources() -> None:
    kube = service()
    kube.ensure_prerequisites = Mock()
    kube.assert_available = Mock()
    kube.core.create_namespaced_persistent_volume_claim.side_effect = ApiException(status=500)

    with pytest.raises(PanelError):
        kube.provision(account(), 'SESSION')

    kube.core.delete_namespaced_secret.assert_called_once_with(
        'userbot-test-account-credentials',
        'userbot',
    )
    kube.core.delete_namespaced_persistent_volume_claim.assert_not_called()
    kube.apps.delete_namespaced_deployment.assert_not_called()


def test_provision_conflict_reports_existing_instance() -> None:
    kube = service()
    kube.assert_available = Mock()
    kube.apps.create_namespaced_deployment.side_effect = ApiException(status=409)

    with pytest.raises(PanelError) as error:
        kube.provision(account(), 'SESSION')

    assert error.value.status_code == 409
    assert 'already exists' in error.value.detail
    # Partial resources created before the 409 must be rolled back.
    kube.core.delete_namespaced_secret.assert_called_once_with(
        'userbot-test-account-credentials',
        'userbot',
    )
    kube.core.delete_namespaced_persistent_volume_claim.assert_called_once_with(
        'userbot-test-account-data',
        'userbot',
    )


def test_delete_retains_pvc_unless_explicitly_requested() -> None:
    kube = service()
    deployment = SimpleNamespace(
        metadata=SimpleNamespace(
            name='userbot-test-account',
            annotations={
                'userbot.forust.xyz/credentials-secret': 'credentials',
                'userbot.forust.xyz/pvc': 'data',
            },
        )
    )
    kube._find_deployment = Mock(return_value=('userbot', deployment))

    kube.delete('test-account', delete_data=False)

    kube.apps.delete_namespaced_deployment.assert_called_once()
    kube.core.delete_namespaced_secret.assert_called_once_with('credentials', 'userbot')
    kube.core.delete_namespaced_persistent_volume_claim.assert_not_called()

    kube.delete('test-account', delete_data=True)
    kube.core.delete_namespaced_persistent_volume_claim.assert_called_once_with(
        'data',
        'userbot',
    )


def test_legacy_delete_is_blocked() -> None:
    kube = service()
    deployment = SimpleNamespace(
        metadata=SimpleNamespace(
            name='forust-userbot-deployment',
            annotations={'userbot.forust.xyz/legacy': 'true'},
        )
    )
    kube._find_deployment = Mock(return_value=('default', deployment))

    with pytest.raises(PanelError) as error:
        kube.delete('forust', delete_data=False)

    assert error.value.status_code == 409


def pod(*, ready: bool, phase: str = 'Running', reason: str | None = None):
    waiting = SimpleNamespace(reason=reason) if reason else None
    state = SimpleNamespace(waiting=waiting, terminated=None)
    status = SimpleNamespace(
        container_statuses=[
            SimpleNamespace(ready=ready, restart_count=2, state=state),
        ],
        phase=phase,
        start_time=datetime.now(UTC),
    )
    return SimpleNamespace(
        metadata=SimpleNamespace(name='pod-1', creation_timestamp=datetime.now(UTC)),
        status=status,
    )


def deployment(replicas: int = 1):
    resources = SimpleNamespace(limits={'cpu': '300m', 'memory': '1536Mi'})
    container = SimpleNamespace(image='userbot:latest', resources=resources)
    template_spec = SimpleNamespace(containers=[container], volumes=[])
    return SimpleNamespace(
        metadata=SimpleNamespace(
            name='userbot-test-account',
            labels={'app.kubernetes.io/instance': 'test-account'},
            annotations={},
            creation_timestamp=datetime.now(UTC),
        ),
        spec=SimpleNamespace(
            replicas=replicas,
            selector=SimpleNamespace(match_labels={'app': 'test'}),
            template=SimpleNamespace(spec=template_spec),
        ),
        status=SimpleNamespace(available_replicas=1 if replicas else 0),
    )


@pytest.mark.parametrize(
    ('replicas', 'pod_value', 'expected'),
    [
        (0, None, 'stopped'),
        (1, pod(ready=True), 'running'),
        (1, pod(ready=False, phase='Pending'), 'pending'),
        (1, pod(ready=False, reason='CrashLoopBackOff'), 'error'),
        (1, pod(ready=False, phase='Failed'), 'error'),
    ],
)
def test_status_classification(replicas, pod_value, expected) -> None:
    kube = service()
    kube._pods_for_deployment = Mock(return_value=[pod_value] if pod_value else [])
    kube._pvc_storage = Mock(return_value=None)
    kube._pod_metrics = Mock(return_value=(None, None))

    result = kube._summarize('userbot', deployment(replicas))

    assert result.status == expected