"""Retiring an unadopted spare must not log an invalid-adoption traceback.

The daemon discards unused spares by closing their private adoption channel.
E2E teardown does not expose the launcher's inherited stderr or exit status.
"""

import json
import os
from pathlib import Path
import socket
import subprocess
import sys
import tempfile
import unittest

PYTHON = Path(__file__).resolve().parents[2] / "priv/python"


class SpareProtocolTest(unittest.TestCase):
    def test_unused_spare_exits_cleanly_on_eof(self):
        result = subprocess.run(
            [sys.executable, str(PYTHON / "albedo_spare.py")],
            input=b"",
            capture_output=True,
            timeout=30,
        )
        self.assertIn("ready", json.loads(result.stdout))
        self.assertEqual(result.returncode, 0, result.stderr.decode())
        self.assertEqual(result.stderr, b"")

    def test_partial_adoption_is_still_an_error(self):
        result = subprocess.run(
            [sys.executable, str(PYTHON / "albedo_spare.py")],
            input=b"{",
            capture_output=True,
            timeout=30,
        )
        self.assertNotEqual(result.returncode, 0)
        self.assertIn(b"invalid spare adoption frame", result.stderr)

    @unittest.skipUnless(hasattr(os, "fork"), "requires fork")
    def test_unused_forked_spare_exits_cleanly_on_eof(self):
        with tempfile.TemporaryDirectory(prefix="spare-") as directory:
            path = str(Path(directory) / "zygote.sock")
            with subprocess.Popen(
                [sys.executable, str(PYTHON / "albedo_zygote.py"), path],
                stdin=subprocess.PIPE,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
            ) as zygote:
                try:
                    assert zygote.stdout is not None
                    self.assertIn("ready", json.loads(zygote.stdout.readline()))
                    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as channel:
                        channel.settimeout(30)
                        channel.connect(path)
                        with channel.makefile("rb") as reader:
                            self.assertIn("ready", json.loads(reader.readline()))
                            channel.shutdown(socket.SHUT_WR)
                            self.assertEqual(reader.read(), b"")
                    _, errors = zygote.communicate(timeout=30)
                    self.assertEqual(zygote.returncode, 0, errors.decode())
                    self.assertEqual(errors, b"")
                finally:
                    zygote.kill()