import json
import hashlib
from pathlib import Path
import subprocess
import tempfile
import unittest


ROOT = Path(__file__).resolve().parents[1]


class WorkerHealthContractTests(unittest.TestCase):
    def test_runtime_identity_prefers_apache_worker_identity_over_stale_dotenv(self) -> None:
        runtime_identity = ROOT / 'classes' / 'RuntimeIdentity.php'
        program = f'''<?php
require {json.dumps(str(runtime_identity))};
$_ENV['BEE_WORKER_ID'] = '1';
$_ENV['BEE_WORKER_GENERATION'] = '1';
$_SERVER['BEE_WORKER_ID'] = '2';
$_SERVER['BEE_WORKER_GENERATION'] = '3';
$class = new ReflectionClass('RuntimeIdentity');
foreach (['workerId', 'generation'] as $methodName) {{
    $method = $class->getMethod($methodName);
    $method->setAccessible(true);
    echo $method->invoke(null), "\\n";
}}
'''
        result = subprocess.run(
            ['php', '-r', program.removeprefix('<?php\n')],
            check=True,
            capture_output=True,
            text=True,
        )
        self.assertEqual(['2', '3'], result.stdout.splitlines())

    def test_runtime_identity_contains_only_worker_health_contract(self) -> None:
        source = (ROOT / 'classes' / 'RuntimeIdentity.php').read_text(encoding='utf-8')
        for field in (
            'worker_id', 'bee_runtime_sha256', 'bot_runtime_sha256', 'bot_tree_sha256',
            'generation', 'timing_policy_version', 'redis', 'data_handler_status',
            'ws_status', 'database', 'active_test_count', 'disk_free_bytes', 'timestamp',
            'active_test_ids',
        ):
            self.assertIn(field, source)
        health_return = source[source.index('return ['):source.index('];', source.index('return ['))]
        self.assertNotIn("'password' =>", health_return)
        self.assertNotIn("'dsn' =>", health_return)

    def test_runtime_identity_keeps_manifest_and_tree_identities_separate(self) -> None:
        source = (ROOT / 'classes' / 'RuntimeIdentity.php').read_text(encoding='utf-8')
        self.assertIn("'tree_sha256'", source)
        bot_digest = source[source.index('private static function botDigest'):source.index('private static function redisStatus')]
        self.assertIn("hash_file('sha256', $manifest)", bot_digest)
        self.assertIn('private static function botTreeDigest', bot_digest)

    def test_runtime_identity_reports_manifest_and_tree_sha(self) -> None:
        runtime_identity = ROOT / 'classes' / 'RuntimeIdentity.php'
        with tempfile.TemporaryDirectory() as temporary:
            manifest = Path(temporary) / 'runtime-manifest.json'
            manifest.write_text(json.dumps({'tree_sha256': 'a' * 64}), encoding='utf-8')
            manifest_sha256 = hashlib.sha256(manifest.read_bytes()).hexdigest()
            program = f'''<?php
require {json.dumps(str(runtime_identity))};
$_ENV['BOT_RUNTIME_MANIFEST'] = {json.dumps(str(manifest))};
$class = new ReflectionClass('RuntimeIdentity');
foreach (['botDigest', 'botTreeDigest'] as $methodName) {{
    $method = $class->getMethod($methodName);
    $method->setAccessible(true);
    echo $method->invoke(null), "\\n";
}}
'''
            result = subprocess.run(
                ['php', '-r', program.removeprefix('<?php\n')],
                check=True,
                capture_output=True,
                text=True,
            )
        self.assertEqual(
            [manifest_sha256, 'a' * 64],
            result.stdout.splitlines(),
        )

    def test_health_exposes_only_active_test_identifiers_owned_by_the_worker(self) -> None:
        source = (ROOT / 'classes' / 'RuntimeIdentity.php').read_text(encoding='utf-8')
        self.assertIn('SELECT `id`, `speed` FROM `bee_tests` WHERE `status` = 1 AND `bee_worker_id` = ?', source)
        self.assertIn("'active_test_ids' => $activeTestIds", source)
        self.assertNotIn("SELECT * FROM `bee_tests`", source)

    def test_health_reports_active_capacity_units_from_the_worker_test_speeds(self) -> None:
        source = (ROOT / 'classes' / 'RuntimeIdentity.php').read_text(encoding='utf-8')
        self.assertIn('SELECT `id`, `speed` FROM `bee_tests`', source)
        self.assertIn("$activeCapacityUnits += max(1, (int)$row['speed'])", source)
        self.assertIn("'active_capacity_units' => $activeCapacityUnits", source)

    def test_health_requires_real_redisjson_commands_and_a_positive_worker_generation(self) -> None:
        source = (ROOT / 'classes' / 'RuntimeIdentity.php').read_text(encoding='utf-8')
        self.assertIn("['COMMAND', 'INFO', 'JSON.SET', 'JSON.GET']", source)
        self.assertIn("BEE_WORKER_GENERATION", source)
        self.assertIn("'generation' => $generation", source)
        self.assertNotIn("'json_module_ready' => is_array($modules)", source)

    def test_private_health_uses_authenticated_init_and_identity(self) -> None:
        endpoint = (ROOT / 'api' / 'private' / 'health.php').read_text(encoding='utf-8')
        self.assertIn("define('BEE_WORKER_HEALTH_REQUEST', true)", endpoint)
        self.assertIn("require_once(ROOT_PATH . '/init.inc.php')", endpoint)
        self.assertIn('RuntimeIdentity::health()', endpoint)
        init = (ROOT / 'init.inc.php').read_text(encoding='utf-8')
        self.assertIn("$is_worker_health_endpoint", init)
        self.assertIn("defined('BEE_WORKER_HEALTH_REQUEST')", init)
        self.assertIn("hash_equals($expected_signature, $supplied_signature)", init)
        self.assertIn("bee-worker.env", init)

    def test_websocket_writes_and_removes_readiness_marker(self) -> None:
        source = (ROOT / 'ws' / 'main.py').read_text(encoding='utf-8')
        self.assertIn("/run/razgar/bee-ws.ready", source)
        self.assertIn("os.unlink('/run/razgar/bee-ws.ready')", source)
        self.assertIn("raise RuntimeError('BEE websocket readiness marker could not be written')", source)


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