"""Inert actual-function tests for active notification payloads; no app imports."""
import ast
import copy
import html
import json
import math
import re
import sys
import time
import types
import unittest
from collections import defaultdict
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
SCRIPTS = ROOT / 'AppImage/scripts'
MESSAGES = ROOT / 'AppImage/messages'


def extracted(path, name, cls=None, namespace=None):
    tree = ast.parse((SCRIPTS / path).read_text())
    nodes = tree.body
    if cls:
        nodes = next(n for n in nodes if isinstance(n, ast.ClassDef) and n.name == cls).body
    node = next(n for n in nodes if isinstance(n, ast.FunctionDef) and n.name == name)
    # Never execute module code or decorators; no operational module imports.
    node = copy.deepcopy(node)
    node.decorator_list = []
    scope = namespace if namespace is not None else {}
    exec(compile(ast.fix_missing_locations(ast.Module(body=[node], type_ignores=[])),
                 '<extracted ' + name + '>', 'exec'), scope)
    return scope[name]


def render_fixture(language='en', overlays=None):
    catalogs = {lang: json.loads((MESSAGES / lang / 'common.json').read_text())['runtime']['notifications']
                for lang in ('en', 'it')}
    if overlays:
        for lang, title, body in overlays:
            catalogs[lang]['templates']['temp_high'].update(title=title, body=body)
    def lookup(lang, key):
        value = catalogs.get(lang, {})
        for segment in key.split('.'):
            value = value.get(segment) if isinstance(value, dict) else None
        return value if isinstance(value, str) and value else None
    def message(key, lang, **kw):
        value = lookup(lang, key) or lookup('en', key) or ''
        return value.format(**kw) if value else ''
    src = ast.parse((SCRIPTS / 'notification_templates.py').read_text())
    definition = next(n for n in src.body if isinstance(n, ast.Assign)
                      and any(isinstance(t, ast.Name) and t.id == 'TEMPLATES' for t in n.targets))
    templates = ast.literal_eval(definition.value)
    cpu_pattern = next(n for n in src.body if isinstance(n, ast.Assign)
                       and any(isinstance(t, ast.Name) and t.id == '_CPU_SUSTAINED_REASON'
                               for t in n.targets))
    assert isinstance(cpu_pattern.value, ast.Call)
    assert isinstance(cpu_pattern.value.func, ast.Attribute) and cpu_pattern.value.func.attr == 'compile'
    assert len(cpu_pattern.value.args) == 1
    scope = {'TEMPLATES': templates, '_load_runtime_catalog': lambda lang: catalogs.get(lang, {}),
             '_catalog_value': lambda cat, key: lookup(language if cat is catalogs.get(language) else 'en', key),
             'runtime_message': message, '_get_hostname': lambda: 'node.example',
             'time': time, 'html': html, 're': re, 'Dict': dict, 'Any': object, 'Tuple': tuple,
             '_CPU_SUSTAINED_REASON': re.compile(ast.literal_eval(cpu_pattern.value.args[0]))}
    extracted('notification_templates.py', '_format_health_degraded', namespace=scope)
    render = extracted('notification_templates.py', 'render_template', namespace=scope)
    return lambda kind, data: render(kind, data, language=language)


def temperature_case(history, reading=89., active=False):
    class Persistence:
        emitted = []
        resolved = []
        def record_error(self, **kw): self.emitted.append(kw)
        def is_error_active(self, *a, **kw): return active
        def resolve_error(self, *a, **kw): self.resolved.append((a, kw))
    db = Persistence()
    env = {'time': types.SimpleNamespace(time=lambda: 100000),
           'subprocess': types.SimpleNamespace(run=lambda *a, **kw: types.SimpleNamespace(
               returncode=0, stdout=f'temp1_input: {reading:.3f}\ntemp2_input: 45.000\n')),
           'health_persistence': db, 'Optional': __import__('typing').Optional,
           'Dict': dict, 'Any': object}
    check = extracted('health_monitor.py', '_check_cpu_temperature', 'HealthMonitor', env)
    monitor = types.SimpleNamespace(last_check_times={}, cached_results={}, state_history=defaultdict(list))
    monitor.state_history['cpu_temp_history'] = list(history)
    result = check(monitor)
    return result, db


def produced_temperature():
    result, db = temperature_case([
        {'value': 89., 'time': 99900 + n * 5} for n in range(17)])
    assert result['status'] == 'WARNING' and len(db.emitted) == 1
    return db.emitted[0]


