"""Synthetic local-vault, prompt-order and legacy --serve regression tests."""
import asyncio
from contextlib import redirect_stdout
from decimal import Decimal as D
import io
import json
import os
from pathlib import Path
import pty
import select
import signal
import socket
import subprocess
import sys
import tempfile
import threading
import time
import unittest
from unittest.mock import patch

sys.path.insert(0,str(Path(__file__).resolve().parents[1]))
import aiohttp
import gauntlet_v2 as g

SEED='GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ'


class VaultTests(unittest.TestCase):
    def setUp(self):
        self.temp=tempfile.TemporaryDirectory();self.root=Path(self.temp.name)
        self.vault=g.LocalVault(self.root/'auth.vault.json');self.vault.create('SYNTHETIC-PASSWORD',SEED)
    def tearDown(self):self.temp.cleanup()
    def test_rfc6238_sha1_six_digit_vectors(self):
        for instant,expected in [(59,'287082'),(1111111109,'081804'),(1111111111,'050471'),
                                 (1234567890,'005924'),(2000000000,'279037'),(20000000000,'353130')]:
            self.assertEqual(g.LocalVault.token(SEED,instant//30),expected)
    def test_encrypted_seed_and_exclusive_creation(self):
        data=self.vault.path.read_text();self.assertNotIn(SEED,data);self.assertNotIn('SYNTHETIC-PASSWORD',data)
        self.assertEqual(self.vault.path.stat().st_mode&0o777,0o600)
        with self.assertRaises(FileExistsError):self.vault.create('SYNTHETIC-PASSWORD',SEED)
    def test_password_and_token_are_both_verified(self):
        with self.assertRaises(g.AuthenticationFailed):self.vault.verify('wrong','287082',at=59)
        with self.assertRaises(g.AuthenticationFailed):self.vault.verify('SYNTHETIC-PASSWORD','111111',at=59)
        self.vault.verify('SYNTHETIC-PASSWORD','287082',at=59)
    def test_replay_survives_new_vault_instance(self):
        self.vault.verify('SYNTHETIC-PASSWORD','287082',at=59)
        with self.assertRaises(g.AuthenticationFailed):g.LocalVault(self.vault.path).verify('SYNTHETIC-PASSWORD','287082',at=59)
    def test_simultaneous_replay_only_one_succeeds(self):
        results=[]
        def attempt():
            try:self.vault.verify('SYNTHETIC-PASSWORD','287082',at=59);results.append(True)
            except g.AuthenticationFailed:results.append(False)
        workers=[threading.Thread(target=attempt) for _ in range(2)]
        for worker in workers:worker.start()
        for worker in workers:worker.join()
        self.assertEqual(sorted(results),[False,True])
    def test_non_ascii_or_wrong_length_token_rejected(self):
        for token in ('１２３４５６','12345','1234567','123 45'):
            with self.assertRaises(g.AuthenticationFailed):self.vault.verify('SYNTHETIC-PASSWORD',token,at=59)
    def test_ciphertext_tamper_rejected(self):
        value=json.loads(self.vault.path.read_text());value['ciphertext']='A'*48
        self.vault.path.write_text(json.dumps(value))
        with self.assertRaises(g.AuthenticationFailed):self.vault.verify('SYNTHETIC-PASSWORD','287082',at=59)
    def test_unsafe_file_permissions_rejected(self):
        self.vault.path.chmod(0o644)
        with self.assertRaises(g.AuthenticationFailed):self.vault.verify('SYNTHETIC-PASSWORD','287082',at=59)

    def test_auth_check_needs_no_config_keys_or_network(self):
        capture=io.StringIO()
        with patch.object(g.AdministratorGate,'prompt_factors',return_value=('SYNTHETIC-PASSWORD','287082')),             patch.object(g.time,'time',return_value=59),patch.dict(os.environ,{},clear=True),             patch.object(g.aiohttp,'ClientSession') as session,redirect_stdout(capture):
            result=g.main(['auth-check','--project',str(self.root)])
        self.assertEqual(result,0);session.assert_not_called()
        self.assertIn('LOCAL_FACTORS_VERIFIED',capture.getvalue())
        self.assertIn('"gateway_session":false',capture.getvalue())


class StartupTests(unittest.TestCase):
    def test_flag_aliases(self):
        for flags,expected in [(['--live'],'run'),(['--serve'],'observe'),(['--live','--serve'],'run')]:
            args=g.arguments(['--project','/tmp/project',*flags]);self.assertEqual(args.command,expected)
    def test_legacy_gateway_default_does_not_override_configured_https_url(self):
        with tempfile.TemporaryDirectory() as folder:
            root=Path(folder)
            settings={'gateway_hostname':'qzotic-gateway.sotality.com','gateway_port':443}
            (root/'security.json').write_text(json.dumps(settings))
            selected=g.security_settings(root,{})
            gate=g.AdministratorGate(selected,{'LIVE_AUTH_GATEWAY_URL':'https://localhost:8443'})
            self.assertEqual(gate.base,'https://localhost:8443')
            settings['pin_gateway_origin']=True
            (root/'security.json').write_text(json.dumps(settings))
            with self.assertRaises(g.AuthenticationFailed):
                g.AdministratorGate(g.security_settings(root,{}),{'LIVE_AUTH_GATEWAY_URL':'https://localhost:8443'})

    def test_prompts_precede_config_credentials_and_journal(self):
        calls=[]
        def prompt():calls.extend(['password','mfa']);return ('SYNTHETIC','123456')
        def config(root):
            self.assertEqual(calls,['password','mfa']);raise g.Fault('synthetic_config_error')
        with patch.object(g.AdministratorGate,'prompt_factors',side_effect=prompt),\
             patch.object(g,'configuration',side_effect=config),patch.object(g,'Journal') as journal:
            with self.assertRaises(g.Fault):g.main(['--live','--serve','--project','/tmp/nonexistent'])
        journal.assert_not_called()
    def test_real_terminal_exact_prompt_order_and_hidden_inputs(self):
        pid,fd=pty.fork()
        if pid==0:
            environment={k:v for k,v in os.environ.items() if k not in ('LIVE_BROKER_API_KEY','LIVE_BROKER_SECRET_KEY')}
            os.execve(sys.executable,[sys.executable,g.__file__,'--live','--serve','--project','/tmp/alphabot-synthetic-absent-project'],environment)
        transcript=b'';step=0;status=None;deadline=time.monotonic()+15
        prompts=[g.AdministratorGate.PROMPT.encode(),g.AdministratorGate.MFA_PROMPT.encode()]
        try:
            while time.monotonic()<deadline:
                readable,_,_=select.select([fd],[],[],.1)
                if readable:
                    try:chunk=os.read(fd,65536)
                    except OSError:break
                    if not chunk:break
                    transcript+=chunk
                    if step<2 and prompts[step] in transcript:
                        os.write(fd,b'SYNTHETIC-HIDDEN-PASSWORD\n' if step==0 else b'000000\n');step+=1
                result,status=os.waitpid(pid,os.WNOHANG)
                if result:break
            self.assertEqual(step,2,transcript.decode(errors='replace'))
            self.assertLess(transcript.index(prompts[0]),transcript.index(prompts[1]))
            self.assertNotIn(b'SYNTHETIC-HIDDEN-PASSWORD',transcript)
            self.assertNotIn(b'000000',transcript)
        finally:
            os.close(fd)
            try:
                result,_=os.waitpid(pid,os.WNOHANG)
                if result==0:os.kill(pid,signal.SIGTERM);os.waitpid(pid,0)
            except ChildProcessError:pass
    def test_deferred_binding_preserves_original_hash(self):
        with tempfile.TemporaryDirectory() as folder:
            root=Path(folder);g.initialize(root,'alpaca','SPY');cfg=g.configuration(root)
            old=g.Journal(root,cfg,'old-key');binding=old.get('binding');old.halt('synthetic_preserved_halt');old.close()
            new=g.Journal(root,cfg,'new-key',defer_binding_review=True)
            try:
                self.assertTrue(new.binding_mismatch);self.assertEqual(new.get('binding'),binding)
                self.assertTrue(new.halted.is_set());self.assertEqual(new.get('halt')['reason'],'synthetic_preserved_halt')
                with self.assertRaises(g.Fault):new.intent('buy',D('1'),D('100'))
            finally:new.close()


class ServeTests(unittest.IsolatedAsyncioTestCase):
    async def test_serve_ignores_config_auth_and_broker_credentials(self):
        with tempfile.TemporaryDirectory() as folder:
            root=Path(folder);g.initialize(root,'alpaca','SPY');cfg=g.configuration(root)
            journal=g.Journal(root,cfg,'synthetic')
            journal.observe('PREFLIGHT',{'cash':D('500'),'equity':D('500'),'gross_asset_value':D('0'),'quantity':D('0')})
            journal.close();(root/'config.json').write_text('intentionally invalid config')
            with socket.socket() as listener:listener.bind(('127.0.0.1',0));port=listener.getsockname()[1]
            env={k:v for k,v in os.environ.items() if not k.startswith(('LIVE_','CF_ACCESS_','ALPHABOT_AUTH_'))}
            process=subprocess.Popen([sys.executable,g.__file__,'--serve','--project',str(root),'--port',str(port),'--duration','3'],
                stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True,env=env)
            try:
                async with aiohttp.ClientSession() as client:
                    deadline=time.monotonic()+5;body=None
                    while time.monotonic()<deadline:
                        try:
                            async with client.get(f'http://127.0.0.1:{port}/api/window') as response:
                                body=await response.json();break
                        except aiohttp.ClientError:await asyncio.sleep(.05)
                    self.assertIsNotNone(body);self.assertEqual(body['selected']['snapshot']['equity'],'500')
                stdout,stderr=await asyncio.to_thread(process.communicate,timeout=6)
                self.assertEqual(process.returncode,0,stdout+stderr)
                self.assertNotIn('Password:',stdout+stderr)
                self.assertNotIn('STARTUP_FAILED',stdout+stderr)
            finally:
                if process.poll() is None:process.terminate();await asyncio.to_thread(process.wait,timeout=3)