NODEDC_MISSION_CORE/tests/test_device_lifecycle.py

from __future__ import annotations

from datetime import UTC, datetime

import pytest

from k1link.web.device_lifecycle import (
    AcquisitionRecord,
    OperationJournal,
    new_acquisition_id,
)


def test_operation_journal_reuses_an_idempotent_request_without_storing_input() -> None:
    journal = OperationJournal(clock=lambda: datetime(2026, 7, 16, 12, 0, tzinfo=UTC))

    first, created = journal.begin(
        "network.provision",
        idempotency_key="provision-owned-k1-once",
        device_id="device-test",
        device_session_id="device-session-test",
        deadline_seconds=45,
    )
    repeated, repeated_created = journal.begin(
        "network.provision",
        idempotency_key="provision-owned-k1-once",
        device_id="device-test",
        device_session_id="device-session-test",
        deadline_seconds=45,
    )

    assert created is True
    assert repeated_created is False
    assert repeated is first
    assert "password" not in str(journal.snapshot()).lower()


def test_operation_journal_records_ack_progress_and_terminal_result() -> None:
    journal = OperationJournal()
    operation, _ = journal.begin("acquisition.prepare", cancellable=True)

    journal.transition(
        operation.operation_id,
        "running",
        stage_code="compatibility-check",
        message_code="acquisition.compatibility_check",
    )
    journal.transition(
        operation.operation_id,
        "succeeded",
        stage_code="prepared",
        message_code="acquisition.prepared",
        result={"acquisition_id": "acq-test"},
        evidence_refs=("evidence-manifest-test",),
    )

    document = journal.get(operation.operation_id).as_dict()
    assert document["schema_version"] == "missioncore.operation-snapshot/v1alpha2"
    assert document["status"] == "succeeded"
    assert document["sequence"] == 3
    assert document["result"] == {"acquisition_id": "acq-test"}
    assert document["evidence_refs"] == ["evidence-manifest-test"]
    assert document["completed_at"] is not None

    with pytest.raises(ValueError, match="already terminal"):
        journal.transition(
            operation.operation_id,
            "failed",
            stage_code="late-failure",
            message_code="operation.failed",
        )


def test_operation_id_and_idempotency_key_cannot_be_rebound() -> None:
    journal = OperationJournal()
    operation, _ = journal.begin("device.inspect", idempotency_key="same-key")

    with pytest.raises(ValueError, match="another action"):
        journal.begin("network.provision", idempotency_key="same-key")
    with pytest.raises(ValueError, match="another action"):
        journal.begin("sensor.catalog.read", operation_id=operation.operation_id)


def test_idempotency_identity_cannot_be_reused_for_different_request_fingerprint() -> None:
    journal = OperationJournal()
    operation, _ = journal.begin(
        "network.provision",
        idempotency_key="same-request-only",
        request_fingerprint="fingerprint-a",
    )

    repeated, created = journal.begin(
        "network.provision",
        idempotency_key="same-request-only",
        request_fingerprint="fingerprint-a",
    )
    assert created is False
    assert repeated is operation

    with pytest.raises(ValueError, match="different request"):
        journal.begin(
            "network.provision",
            idempotency_key="same-request-only",
            request_fingerprint="fingerprint-b",
        )
    with pytest.raises(ValueError, match="different request"):
        journal.begin(
            "network.provision",
            operation_id=operation.operation_id,
            request_fingerprint="fingerprint-b",
        )

    assert "fingerprint-a" not in str(journal.snapshot())


def test_bounded_journal_never_evicts_an_operation_that_is_still_active() -> None:
    journal = OperationJournal(max_records=1)
    active, _ = journal.begin("acquisition.start", request_fingerprint="start")
    completed, _ = journal.begin("device.inspect", request_fingerprint="inspect")
    journal.transition(
        completed.operation_id,
        "succeeded",
        stage_code="completed",
        message_code="device.inspect.completed",
    )

    assert journal.get(active.operation_id).status == "accepted"
    assert [item["operation_id"] for item in journal.snapshot(limit=10)] == [active.operation_id]


def test_non_cancellable_operation_rejects_cancel_request() -> None:
    journal = OperationJournal()
    operation, _ = journal.begin("network.provision", cancellable=False)

    with pytest.raises(ValueError, match="not cancellable"):
        journal.request_cancel(operation.operation_id)


def test_acquisition_record_keeps_device_session_profile_and_manual_control_separate() -> None:
    acquisition = AcquisitionRecord(
        acquisition_id=new_acquisition_id(),
        device_id="device-test",
        device_session_id="device-session-test",
        compatibility_profile_id="xgrids.lixelkity-k1.fw-3.0.2.direct-lan.v1",
        control_mode="operator-manual",
        requested_streams=("spatial.point-cloud.live", "spatial.pose.live"),
        target_host="192.168.1.20",
        duration_seconds=60,
        evidence_policy="required",
    )

    acquisition.transition("prepared", message_code="acquisition.prepared")
    acquisition.transition(
        "awaiting_external_start",
        message_code="acquisition.operator_start_required",
        operator_instructions=("Дважды нажмите физическую кнопку устройства.",),
    )

    document = acquisition.as_dict()
    assert document["schema_version"] == "missioncore.acquisition-snapshot/v1alpha2"
    assert document["control_mode"] == "operator-manual"
    assert document["state"] == "awaiting_external_start"
    assert document["device_id"] != document["device_session_id"]
    assert document["operator_instructions"]