def polled_event(entry):
    raw_row = {'error_key': entry['error_key'], 'category': entry['category'],
               'severity': entry['severity'], 'reason': entry['reason'],
               'details': json.dumps(entry['details']), 'last_seen': '',
               'first_seen': '', 'acknowledged': 0}
    class Cursor:
        def execute(self, *a): pass
        def fetchall(self): return [raw_row]
    class Connection:
        def __enter__(self): return self
        def __exit__(self, *a): return False
        def cursor(self): return Cursor()
    typing = __import__('typing')
    get_active = extracted('health_persistence.py', 'get_active_errors', 'HealthPersistence',
        {'json': json, 'Optional': typing.Optional, 'List': typing.List,
         'Dict': typing.Dict, 'Any': typing.Any})
    persisted = get_active(types.SimpleNamespace(_db_connection=lambda **kw: Connection()))
    assert len(persisted) == 1 and persisted[0]['details'] == entry['details']
    class Persistence:
        def get_active_errors(self): return persisted
    class Event:
        def __init__(self, event_type, severity, data, **kw):
            self.event_type, self.severity, self.data = event_type, severity, data
    q = types.SimpleNamespace(items=[], put=lambda event: q.items.append(event))
    tree = ast.parse((SCRIPTS / 'notification_events.py').read_text())
    cls = next(n for n in tree.body if isinstance(n, ast.ClassDef) and n.name == 'PollingCollector')
    maps = {target.id: ast.literal_eval(n.value) for n in cls.body if isinstance(n, ast.Assign)
            for target in n.targets if isinstance(target, ast.Name) and target.id in
            ('_CATEGORY_TO_EVENT_TYPE', '_ENTITY_MAP', '_CATEGORY_COOLDOWNS')}
    collector = types.SimpleNamespace(_CATEGORY_TO_EVENT_TYPE=maps['_CATEGORY_TO_EVENT_TYPE'],
        _ENTITY_MAP=maps['_ENTITY_MAP'], _CATEGORY_COOLDOWNS=maps['_CATEGORY_COOLDOWNS'],
        DIGEST_INTERVAL=86400, SAME_ERROR_COOLDOWN=86400, _queue=q, _hostname='node.example',
        _first_poll_done=True, _known_errors={}, _last_notified={}, _notified_severity={},
        _is_usb_cache={}, _get_cooldown_from_db=lambda _: None,
        _persist_last_notified=lambda *a: None, _save_known_errors_meta=lambda: None)
    old = sys.modules.get('health_persistence')
    sys.modules['health_persistence'] = types.SimpleNamespace(health_persistence=Persistence())
    try:
        poll = extracted('notification_events.py', '_check_persistent_health', 'PollingCollector',
            {'time': types.SimpleNamespace(time=lambda: 100000), 'json': json,
             'startup_grace': types.SimpleNamespace(should_suppress_category=lambda _: False),
             'NotificationEvent': Event, 'Dict': dict})
        poll(collector)
    finally:
        if old is None: sys.modules.pop('health_persistence', None)
        else: sys.modules['health_persistence'] = old
    assert len(q.items) == 1
    return q.items[0]


