import multiprocessing
import os
from pathlib import Path
import queue
import tempfile
import time
import unittest
from unittest.mock import patch

from app.channel_child import parent_death_guard, send_latest
from app.channel_process import ProcessSlots, ProcessStreamGroup
from app.stream_manager import StreamManager
from app.stream_config_store import ConfigPersistenceError


def fixture_child(spec, run_id, generation, settings, token, commands, output, stop, ack, parent_pid):
    parent_death_guard(parent_pid)
    output.cancel_join_thread()
    value = dict(run_id=run_id,run_generation=generation,state='running',observed_state='running',
                 desired_state='running',roles={},last_error=None,cleanup_error=None,
                 metadata_path=dict(state='CONNECTED',processed=1),child_pid=os.getpid())
    while not stop.is_set():
        send_latest(output,dict(kind='status',run_id=run_id,run_generation=generation,value=value))
        try:
            cmd = commands.get(timeout=.05)
            send_latest(output,dict(kind='reply',run_id=run_id,run_generation=generation,
                                    id=cmd['id'],value=dict(detail=cmd['enabled'])))
        except queue.Empty:
            pass
    value.update(state='stopped',observed_state='stopped',desired_state='stopped')
    for _ in range(20):
        send_latest(output,dict(kind='terminal',run_id=run_id,run_generation=generation,value=value,failed=False))
        if ack.wait(.05):
            break


def stubborn_child(spec, run_id, generation, settings, token, commands, output, stop, ack, parent_pid):
    parent_death_guard(parent_pid)
    time.sleep(30)


def crash_child(*args):
    os._exit(9)


def orphan_child(parent_pid, output):
    parent_death_guard(parent_pid)
    output.put(os.getpid())
    time.sleep(30)


def orphan_parent(output):
    context = multiprocessing.get_context('spawn')
    child = context.Process(target=orphan_child,args=(os.getpid(),output))
    child.start()
    time.sleep(30)


def wait(predicate, seconds=10):
    deadline = time.monotonic()+seconds
    while time.monotonic()<deadline:
        if predicate():
            return
        time.sleep(.02)
    raise AssertionError('Bounded fixture wait expired.')


class ProcessTest(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp.cleanup)
        self.slots = ProcessSlots(lock_path=self.temp.name+'/parent.lock')
        self.groups = []
        self.addCleanup(self.close_groups)

    def close_groups(self):
        for g in self.groups:
            g.stop()
        for g in self.groups:
            self.assertTrue(g._cleanup_done.wait(10))
        self.assertTrue(self.slots.close())

    def group(self, name, target=fixture_child):
        group=ProcessStreamGroup(dict(stream_id=name,server_id='server',source_id=name,uri='unused',
            tracking={'enabled':False},store=False,active_learning=False,scenarios='off'),
            slots=self.slots,child_target=target)
        self.groups.append(group)
        return group

    def test_sixteen_distinct_children_limit_and_reap(self):
        for i in range(16):
            self.group('fixture-'+str(i)).start()
        self.assertEqual(16,len({g._process.pid for g in self.groups}))
        extra=self.group('extra')
        with self.assertRaises(RuntimeError):
            extra.start()
        wait(lambda:all(g.state=='running' for g in self.groups[:-1]),15)
        for g in self.groups[:-1]:
            pid=g._process.pid; generation=g.run_generation
            self.assertEqual(generation,g.start())
            self.assertEqual(pid,g._process.pid)
        for g in self.groups:
            g.stop()
        wait(lambda:all(g._cleanup_done.is_set() for g in self.groups),15)
        self.assertTrue(all(g._process is None for g in self.groups))
        self.assertTrue(all(g._last_cleanup['completed'] for g in self.groups[:-1]))

    def test_restart_identity_stale_reply_metrics_and_status(self):
        g=self.group('one');g.start();wait(lambda:g.state=='running')
        old=g.lifecycle.snapshot();pid=g._process.pid
        self.assertEqual({'detail':True},g.configure_metrics(g.run_generation,True,1))
        with self.assertRaises(RuntimeError):g.configure_metrics(999,True,1)
        g.restart();g.restart()
        wait(lambda:g.state=='running' and g.run_generation==2)
        self.assertNotEqual(pid,g._process.pid)
        current=g.lifecycle.snapshot()
        self.assertFalse(g._accept(dict(kind='status',run_id=old['run_id'],run_generation=1,value=old),
                                   current['run_id'],2))
        self.assertEqual('running',g.state)

    def test_crash_and_forced_exit_are_not_normal_cleanup(self):
        g=self.group('crash',crash_child);g.start();wait(lambda:g._cleanup_done.is_set())
        self.assertEqual('error',g.state)
        self.assertFalse(g._last_cleanup['completed'])
        self.assertTrue(g._last_cleanup['child_reaped'])
        slow=self.group('slow',stubborn_child);slow.CLEANUP_TIMEOUT_S=.05;slow.start();slow.stop()
        wait(lambda:slow._cleanup_done.is_set(),8)
        self.assertTrue(slow._last_cleanup['forced'])
        self.assertFalse(slow._last_cleanup['completed'])

    def test_settings_cas_noop_and_save_failure_never_start(self):
        with patch.dict(os.environ,CONTEXTRA_STREAM_EXECUTION_MODE='process'):
            manager=StreamManager(self.temp.name+'/base',self.temp.name+'/config.yaml')
        group=manager.add(dict(stream_id='s',server_id='server',source_id='source',uri='unused'))
        revision=manager.settings('s')['configured']['revision']
        manager.patch_settings('s',revision,{'analysisFpsTarget':10})
        self.assertIsNone(group._process)
        with patch.object(manager.config_store,'update',side_effect=ConfigPersistenceError()):
            with self.assertRaises(ConfigPersistenceError):
                manager.patch_settings('s',revision,{'analysisFpsTarget':15})
        self.assertEqual(10,manager.settings('s')['configured']['analysisFpsTarget'])
        manager.patch_settings('s',revision,{'analysisFpsTarget':15})
        with self.assertRaises(ValueError):manager.patch_settings('s',revision,{'analysisFpsTarget':20})
        self.assertIsNone(group._process)
        manager.shutdown_channels()

    def test_legacy_identity_does_not_claim_metadata(self):
        from unittest.mock import Mock
        publisher=Mock()
        g=ProcessStreamGroup(dict(stream_id='legacy',uri='unused',store=False),
            slots=self.slots,metadata_publisher=publisher,child_target=fixture_child)
        self.groups.append(g)
        g.start();wait(lambda:g.state=='running')
        publisher.claim.assert_not_called()

    def test_parent_death_terminates_child(self):
        context=multiprocessing.get_context('spawn');output=context.Queue(maxsize=1)
        parent=context.Process(target=orphan_parent,args=(output,));parent.start()
        try:
            pid=output.get(timeout=10)
            parent.terminate();parent.join(3)
            def dead():
                status=Path('/proc/'+str(pid)+'/status')
                return not status.exists() or '\nState:\tZ' in status.read_text()
            wait(dead,5)
        finally:
            if parent.is_alive():parent.kill();parent.join(2)
            parent.close();output.close()


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