#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-3.0-or-later
"""Failure injection for Phase-1.0AT descriptor ownership."""

from __future__ import annotations

import argparse
from pathlib import Path
import sys
import unittest


PARSER = argparse.ArgumentParser()
PARSER.add_argument("--root", type=Path, required=True)
ROOT = PARSER.parse_args().root
sys.path.insert(0, str(ROOT / "tools"))

from phase10aq_worker_result_record import *  # noqa: E402,F403
from phase10at_fd_deadline_model import *  # noqa: E402,F403


PRECOMMIT = WorkerPrecommit(b"a" * 16, b"b" * 16, 101, 202, 7, 4096)
RAW = encode_record(PRECOMMIT, STATUS_SUCCESS, 4096)


def success_events() -> list[FakeFdEvent]:
    return [FakeFdEvent(CREATE_PIPE, OK),
            FakeFdEvent(SET_PARENT_READ_NONBLOCK, OK),
            FakeFdEvent(SPAWN_RFFDG, OK),
            FakeFdEvent(PARENT_CLOSE_WRITE, OK),
            FakeFdEvent(CHILD_CLOSE_READ, OK),
            FakeFdEvent(CHILD_CLOSE_WRITE, OK),
            FakeFdEvent(REAP_CHILD, OK),
            FakeFdEvent(PARENT_CLOSE_READ, OK)]


class FdDeadlineTests(unittest.TestCase):
    def test_success_with_partial_reads_eintr_and_would_block(self) -> None:
        reads = (FakeRead(EINTR), FakeRead(WOULD_BLOCK),
                 FakeRead(DATA, RAW[:40]), FakeRead(EINTR),
                 FakeRead(DATA, RAW[40:]))
        result = run_fd_transaction(
            PRECOMMIT, FakeFdFacade(tuple(success_events())), reads)
        self.assertTrue(result.success and result.rffdg_used)
        self.assertEqual(result.eintr_count, 2)
        self.assertEqual(result.parent_fds_open + result.child_fds_open, 0)
        self.assertFalse(result.worker_alive or result.live_fd_present)

    def test_every_setup_failure_closes_only_acquired_resources(self) -> None:
        expected_cleanup = {
            0: [],
            1: [PARENT_CLOSE_WRITE, PARENT_CLOSE_READ],
            2: [PARENT_CLOSE_WRITE, PARENT_CLOSE_READ],
            3: [TERMINATE_CHILD, CHILD_CLOSE_READ, CHILD_CLOSE_WRITE,
                REAP_CHILD, PARENT_CLOSE_WRITE, PARENT_CLOSE_READ],
            4: [TERMINATE_CHILD, CHILD_CLOSE_READ, CHILD_CLOSE_WRITE,
                REAP_CHILD, PARENT_CLOSE_READ],
        }
        baseline = success_events()[:5]
        for index, event in enumerate(baseline):
            prefix = baseline[:index] + [FakeFdEvent(event.operation, ERROR)]
            prefix += [FakeFdEvent(operation, OK)
                       for operation in expected_cleanup[index]]
            with self.subTest(index=index):
                result = run_fd_transaction(
                    PRECOMMIT, FakeFdFacade(tuple(prefix)),
                    (FakeRead(DATA, RAW),))
                self.assertFalse(result.success)
                self.assertEqual(result.parent_fds_open + result.child_fds_open, 0)

    def test_deadline_eof_overflow_and_bad_record_contain_worker(self) -> None:
        scripts = (
            (FakeRead(DATA, RAW[:10], ticks=252), FakeRead(DATA, RAW[10:])),
            (FakeRead(DATA, RAW[:10]), FakeRead(EOF)),
            (FakeRead(DATA, RAW[:100]), FakeRead(DATA, RAW[100:] + b"x")),
            (FakeRead(DATA, RAW), FakeRead(WOULD_BLOCK)),
            (FakeRead(DATA, bytes(bytearray(RAW[:1]) + RAW[1:])),),
        )
        # Make the final script genuinely invalid.
        damaged = bytearray(RAW)
        damaged[0] ^= 1
        scripts = scripts[:-1] + ((FakeRead(DATA, bytes(damaged)),),)
        cleanup = [FakeFdEvent(TERMINATE_CHILD, OK),
                   FakeFdEvent(CHILD_CLOSE_WRITE, OK),
                   FakeFdEvent(REAP_CHILD, OK),
                   FakeFdEvent(PARENT_CLOSE_READ, OK)]
        for reads in scripts:
            events = success_events()[:5] + cleanup
            with self.subTest(reads=reads):
                result = run_fd_transaction(
                    PRECOMMIT, FakeFdFacade(tuple(events)), reads)
                self.assertFalse(result.success or result.worker_alive)
                self.assertTrue(result.containment_required)

    def test_cleanup_failure_and_unused_operations_are_hard_errors(self) -> None:
        events = [FakeFdEvent(CREATE_PIPE, OK),
                  FakeFdEvent(SET_PARENT_READ_NONBLOCK, ERROR),
                  FakeFdEvent(PARENT_CLOSE_WRITE, ERROR)]
        with self.assertRaises(FdModelError):
            run_fd_transaction(PRECOMMIT, FakeFdFacade(tuple(events)),
                               (FakeRead(DATA, RAW),))
        with self.assertRaises(FdModelError):
            run_fd_transaction(
                PRECOMMIT,
                FakeFdFacade(tuple(success_events() +
                                   [FakeFdEvent(PARENT_CLOSE_READ, OK)])),
                (FakeRead(DATA, RAW),))

    def test_invalid_read_and_boundary_are_rejected(self) -> None:
        with self.assertRaises(FdModelError):
            FakeRead(DATA, b"")
        with self.assertRaises(FdModelError):
            run_fd_transaction(PRECOMMIT, FakeFdFacade(tuple(success_events())), ())


if __name__ == "__main__":
    unittest.main(argv=[__file__])