import base64
import copy
import hashlib
import json
import os
from pathlib import Path
import subprocess
import tempfile
import unittest
from unittest.mock import patch

import field_profiles as fp
import field_network as fn
import field_runtime as fr
import field_cli as cli
import field_ca
import network_apply
import network_prepare
import connection


def fixture_profile():
    return fp.create('office','스인 시험','eno1','192.0.2.40/24','192.0.2.42')


def fixture_compose():
    d={'services':{k:{'environment':{},'volumes':[],'image':'sha256:fixed'} for k in ('analytics','ai-control-sidecar','ai-gateway','postgresql','redis')}}
    d['services']['redis']['ports']=[{'target':6379,'published':'16379','host_ip':'192.0.2.40'}]
    d['services']['ai-gateway']['ports']=[{'target':38443,'published':'38443'}]
    return d


class FieldTests(unittest.TestCase):
    def test_profile_accepts_dhcp_and_static(self):
        p=fixture_profile();self.assertEqual(fp.validate(p)['network']['mode'],'dhcp')
        p['network'].update(mode='static',address='192.0.2.40/24')
        self.assertEqual(fp.validate(p)['network']['address'],'192.0.2.40/24')

    def test_profile_rejects_secret_and_unknown_fields(self):
        p=fixture_profile();p['password']='never-publish'
        with self.assertRaises(ValueError):fp.validate(p)
        p=fixture_profile();p['webs'][0]['token']='never-publish'
        with self.assertRaises(ValueError):fp.validate(p)

    def test_profile_endpoint_identity_and_allowlist(self):
        p=fixture_profile();p['webs'][0]['event_url']='https://192.0.2.43/api/v1/events'
        with self.assertRaises(ValueError):fp.validate(p)
        p=fixture_profile();p['redis_allow']=[]
        with self.assertRaises(ValueError):fp.validate(p)
        p=fixture_profile();p['webs'].append(copy.deepcopy(p['webs'][0]))
        with self.assertRaises(ValueError):fp.validate(p)

    def test_profile_refuses_shell_and_endpoint_credentials(self):
        p=fixture_profile();p['network']['interface']='eno1;reboot'
        with self.assertRaises(ValueError):fp.validate(p)
        p=fixture_profile();p['webs'][0]['manager']='https://user:pw@192.0.2.42'
        with self.assertRaises(ValueError):fp.validate(p)

    def test_port_boolean_and_collisions(self):
        p=fixture_profile();p['control_port']=True
        with self.assertRaises(ValueError):fp.validate(p)
        p=fixture_profile();p['redis_port']=p['control_port']
        with self.assertRaises(ValueError):fp.validate(p)

    def test_secrets_are_profile_and_web_scoped(self):
        p=fixture_profile();a=fp.secret_path(p)
        p['id']='field';self.assertNotEqual(a,fp.secret_path(p))
        b=fp.secret_path(p);p['webs'][0]['id']='standby';p['selected_web']='standby'
        self.assertNotEqual(b,fp.secret_path(p))

    def test_atomic_and_symlink_protection(self):
        with tempfile.TemporaryDirectory() as root:
            path=Path(root)/'setting.json';fp.atomic(path,{'value':1})
            self.assertEqual(fp.read_json(path),{'value':1})
            self.assertEqual(path.stat().st_mode&0o777,0o600)
            link=Path(root)/'link';link.symlink_to(path)
            with self.assertRaises(ValueError):fp.atomic(link,{'value':2})
            self.assertEqual(fp.read_json(path),{'value':1})

    def test_network_has_no_invented_gateway_dns(self):
        p=fixture_profile();p['network'].update(mode='static',address='192.0.2.40/24')
        e=fn.render(p)['network']['ethernets']['eno1']
        self.assertFalse(e['dhcp4']);self.assertNotIn('routes',e);self.assertNotIn('nameservers',e)

    def test_network_wait_is_bounded_without_stopping_dhcp(self):
        unit=fn.wait_unit('eno1')
        self.assertIn('TimeoutStartSec=60',unit);self.assertIn('--timeout=60',unit)
        self.assertNotIn('systemd-networkd.service',unit);self.assertNotIn('ExecStop=',unit)

    def test_dhcp_requires_unambiguous_actual_address(self):
        with patch.object(fn,'addresses',return_value=['192.0.2.40']):self.assertEqual(fn.ai_ip(fixture_profile()),'192.0.2.40')
        with patch.object(fn,'addresses',return_value=[]):
            with self.assertRaises(ValueError):fn.ai_ip(fixture_profile())

    def test_only_supported_compose_changes_are_accepted(self):
        base=fixture_compose();new=copy.deepcopy(base)
        new['services']['ai-control-sidecar']['environment']['AI_CONTROL_MANAGER_BASE_URL']='https://192.0.2.42'
        new['services']['redis']['ports'][0]['host_ip']='192.0.2.41'
        fr.check_drift(base,new)
        new['services']['analytics']['image']='sha256:other'
        with self.assertRaises(ValueError):fr.check_drift(base,new)

    def test_volume_and_command_changes_not_emergency_bypass(self):
        for field,value in [('command',['sh','-c','anything']),('privileged',True),('volumes',[{'type':'bind','source':'/','target':'/host'}])]:
            base=fixture_compose();new=copy.deepcopy(base);new['services']['analytics'][field]=value
            with self.assertRaises(ValueError):fr.check_drift(base,new)

    def test_redis_target_not_mutable(self):
        base=fixture_compose();new=copy.deepcopy(base);new['services']['redis']['ports'][0]['target']=9999
        with self.assertRaises(ValueError):fr.check_drift(base,new)

    def test_firewall_multiple_peers_is_explicit_and_scoped(self):
        script=network_apply.firewall_script('192.0.2.40',['192.0.2.42','192.0.2.43'],16379)
        compile(script,'allowlist','exec')
        self.assertIn('192.0.2.42/32',script);self.assertIn('192.0.2.43/32',script)
        self.assertNotIn("run('-F','INPUT')",script)
        with self.assertRaises(ValueError):network_apply.firewall_script('192.0.2.40',[],16379)

    def test_ca_picker_ignores_private_extensions_and_symlinks(self):
        with tempfile.TemporaryDirectory() as root:
            root=Path(root);(root/'public.crt').write_text('public');(root/'private.key').write_text('secret')
            (root/'escape.crt').symlink_to(root/'private.key')
            self.assertEqual([p.name for p in field_ca.entries(root)],['public.crt'])

    def test_ca_private_content_rejected_before_copy(self):
        with tempfile.TemporaryDirectory() as root:
            path=Path(root)/'fake.crt';path.write_text('-----BEGIN PRIVATE KEY-----\nfixture')
            with self.assertRaises(ValueError):connection.verify_ca(path)

    def test_enter_keeps_value_clear_is_explicit(self):
        with patch('builtins.input',return_value=''):self.assertEqual(cli.ask('x','current',clear=True),'current')
        with patch('builtins.input',return_value='-'):self.assertEqual(cli.ask('x','current',clear=True),'')

    def test_profile_credentials_store_without_login(self):
        with tempfile.TemporaryDirectory() as root,patch.object(fp,'PROFILES',Path(root)),patch.object(os,'geteuid',return_value=1001),patch('builtins.input',return_value='new-id'),patch.object(cli.getpass,'getpass',side_effect=['new-password','new-password']):
            p=fixture_profile();cli.credentials(p)
            self.assertEqual(fp.read_json(fp.secret_path(p)),{'loginId':'new-id','password':'new-password'})

    def test_root_console_reads_si_credentials_without_sudo_uid(self):
        with tempfile.TemporaryDirectory() as root:
            path=Path(root)/'credentials.json';fp.atomic(path,{'loginId':'fixture','password':'not-real'})
            os.chown(path,1001,1001)
            user=type('User',(),{'pw_uid':1001})()
            with patch.dict(os.environ,{},clear=True),patch.object(fp,'account',return_value=user):
                self.assertEqual(fp.read_credentials(path)['loginId'],'fixture')
                path.chmod(0o644)
                with self.assertRaises(ValueError):fp.read_credentials(path)

    def test_invalid_profile_can_still_be_opened_for_repair(self):
        with tempfile.TemporaryDirectory() as root,patch.object(fp,'STATE',Path(root)/'state'),patch.object(fp,'PROFILES',Path(root)/'profiles'):
            fp.atomic(fp.STATE/'selected.json',{'profile':'office'})
            fp.atomic(fp.path('office'),b'{invalid json')
            def repair(*args,**kwargs):fp.atomic(fp.path('office'),fixture_profile())
            with patch.object(fp,'account',return_value=type('User',(),{'pw_name':'si'})()),patch.object(cli.subprocess,'run',side_effect=repair) as editor:
                cli.profile_action(['file'])
                self.assertEqual(editor.call_args.args[0][-1],str(fp.path('office')))

    def test_network_restore_detects_later_edit(self):
        with tempfile.TemporaryDirectory() as root:
            root=Path(root);state=root/'state';net=root/'net';net.mkdir();state.mkdir()
            (net/'a.yaml').write_bytes(b'user-later')
            fp.atomic(state/'network-pending.json',{'before':{'a.yaml':base64.b64encode(b'old').decode()},'after':{'a.yaml':hashlib.sha256(b'new').hexdigest()}})
            with patch.object(fp,'STATE',state),patch.object(fn,'NETPLAN',net):
                with self.assertRaises(ValueError):fn.restore()
            self.assertEqual((net/'a.yaml').read_bytes(),b'user-later')

    def test_runtime_replay_checks_all_targets_before_writing(self):
        with tempfile.TemporaryDirectory() as root:
            root=Path(root);runtime=root/'runtime';unit=root/'unit'
            paths=[runtime/'state'/n for n in ('site.json','active.json','inflight.json')]+[unit,root/'data/site/redis-access.sh']
            writes=[]
            for p in paths:
                p.parent.mkdir(parents=True,exist_ok=True);p.write_bytes(b'old')
                writes.append({'path':str(p),'before':base64.b64encode(b'old').decode(),'after':base64.b64encode(b'new').decode(),'mode':0o644 if p==unit else 0o700 if p.name.endswith('.sh') else 0o600})
            with patch.object(fp,'RUNTIME',runtime),patch.object(connection,'UNIT',unit):
                paths[-1].write_bytes(b'other')
                with self.assertRaises(ValueError):fr.replay_writes({'writes':writes})
                self.assertEqual(paths[0].read_bytes(),b'old')
                paths[-1].write_bytes(b'old');paths[0].write_bytes(b'new')
                fr.replay_writes({'writes':writes});fr.replay_writes({'writes':writes})
                self.assertTrue(all(p.read_bytes()==b'new' for p in paths))

    def test_runtime_replay_refuses_arbitrary_root_path(self):
        with self.assertRaises(ValueError):fr.replay_writes({'writes':[{'path':'/etc/sudoers','before':None,'after':'','mode':0o600}]})

    def test_incomplete_transaction_preserved_and_rebuilt(self):
        with tempfile.TemporaryDirectory() as root,patch.object(fp,'STATE',Path(root)/'private'):
            tx=Path(root)/'field-test';stage=Path(root)/'stage-test'
            fr.prepare_transaction(tx,stage,'old')
            (tx/'partial').write_text('preserve')
            fr.prepare_transaction(tx,stage,'old')
            self.assertEqual(list(tx.iterdir()),[])
            self.assertEqual(next(Path(root).glob('field-test-incomplete-*')).joinpath('partial').read_text(),'preserve')
            with self.assertRaises(ValueError):fr.prepare_transaction(tx,stage,'field-test')

    def test_prepared_profile_and_credentials_cannot_drift(self):
        with tempfile.TemporaryDirectory() as root,patch.object(fp,'STATE',Path(root)/'private'),patch.object(fp,'PROFILES',Path(root)/'profiles'):
            stage=Path(root)/'stage';stage.mkdir();p=fixture_profile()
            secret=fp.secret_path(p);secret.parent.mkdir(parents=True)
            secret.write_text('fixture-password');ca=fp.ca_path(p);ca.write_text('fixture-public')
            payload=stage/'payload';payload.write_text('frozen')
            record=dict(profile=p['id'],profile_stamp=fp.stamp(p),desired={},base_active_sha256='base',profile_snapshot=p,
                        prepared_files={'payload':fr.sha(payload)},profile_credentials_sha256=fr.sha(secret),profile_ca_sha256=fr.sha(ca))
            fp.atomic(stage/'record.json',record);fp.atomic(fp.STATE/'stages'/(stage.name+'.json'),record)
            with patch.object(fp,'load',return_value=p):
                fr.verify_pending(stage)
                secret.write_text('changed')
                with self.assertRaises(ValueError):fr.verify_pending(stage)
                secret.write_text('fixture-password');p['label']='changed-profile'
                with self.assertRaises(ValueError):fr.verify_pending(stage)

    def test_receipt_does_not_export_credentials(self):
        with tempfile.TemporaryDirectory() as root,patch.object(fp,'STATE',Path(root)/'private'),patch.object(fp,'SETUP',Path(root)/'setup'),patch.object(os,'geteuid',return_value=1001):
            stage=Path(root)/'field-fixture';p=fixture_profile()
            fp.atomic(fp.STATE/'stages'/(stage.name+'.json'),dict(profile=p['id'],profile_snapshot=p,profile_ca_sha256='public-hash',
                      profile_credentials_sha256='private-hash',prepared_files={'password':'never-export'}))
            out=fr.export_receipt(stage,'APPLIED_SERVICES_STOPPED')
            contents=''.join(x.read_text() for x in out.iterdir())
            self.assertNotIn('private-hash',contents);self.assertNotIn('never-export',contents)
            self.assertIn('USER_CHECK_PENDING',contents)

    def test_gateway_mismatched_port_not_silently_imported(self):
        doc=fixture_compose();doc['services']['ai-gateway']['environment']['AI_CONTROL_HOST_PORT']='39000'
        with self.assertRaises(ValueError):fr.public_values(doc)

    def test_apply_resume_after_interruption_then_restore(self):
        with tempfile.TemporaryDirectory() as root:
            root=Path(root);runtime=root/'runtime';state=root/'private';unit=root/'unit';stage=root/'field-fixture'
            (runtime/'state').mkdir(parents=True);stage.mkdir();p=fixture_profile()
            paths=[runtime/'state'/n for n in ('site.json','active.json','inflight.json')]+[unit,root/'data/site/redis-access.sh']
            rows=[]
            for path in paths:
                path.parent.mkdir(parents=True,exist_ok=True);path.write_bytes(b'old')
                rows.append({'path':str(path),'before':base64.b64encode(b'old').decode(),'after':base64.b64encode(b'new').decode(),
                             'mode':0o644 if path==unit else 0o700 if path.name.endswith('.sh') else 0o600})
            fp.atomic(state/'apply-pending.json',{'stage':str(stage),'transaction':stage.name,'writes':rows})
            fp.atomic(state/'pending.json',{'stage':str(stage),'profile':p['id']})
            record={'desired':{'ai_ip':'192.0.2.40'}}
            calls=[]
            def interrupted(args,timeout=30):
                calls.append(args)
                if '-c' in args:raise RuntimeError('fixture firewall interrupted')
                return ''
            with patch.object(fp,'RUNTIME',runtime),patch.object(fp,'STATE',state),patch.object(connection,'UNIT',unit),patch.object(fr,'libraries'),patch.object(fr,'verify_pending',return_value=(record,p)),patch.object(fr,'stage_ready'),patch.object(fr,'remember_baseline'),patch.object(fr,'export_receipt'),patch.object(fr.shutil,'disk_usage',return_value=type('Disk',(),{'free':20*1024**3})()):
                with patch.object(fr,'run',side_effect=interrupted):
                    with self.assertRaises(RuntimeError):fr.apply_stage(stage)
                self.assertTrue((state/'apply-pending.json').exists())
                self.assertTrue(all(path.read_bytes()==b'new' for path in paths))
                with patch.object(fr,'run',return_value='') as execute:
                    fr.apply_stage(stage)
                    self.assertFalse((state/'apply-pending.json').exists())
                    self.assertTrue((state/'last-applied.json').exists())
                    fr.restore_config()
                    self.assertTrue(all(path.read_bytes()==b'old' for path in paths))
                    self.assertFalse((state/'last-applied.json').exists())
                    self.assertFalse(any('start' in call.args[0] or 'up' in call.args[0] for call in execute.call_args_list))

    def test_edit_root_routes_to_si_not_root_home(self):
        import ct
        with patch.object(ct,'command') as command,patch.object(os,'geteuid',return_value=0):
            ct.user_command(['ct','work','end'])
            self.assertEqual(command.call_args.args[0][:4],['runuser','-u','si','--'])

    def test_prepare_uses_new_profile_credentials_and_server_id(self):
        with tempfile.TemporaryDirectory() as root:
            root=Path(root);runtime=root/'runtime';data=root/'data';state=root/'private';state.mkdir()
            (runtime/'state').mkdir(parents=True);(runtime/'state/active.json').write_text('{"transaction":"previous"}')
            (runtime/'state/site.json').write_text('{"ai_ip":"192.0.2.1"}')
            (data/'site').mkdir(parents=True);(data/'site/redis-access.sh').write_text('fixture')
            (root/'setup/outbox').mkdir(parents=True)
            bridge=root/'old-bridge';bridge.mkdir();tls=root/'old-tls';tls.mkdir()
            for name in ('password','control-token','relay-token'):(bridge/name).write_text('old-fixture')
            (bridge/'login-body.json').write_text(json.dumps({'loginId':'old-id','redis':{'host':'192.0.2.1','port':16379}}))
            (tls/'control-tls.key').write_text('fixture-key')
            doc=fixture_compose()
            doc['services']['ai-control-sidecar']['volumes']=[{'type':'bind','source':str(bridge),'target':'/run/bridge'},{'type':'bind','source':str(root/'session'),'target':'/run/bridge-session'}]
            doc['services']['ai-control-sidecar']['environment']={'AI_CONTROL_SESSION_FILE':'/run/bridge-session/session.json'}
            doc['services']['ai-gateway']['volumes']=[{'type':'bind','source':str(tls),'target':'/run/ai-tls'}]
            cp=root/'compose.json';cp.write_text(json.dumps(doc));unit=root/'unit';unit.write_text('unit')
            ca=root/'ca.crt';ca.write_text('public-fixture')
            wanted=dict(ai_ip='192.0.2.40',web_ip='192.0.2.42',manager_base_url='https://192.0.2.42',event_endpoint='https://192.0.2.42/api/v1/events',control_port=38443,redis_port=16379,artifact_lan_cidr='192.0.2.0/24')
            plan=state/'field-fixture'/'plan.json';plan.parent.mkdir()
            plan.write_text(json.dumps({'desired':wanted,'baseline':{'compose_sha256':fr.sha(cp),'unit_sha256':fr.sha(unit)}}))
            def fake_tls(path,ip):(path/'control-tls.csr').write_text('fixture-csr')
            with patch.object(fp,'STATE',state),patch.object(network_prepare,'RUNTIME',runtime),patch.object(network_prepare,'DATA',data),patch.object(connection,'STATE',state),patch.object(connection,'UNIT',unit),patch.object(connection,'active_compose',return_value=cp),patch('tools.secure_bridge_provision.validate_ca'),patch('tools.secure_bridge_tls.prepare',side_effect=fake_tls),patch.object(os,'chown'):
                stage=network_prepare.prepare(plan,ca,fr.sha(ca),credentials={'loginId':'new-id','password':'new-password'},server_id='AI-NEW')
            self.assertEqual((stage/'bridge/password').read_text(),'new-password\n')
            self.assertEqual(json.loads((stage/'bridge/login-body.json').read_text())['loginId'],'new-id')
            env=json.loads((stage/'host-enrollment.json').read_text())
            self.assertEqual(env['AI_CONTROL_SERVER_ID'],'AI-NEW')
            self.assertEqual(list((stage/'bridge-session').iterdir()),[])
            self.assertEqual((bridge/'password').read_text(),'old-fixture')
            self.assertTrue((root/'setup/outbox/field-fixture.csr').is_file())


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