class ActivePayloadTests(unittest.TestCase):
    def test_temperature_sampling_boundaries_and_recovery_policy(self):
        # These characterize the existing threshold/guard, not new policy.
        highs = [{'value': 89., 'time': 99900 + n * 5} for n in range(17)]
        below, db = temperature_case(highs[:-1])  # current sample makes 17
        self.assertEqual(below['status'], 'OK')
        self.assertEqual((db.emitted, db.resolved), ([], []))
        warning, db = temperature_case(highs)  # current sample makes 18
        self.assertEqual(warning['status'], 'WARNING')
        self.assertEqual(len(db.emitted), 1)
        self.assertEqual(db.emitted[0]['details']['duration'], 100)
        self.assertEqual((db.emitted[0]['category'], db.emitted[0]['severity'],
                          db.emitted[0]['details']['dismissable']), ('temperature', 'WARNING', True))
        low = [{'value': 79., 'time': t} for t in (99980, 99990)]
        pending, db = temperature_case(low[:1], reading=79., active=True)
        self.assertEqual(pending['status'], 'WARNING')
        self.assertEqual((db.emitted, db.resolved), ([], []))
        recovered, db = temperature_case(low, reading=79., active=True)
        self.assertEqual(recovered['status'], 'OK')
        self.assertEqual(db.emitted, [])
        self.assertEqual(db.resolved[0][0][0], 'cpu_temperature')

    def test_temperature_record_queue_and_english_render_use_measured_sensor(self):
        record = produced_temperature()
        self.assertIn('sensor', record['reason'].lower())
        self.assertEqual(record['details']['temperature'], 89.)
        self.assertTrue(record['details']['dismissable'])
        event = polled_event(record)
        self.assertEqual((event.event_type, event.severity), ('temp_high', 'WARNING'))
        self.assertEqual(event.data['value'], 89.)
        self.assertEqual(event.data['threshold'], 80)
        self.assertEqual(event.data['duration'], record['details']['duration'])
        self.assertIn('High samples span', event.data['details'])
        rendered = render_fixture()('temp_high', event.data)
        self.assertIn('sensor temperature', rendered['title'].lower())
        self.assertIn('89', rendered['title'])
        self.assertIn('80', rendered['body'])
        self.assertNotIn('CPU', rendered['title'] + rendered['body'])

    def test_missing_temperature_measurement_never_renders_blank_celsius_claim(self):
        event = polled_event({'error_key': 'cpu_temperature', 'category': 'temperature',
                              'severity': 'WARNING', 'reason': 'legacy raw reason',
                              'details': {'duration': 90}})
        for lang in ('en', 'it'):
            out = render_fixture(lang)('temp_high', event.data)
            self.assertNotIn('°C', out['title'] + out['body'])
            self.assertNotIn('CPU', out['title'] + out['body'])
            self.assertIn('temperature', (out['title'] + out['body']).lower())

    def test_nonfinite_temperature_payload_is_not_presented_as_a_reading(self):
        render = render_fixture()
        for value, threshold in ((math.nan, 80), (math.inf, 80), (89, math.inf),
                                 (True, 80), (10**400, 80), (89, 10**400)):
            with self.subTest(value=value, threshold=threshold):
                out = render('temp_high', {'value': value, 'threshold': threshold})
                self.assertNotIn('°C', out['title'] + out['body'])

    def test_numeric_strings_from_manual_payloads_retain_finite_readings(self):
        out = render_fixture()('temp_high', {'value': '89.0', 'threshold': '80'})
        self.assertIn('89.0°C', out['title'])
        self.assertIn('80°C', out['body'])
        for bad in ('', 'NaN', 'Infinity', '1e999', 'eighty', '  ', object()):
            with self.subTest(value=repr(bad)):
                out = render_fixture()('temp_high', {'value': bad, 'threshold': '80'})
                self.assertNotIn('°C', out['title'] + out['body'])

    def test_incomplete_measurement_preserves_recorded_context_once(self):
        render = render_fixture()
        for data, fragments in (
            ({'reason': 'CPU temperature 89°C >80°C sustained for 2m'},
             ('Recorded reason: CPU temperature 89°C >80°C sustained for 2m',)),
            ({'value': 89, 'details': 'High samples span 2m.'},
             ('Recorded details: High samples span 2m.',)),
            ({'value': 89, 'reason': 'old raw text', 'details': 'old raw text'},
             ('Recorded reason: old raw text',)),
        ):
            with self.subTest(data=data):
                out = render('temp_high', data)
                self.assertNotIn('°C', out['title'] + out['body'].split('\n')[0])
                self.assertIn('without a complete measurement', out['body'])
                for fragment in fragments:
                    self.assertIn(fragment, out['body'])
                if data.get('reason') == data.get('details'):
                    self.assertEqual(out['body'].count('old raw text'), 1)
        self.assertNotIn('Recorded', render('temp_high', {})['body'])

    def test_incomplete_measurement_keeps_literal_braces_in_raw_context(self):
        out = render_fixture()('temp_high', {
            'reason': 'legacy {value} at {unknown}', 'details': 'sensor {threshold}',
        })
        self.assertIn('Recorded reason: legacy {value} at {unknown}', out['body'])
        self.assertIn('Recorded details: sensor {threshold}', out['body'])

    def test_shipped_italian_stale_override_and_isolated_locale_overlay(self):
        event = polled_event(produced_temperature())
        shipped = render_fixture('it')('temp_high', event.data)
        self.assertIn('CPU', shipped['title'] + shipped['body'])  # delivery gate, not a pass
        overlay = render_fixture('it', overlays=[('it',
            '{hostname}: temperatura elevata del sensore — {value}°C',
            'La temperatura del sensore ha raggiunto {value}°C (soglia: {threshold}°C).\n{details}')])
        proposed = overlay('temp_high', event.data)
        self.assertNotIn('CPU', proposed['title'] + proposed['body'])
        self.assertIn('89', proposed['title'])
        self.assertIn('80', proposed['body'])

    def test_collector_schema_categories_and_cpu_reason_are_localized(self):
        # Execute actual collector-to-formatter path against inert per-cycle inputs.
        class Stop(Exception): pass
        class Clock:
            count = 0
            def time(self): return 1.
            def sleep(self, _):
                self.count += 1
                if self.count > 1: raise Stop
        class Manager:
            _enabled = True
            _config = {}
            events = []
            def is_event_enabled(self, *_): return True
            def emit_event(self, **kw): self.events.append(kw)
        manager = Manager()
        scope = {'notification_manager': manager, 'time': Clock(),
                 'startup_grace': types.SimpleNamespace(should_suppress_category=lambda *_: False),
                 '_capture_health_journal_context': lambda *_: '',
                 'resolve_notification_hostname': lambda name, *_: name,
                 'health_monitor': types.SimpleNamespace(get_detailed_status=lambda: {
                     'hostname': 'node.example', 'details': {'cpu': {'status': 'WARNING',
                         'reason': 'CPU >85% sustained for 300s'}}, 'overall': 'WARNING', 'summary': 'x'},
                     cached_results={}, last_check_times={})}
        # Remove only infrastructure imports from the extracted function.
        tree = ast.parse((SCRIPTS / 'flask_server.py').read_text())
        fn = copy.deepcopy(next(n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name == '_health_collector_loop'))
        for node in ast.walk(fn):
            for field, value in ast.iter_fields(node):
                if isinstance(value, list):
                    setattr(node, field, [child for child in value if not (
                        isinstance(child, ast.ImportFrom) and child.module == 'health_monitor'
                        or isinstance(child, ast.Import) and any(a.name == 'startup_grace' for a in child.names))])
        exec(compile(ast.fix_missing_locations(ast.Module(body=[fn], type_ignores=[])), '<collector>', 'exec'), scope)
        with self.assertRaises(Stop): scope['_health_collector_loop']()
        self.assertEqual(len(manager.events), 1)
        payload = manager.events[0]['data']
        item = payload['health_degraded']['categories'][0]
        self.assertEqual(item['cat_key'], 'cpu')
        out = render_fixture('it')('health_degraded', {**payload, 'severity': 'WARNING'})
        self.assertIn('Utilizzo e temperatura della CPU', out['title'])
        self.assertEqual(out['body'], 'La CPU è superiore al 85% per 300s.')

    def test_legacy_key_and_unknown_reason_remain_honest(self):
        render = render_fixture('it')
        def payload(item):
            return {'hostname': 'node.example', 'health_degraded': {'categories': [item]}}
        legacy = render('health_degraded', payload({'key': 'cpu', 'status': 'WARNING',
                  'reason': 'CPU >85% sustained for 300s'}))
        self.assertIn('Utilizzo e temperatura della CPU', legacy['title'])
        self.assertIn('La CPU', legacy['body'])
        raw = render('health_degraded', payload({'cat_key': 'services', 'category': 'PVE Services',
                     'status': 'CRITICAL', 'reason': 'Services inactive: pvedaemon', 'entity': 'pvedaemon'}))
        self.assertIn('Servizi PVE', raw['title'])
        self.assertIn('pvedaemon', raw['title'])
        self.assertEqual(raw['body'], 'Services inactive: pvedaemon')
        unknown = render('health_degraded', payload({'cat_key': 'other', 'category': 'External category',
                         'reason': 'raw unknown', 'status': 'WARNING'}))
        self.assertIn('External category', unknown['title'])
        self.assertEqual(unknown['body'], 'raw unknown')

        conflicting = render('health_degraded', payload({'key': 'memory', 'cat_key': 'cpu',
            'status': 'WARNING', 'reason': 'CPU >85% sustained for 300s'}))
        self.assertIn('Memoria e scambio', conflicting['title'])
        self.assertEqual(conflicting['body'], 'CPU >85% sustained for 300s')
        multi = render('health_degraded', {'hostname': 'node.example',
            'health_degraded': {'categories': [
                {'cat_key': 'cpu', 'status': 'WARNING', 'reason': 'CPU >85% sustained for 300s'},
                {'cat_key': 'services', 'status': 'CRITICAL', 'reason': 'Services inactive: pvedaemon'}]}})
        self.assertIn('Utilizzo e temperatura della CPU', multi['body'])
        self.assertIn('Servizi PVE', multi['body'])
        self.assertIn('Services inactive: pvedaemon', multi['body'])


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