truf-server/tests/test_host_agent_state.py

import hashlib
import json
import os
from pathlib import Path
import sys
import tempfile
import unittest
import uuid
from unittest import mock


ROOT = Path(__file__).resolve().parents[1]
APP_DIR = ROOT / 'app'
sys.path.insert(0, str(APP_DIR))

import host_agent_state
from host_agent_protocol import HostAgentAction, HostAgentRequest


class HostAgentStateTests(unittest.TestCase):
    def request(self, *, operation_id=None):
        return HostAgentRequest(
            operation_id=operation_id or str(uuid.uuid4()),
            action=HostAgentAction.APPLY_BOTH,
            active_config_sha256=hashlib.sha256(b'old config').hexdigest(),
            active_secrets_sha256=hashlib.sha256(b'old secrets').hexdigest(),
            candidate_config_sha256=hashlib.sha256(b'new config').hexdigest(),
            candidate_secrets_sha256=hashlib.sha256(b'new secrets').hexdigest(),
        )

    def store(self):
        temporary = tempfile.TemporaryDirectory()
        root = Path(temporary.name)
        state = root / 'state'
        operations = state / 'operations'
        results = state / 'results'
        state.mkdir(mode=0o700)
        operations.mkdir(mode=0o700)
        results.mkdir(mode=0o750)
        for path, mode in ((state, 0o700), (operations, 0o700), (results, 0o750)):
            os.chmod(path, mode)
        details = os.stat(root)
        patch = mock.patch.multiple(
            host_agent_state,
            HOST_ROOT_UID=getattr(details, 'st_uid', 0),
            HOST_ROOT_GID=getattr(details, 'st_gid', 0),
            HOST_RUNTIME_GID=getattr(details, 'st_gid', 0),
            HOST_STATE_ROOT=state,
            HOST_OPERATION_DIRECTORY=operations,
            HOST_RESULT_DIRECTORY=results,
            HOST_FAILED_HOLD_PATH=state / 'failed-hold.json',
        )
        return temporary, root, patch

    def test_persists_canonical_phase_and_exact_terminal_result(self):
        temporary, root, patch = self.store()
        with temporary, patch:
            request = self.request()
            state = host_agent_state.HostOperationState(request)
            prepared = state.initialize()
            self.assertEqual(prepared['phase'], 'prepared')
            state.advance('prepared', 'forward_started', 'original')
            identity = {
                'active_config_sha256': request.candidate_config_sha256,
                'active_secrets_sha256': request.candidate_secrets_sha256,
            }
            state.advance('forward_started', 'succeeded', 'candidate')
            expected = state.publish_result(
                'succeeded', safe_category=None, safe_detail=None,
                resulting_identity=identity,
            )

            replay = host_agent_state.HostOperationState(request)
            self.assertEqual(replay.initialize()['phase'], 'succeeded')
            self.assertEqual(replay.terminal_result(), expected)
            payload = replay.result_path.read_bytes()
            self.assertEqual(
                payload,
                json.dumps(
                    expected, sort_keys=True, separators=(',', ':'),
                ).encode('ascii'),
            )

    def test_result_replay_is_exact_and_conflicting_bytes_are_rejected(self):
        temporary, _root, patch = self.store()
        with temporary, patch:
            request = self.request()
            state = host_agent_state.HostOperationState(request)
            state.initialize()
            identity = {
                'active_config_sha256': request.active_config_sha256,
                'active_secrets_sha256': request.active_secrets_sha256,
            }
            state.publish_result(
                'rolled_back', safe_category='health_check_failed',
                safe_detail='health_check_failed', resulting_identity=identity,
            )
            state.publish_result(
                'rolled_back', safe_category='health_check_failed',
                safe_detail='health_check_failed', resulting_identity=identity,
            )
            with self.assertRaises(host_agent_state.HostStateError) as raised:
                state.publish_result(
                    'rolled_back', safe_category='restart_failed',
                    safe_detail='restart_failed', resulting_identity=identity,
                )
            self.assertEqual(raised.exception.category, 'conflict')

    def test_failed_hold_is_durable_and_fences_new_operations(self):
        temporary, _root, patch = self.store()
        with temporary, patch:
            request = self.request()
            state = host_agent_state.HostOperationState(request)
            state.initialize('partial')
            state.advance(
                'prepared', 'rollback_started', 'partial',
                forward_category='apply_failed',
            )
            state.publish_failed_hold(
                forward_category='apply_failed', publication_state='partial',
                containment_confirmed=True,
            )
            state.advance(
                'rollback_started', 'failed_hold', 'partial',
                forward_category='apply_failed', safe_detail='rollback_failed',
                containment_confirmed=True,
            )
            self.assertEqual(
                host_agent_state.failed_hold_operation(), request.operation_id,
            )
            self.assertEqual(
                host_agent_state.HostOperationState(request).initialize()['phase'],
                'failed_hold',
            )
            with self.assertRaises(host_agent_state.HostStateError) as raised:
                host_agent_state.HostOperationState(self.request()).initialize()
            self.assertEqual(raised.exception.category, 'failed_hold')

    def test_post_replace_phase_write_is_reported_as_uncertain(self):
        temporary, _root, patch = self.store()
        with temporary, patch:
            request = self.request()
            state = host_agent_state.HostOperationState(request)
            state.initialize()
            with mock.patch.object(
                host_agent_state, 'fsync_directory',
                side_effect=OSError('sync detail'),
            ):
                with self.assertRaises(host_agent_state.HostStateError) as raised:
                    state.advance('prepared', 'forward_started', 'original')

            self.assertEqual(raised.exception.category, 'uncertain')
            self.assertEqual(state._read_phase()['phase'], 'forward_started')

    def test_post_replace_cancellation_is_retained_as_uncertain(self):
        temporary, _root, patch = self.store()
        with temporary, patch:
            request = self.request()
            state = host_agent_state.HostOperationState(request)
            state.initialize()
            cancellation = KeyboardInterrupt()
            with mock.patch.object(
                host_agent_state, 'fsync_directory',
                side_effect=cancellation,
            ):
                with self.assertRaises(host_agent_state.HostStateError) as raised:
                    state.advance('prepared', 'forward_started', 'original')

            self.assertEqual(raised.exception.category, 'uncertain')
            self.assertIs(raised.exception.cancellation, cancellation)
            self.assertEqual(state._read_phase()['phase'], 'forward_started')

    def test_phase_request_identity_and_transition_are_closed(self):
        temporary, _root, patch = self.store()
        with temporary, patch:
            request = self.request()
            state = host_agent_state.HostOperationState(request)
            state.initialize()
            with self.assertRaises(host_agent_state.HostStateError):
                state.advance('prepared', 'succeeded', 'candidate')
            altered = HostAgentRequest(
                operation_id=request.operation_id,
                action=request.action,
                active_config_sha256=request.active_config_sha256,
                active_secrets_sha256=request.active_secrets_sha256,
                candidate_config_sha256='f' * 64,
                candidate_secrets_sha256=request.candidate_secrets_sha256,
            )
            with self.assertRaises(host_agent_state.HostStateError) as raised:
                host_agent_state.HostOperationState(altered).initialize()
            self.assertEqual(raised.exception.category, 'evidence')


if __name__ == '__main__':
    unittest.main()