#!/usr/bin/env python3
"""Offline tests for maintenance isolation, selection, inventory and protections."""
import contextlib
import io
import json
import os
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile
import unittest
from unittest.mock import Mock, patch

import test_maintenance_review as prior

MAINT = prior.MAINT
BASH = prior.BASH
sdputils = prior.sdputils
sys.path.insert(0, str(MAINT))
import file_inventory


class Workspaces(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory(prefix='sdp-workspace-test-')
        self.addCleanup(self.tmp.cleanup)
        self.root = Path(self.tmp.name)
        self.script = self.root / 'runner.sh'
        self.script.write_text('''#!/bin/bash
source "$1" || exit 1
MAINT_DIR=$2
P4TMP=$3
_sdp_on_exit() { echo "$1" > "$MAINT_DIR/exit.$4"; }
maintenance_workspace_init "$4" || exit $?
echo "$PWD"
echo report > clients.txt
read -r ignored
exit "${TEST_EXIT:-0}"
'''.replace('"$MAINT_DIR/exit.$4"', '"$MAINT_DIR/exit"'))

    def start(self, instance='1', exitcode=0):
        process = subprocess.Popen([BASH, str(self.script), str(MAINT / 'maintenance_workspace.sh'),
                                    str(self.root), str(self.root / instance), instance],
                                   stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                   text=True, env=dict(os.environ, TEST_EXIT=str(exitcode)))
        self.addCleanup(lambda: process.poll() is None and process.kill())
        return process

    def test_same_instance_excluded_other_instance_independent_cleanup_and_status(self):
        first = self.start(exitcode=7)
        self.assertIn('Maintenance run directory:', first.stdout.readline())
        first_dir = Path(first.stdout.readline().strip())
        self.assertTrue(first_dir.is_dir())
        blocked = self.start()
        out, err = blocked.communicate(timeout=10)
        self.assertEqual(blocked.returncode, 75, err)
        self.assertIn('Another maintenance run', out)
        second = self.start('2')
        self.assertIn('Maintenance run directory:', second.stdout.readline())
        second_dir = Path(second.stdout.readline().strip())
        self.assertNotEqual(first_dir, second_dir)
        self.assertTrue((first_dir / 'clients.txt').exists())
        second.communicate('\n', timeout=10)
        self.assertFalse(second_dir.exists())
        self.assertTrue(first_dir.exists())
        first.communicate('\n', timeout=10)
        self.assertEqual(first.returncode, 7)
        self.assertEqual((self.root / 'exit').read_text().strip(), '7')
        self.assertFalse(first_dir.exists())
        # Kernel lock releases despite the persistent lock filename.
        again = self.start()
        self.assertIn('Maintenance run directory:', again.stdout.readline())
        again.communicate('\n', timeout=10)
        self.assertEqual(again.returncode, 0)

    def test_workspace_setup_failure_does_not_remove_unowned_directory(self):
        path = self.root / '1'
        path.write_text('keep')
        result = self.start()
        out, err = result.communicate(timeout=10)
        self.assertNotEqual(result.returncode, 0, out + err)
        self.assertEqual(path.read_text(), 'keep')

    def test_wrappers_share_helper_keep_limit_and_server_lock_action(self):
        for name in ['maintenance', 'edge_maintenance']:
            text = (MAINT / name).read_text()
            self.assertIn('maintenance_workspace_init "$P4INSTANCE"', text)
            self.assertIn('--limit 200 --clients-file client_snapshot.json', text)
            self.assertIn('rm -rf "/p4/${P4INSTANCE}/root/server.locks"', text)


class Selection(unittest.TestCase):
    NOW = 2_000_000_000
    WEEK = 604800

    def records(self):
        return [{'client': 'z-old', 'Access': str(self.NOW - 5*self.WEEK)},
                {'client': 'at-cutoff', 'Access': str(self.NOW - 4*self.WEEK)},
                {'client': 'recent', 'Access': str(self.NOW - 4*self.WEEK + 1)},
                {'client': 'a-SWARM-old', 'Access': '1'},
                {'client': 'invalid', 'Access': 'bad'},
                {'client': 'z-old', 'Access': '1'}]

    def test_snapshot_preserves_server_order_age_boundary_and_swarm_exclusion(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = Path(tmp) / 'clients.json'
            sdputils.write_client_snapshot(path, '1', 'server', 4, self.records(), self.NOW)
            self.assertEqual(sdputils.read_client_snapshot(path, '1', 'server', 4, self.NOW),
                             ['z-old', 'at-cutoff'])

    def test_snapshot_rejects_other_instance_policy_server_stale_future_or_active(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = Path(tmp) / 'clients.json'
            sdputils.write_client_snapshot(path, '1', 'server', 4, self.records(), self.NOW)
            data = json.loads(path.read_text())
            invalid = [('instance', '2'), ('server', 'different'), ('weeks', 5),
                       ('generated_at', self.NOW - 3601), ('generated_at', self.NOW + 1),
                       ('clients', [{'client': 'active', 'access': self.NOW}]),
                       ('clients', [{'client': 'bad\nname', 'access': 1}]),
                       ('clients', [{'client': 'x', 'access': True}])]
            for key, value in invalid:
                with self.subTest(key=key, value=value):
                    path.write_text(json.dumps(dict(data, **{key: value})))
                    with self.assertRaises(ValueError):
                        sdputils.read_client_snapshot(path, '1', 'server', 4, self.NOW)

    def test_snapshot_unload_rechecks_candidates_and_keeps_200_cap(self):
        with tempfile.TemporaryDirectory() as tmp, prior.module('unload_clients') as mod:
            path = Path(tmp) / 'clients.json'
            records = [{'client': str(index), 'Access': '1'} for index in range(250)]
            sdputils.write_client_snapshot(path, '1', 'test:1666', 4, records)
            api = Mock()
            api.run.side_effect = lambda *args: [{'client': args[-1], 'Access': '1'}] if args[0] == 'clients' else []
            with patch.object(mod, 'connect_p4', return_value=api):
                self.assertEqual(mod.main(['1', '--limit', '200', '--clients-file', str(path)]), 0)
            self.assertEqual(api.run.call_count, 400)
            self.assertEqual([call.args[0] for call in api.run.call_args_list], ['clients', 'unload'] * 200)
            self.assertEqual(api.run.call_args_list[0].args[-1], '0')
            self.assertEqual(api.run.call_args_list[-1].args[-1], '199')
            api.disconnect.assert_called_once()

    def test_snapshot_rechecks_each_client_after_previous_mutation_and_fills_limit_after_skips(self):
        with prior.module('unload_clients') as mod:
            api = Mock()
            unloaded = []
            def command(*args):
                if args[0] == 'clients':
                    # Client second became active while the first was unloaded.
                    access = self.NOW if args[-1] == 'second' and unloaded else 1
                    return [{'client': args[-1], 'Access': str(access)}]
                unloaded.append(args[-1])
                return []
            api.run.side_effect = command
            with patch.object(mod.time, 'time', return_value=self.NOW):
                self.assertTrue(mod.unload_snapshot_clients(api, ['first', 'second', 'third', 'fourth'], 4, 2))
            self.assertEqual(unloaded, ['first', 'third'])
            self.assertNotIn(('clients', '-e', 'fourth'), [call.args for call in api.run.call_args_list])

    def test_snapshot_recheck_ignores_missing_other_name_and_invalid_access(self):
        with prior.module('unload_clients') as mod:
            api = Mock()
            api.run.side_effect = [[], [{'client': 'different', 'Access': '1'}],
                                   [{'client': 'invalid', 'Access': 'bad'}]]
            self.assertTrue(mod.unload_snapshot_clients(api, ['missing', 'pattern', 'invalid'], 4))
            self.assertTrue(all(call.args[0] == 'clients' for call in api.run.call_args_list))

    def test_invalid_snapshot_never_unloads(self):
        with tempfile.TemporaryDirectory() as tmp, prior.module('unload_clients') as mod:
            path = Path(tmp) / 'clients.json'
            path.write_text('{}')
            api = Mock()
            with patch.object(mod, 'connect_p4', return_value=api), self.assertRaises(ValueError):
                mod.main(['1', '--clients-file', str(path)])
            api.run.assert_not_called()
            api.disconnect.assert_called_once()

    def test_report_queries_clients_once_and_publishes_only_after_all_queries_succeed(self):
        with tempfile.TemporaryDirectory() as tmp, prior.module('accessdates') as mod:
            path = Path(tmp) / 'clients.json'
            api = Mock()
            api.run.side_effect = lambda cmd: self.records() if cmd == 'clients' else []
            with patch.object(mod, 'connect_p4', return_value=api), patch.object(mod, 'write_list'), patch.object(mod.time, 'time', return_value=self.NOW):
                self.assertEqual(mod.main(['1', '--client-snapshot', str(path)]), 0)
            self.assertEqual([c.args for c in api.run.call_args_list].count(('clients',)), 1)
            previous = path.read_bytes()
            api.run.side_effect = lambda cmd: (_ for _ in ()).throw(prior.P4Exception('labels denied')) if cmd == 'labels' else []
            with patch.object(mod, 'connect_p4', return_value=api), patch.object(mod, 'write_list'), self.assertRaises(prior.P4Exception):
                mod.main(['1', '--client-snapshot', str(path)])
            self.assertEqual(path.read_bytes(), previous)


class Notifications(unittest.TestCase):
    def test_single_client_scan_cached_owner_lookup_and_same_threshold_messages(self):
        now, week = 2_000_000_000, 604800
        records = [{'client': 'two-weeks', 'Access': str(now - 2*week), 'Owner': 'owner'},
                   {'client': 'one-week', 'Access': str(now - 3*week), 'Owner': 'owner'},
                   {'client': 'swarm-internal', 'Access': '1', 'Owner': 'owner'},
                   {'client': 'recent', 'Access': str(now), 'Owner': 'owner'}]
        with prior.module('email_pending_client_deletes') as mod:
            api = Mock()
            api.run.side_effect = lambda *args: records if args[0] == 'clients' else [{'Email': 'owner@example.test'}]
            with patch.object(mod.time, 'time', return_value=now), patch.object(mod.smtplib, 'SMTP') as smtp, patch.object(mod, '_send_warning') as send:
                mod.warn_users(api, 2, 3, 'smtp', 'admin@example.test', '', None, 'test')
            self.assertEqual([c.args for c in api.run.call_args_list], [('clients', '-a'), ('user', '-o', 'owner')])
            self.assertEqual([(c.args[4], c.args[5]) for c in send.call_args_list],
                             [('two-weeks', 'two weeks'), ('one-week', 'one week')])
            smtp.assert_called_once_with('smtp')

    def test_no_candidates_does_not_connect_to_mail_server(self):
        with prior.module('email_pending_client_deletes') as mod:
            api = Mock()
            api.run.return_value = []
            with patch.object(mod.smtplib, 'SMTP') as smtp:
                mod.warn_users(api, 2, 3, 'smtp', '', '', None, 'test')
            api.run.assert_called_once_with('clients', '-a')
            smtp.assert_not_called()

    def test_missing_list_owner_falls_back_to_spec_and_caches_user(self):
        with prior.module('email_pending_client_deletes') as mod:
            api = Mock()
            api.run.side_effect = [[{'Owner': 'alice'}], [{'Email': 'alice@example.test'}], [{'Owner': 'alice'}]]
            cache = {}
            self.assertEqual(mod._get_owner_email(api, 'a', email_cache=cache), 'alice@example.test')
            self.assertEqual(mod._get_owner_email(api, 'b', email_cache=cache), 'alice@example.test')
            self.assertEqual(api.run.call_count, 3)


class Inventory(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory(prefix='sdp-inventory-test-')
        self.addCleanup(self.tmp.cleanup)
        self.root = Path(self.tmp.name)
        self.output = self.root / 'p4_file_list.txt'
        self.output.write_text('previous\n')
        self.fixture = self.root / 'fixture.json'
        self.fake = self.root / 'p4'
        self.fake.write_text('#!' + sys.executable + '\n' + '''import json, marshal, sys
from pathlib import Path
record = json.loads(Path(__file__).with_name('fixture.json').read_text())[' '.join(sys.argv[3:])]
if 'raw' in record:
    sys.stdout.buffer.write(record['raw'].encode())
else:
    for item in record.get('records', []):
        marshal.dump({key.encode(): value.encode() if isinstance(value, str) else value for key, value in item.items()}, sys.stdout.buffer)
if record.get('stderr'): print(record['stderr'], file=sys.stderr)
sys.exit(record.get('rc', 0))
''')
        self.fake.chmod(0o755)
        self.data = {
            'depots': {'records': [{'code': 'stat', 'name': 'depot'}]},
            'files //depot/*': {'records': [{'code': 'stat', 'depotFile': '//depot/root file'}]},
            'dirs -D //depot/*': {'records': [{'code': 'stat', 'dir': '//depot/space directory'},
                                             {'code': 'stat', 'dir': '//depot/deleted'}]},
            'files //depot/space directory/...': {'records': [{'code': 'stat', 'depotFile': '//depot/space directory/deep/file'}]},
            'files //depot/deleted/...': {'records': [{'code': 'stat', 'depotFile': '//depot/deleted/old', 'action': 'delete'}]},
        }

    def run_inventory(self):
        self.fixture.write_text(json.dumps(self.data))
        return file_inventory.create_inventory(str(self.fake), self.output)

    def test_complete_root_space_recursive_deleted_paths_and_preserved_output_mode(self):
        self.output.chmod(0o640)
        self.assertEqual(self.run_inventory(), 3)
        self.assertEqual(self.output.read_text().splitlines(),
                         ['//depot/root file', '//depot/space directory/deep/file', '//depot/deleted/old'])
        self.assertEqual(self.output.stat().st_mode & 0o777, 0o640)
        self.assertEqual(list(self.root.glob('.p4_file_list.txt.*')), [])

    def test_empty_server_publishes_empty_list(self):
        self.data = {'depots': {'records': []}}
        self.assertEqual(self.run_inventory(), 0)
        self.assertEqual(self.output.read_text(), '')

    def test_no_matching_files_warning_is_valid_empty(self):
        empty = {'records': [{'code': 'error', 'generic': 17, 'severity': 2, 'data': 'no such file(s)'}], 'rc': 1}
        self.data['files //depot/*'] = empty
        self.data['dirs -D //depot/*'] = empty
        self.assertEqual(self.run_inventory(), 0)
        self.assertEqual(self.output.read_text(), '')

    def test_late_permission_failure_preserves_previous_list_and_exit_status(self):
        self.data['files //depot/deleted/...'] = {'records': [{'code': 'error', 'generic': 6, 'severity': 3, 'data': 'permission denied'}], 'rc': 7}
        with self.assertRaises(file_inventory.InventoryError) as result:
            self.run_inventory()
        self.assertEqual(result.exception.returncode, 7)
        self.assertIn('permission denied', str(result.exception))
        self.assertEqual(self.output.read_text(), 'previous\n')
        self.assertEqual(list(self.root.glob('.p4_file_list.txt.*')), [])

    def test_unknown_empty_error_is_not_silently_accepted(self):
        self.data['files //depot/*'] = {'records': [{'code': 'error', 'generic': 17, 'severity': 3, 'data': 'hard error'}], 'rc': 1}
        with self.assertRaises(file_inventory.InventoryError):
            self.run_inventory()
        self.assertEqual(self.output.read_text(), 'previous\n')

    def test_truncated_marshaled_output_preserves_previous_list(self):
        self.data['dirs -D //depot/*'] = {'raw': '{'}
        with self.assertRaises(file_inventory.InventoryError):
            self.run_inventory()
        self.assertEqual(self.output.read_text(), 'previous\n')

    def test_publication_failure_preserves_previous_list_and_cleans_temporary(self):
        with patch.object(file_inventory.os, 'replace', side_effect=PermissionError('publish denied')):
            with self.assertRaises(PermissionError):
                self.run_inventory()
        self.assertEqual(self.output.read_text(), 'previous\n')
        self.assertEqual(list(self.root.glob('.p4_file_list.txt.*')), [])

    def test_missing_filename_refuses_partial_publication(self):
        self.data['files //depot/*'] = {'records': [{'code': 'stat', 'wrong': 'field'}]}
        with self.assertRaises(file_inventory.InventoryError):
            self.run_inventory()
        self.assertEqual(self.output.read_text(), 'previous\n')


class Protections(unittest.TestCase):
    def test_selected_group_removed_for_all_hosts_others_preserved_byte_for_byte(self):
        source = ('# Comment group target * //...\r\nProtections:\r\n'
                  '\tread group target * //...\r\n'
                  '\twrite group target 10.2.0.0/16 //depot/...\r\n'
                  '\tread group "two words" proxy-10.0.0.1 "-//depot/space path/..."\r\n'
                  '\tread group keeper 10.0.0.1 //...\r\n'
                  '\tread user target * //...\r\n'
                  '\tread group target * "unterminated\r\n'
                  '\tread group last * //...')
        expected = ('# Comment group target * //...\r\nProtections:\r\n'
                    '\tread group keeper 10.0.0.1 //...\r\n'
                    '\tread user target * //...\r\n'
                    '\tread group target * "unterminated\r\n'
                    '\tread group last * //...')
        with tempfile.TemporaryDirectory() as tmp, prior.module('clean_protect') as mod:
            groups, protect = Path(tmp) / 'groups', Path(tmp) / 'protect'
            groups.write_text('# ignore\nTARGET\ntwo words\n')
            protect.write_bytes(source.encode())
            output = io.StringIO(newline='')
            with patch.object(sys, 'argv', ['clean_protect.py', str(groups), str(protect)]), contextlib.redirect_stdout(output):
                mod.main()
            self.assertEqual(output.getvalue(), expected)


class ScratchFiles(unittest.TestCase):
    @contextlib.contextmanager
    def working_directory(self):
        with tempfile.TemporaryDirectory(prefix='sdp-scratch-test-') as tmp:
            previous = os.getcwd()
            os.chdir(tmp)
            try:
                yield Path(tmp)
            finally:
                os.chdir(previous)

    def test_private_workspaces_are_unique_preserve_cwd_and_keep_failed_recovery(self):
        with self.working_directory() as root:
            with sdputils.private_workspace('test') as first, sdputils.private_workspace('test') as second:
                self.assertNotEqual(first, second)
                self.assertEqual(Path.cwd(), root.resolve())
            self.assertFalse(first.exists())
            self.assertFalse(second.exists())
            with self.assertLogs(level='ERROR') as logs, self.assertRaises(RuntimeError):
                with sdputils.private_workspace('recovery', preserve_on_failure=True) as recovery:
                    (recovery / 'owner.txt').write_text('original-owner')
                    raise RuntimeError('failed restore')
            self.assertEqual((recovery / 'owner.txt').read_text(), 'original-owner')
            self.assertIn(str(recovery), '\n'.join(logs.output))
            self.assertEqual(recovery.stat().st_mode & 0o777, 0o700)

    def test_user_and_group_creation_leave_existing_scratch_names_untouched(self):
        with self.working_directory() as root:
            (root / 'user.txt').write_text('unrelated user file')
            (root / 'group.txt').write_text('unrelated group file')
            with prior.module('createusers', ['input.csv']) as mod, patch.object(sdputils, 'run_checked') as run:
                mod.process_servers([('alice', 'a@example.test', 'Alice')])
                self.assertIn('User:\talice\n', run.call_args.kwargs['input'])
            (root / 'groups.txt').write_text('group,team\nalice\ngroup,other\nbob\n')
            with prior.module('creategroups') as mod, patch.object(sdputils, 'run_checked') as run:
                mod.main()
                self.assertEqual([c.kwargs['input'] for c in run.call_args_list],
                                 ['Group:\tteam\nUsers:\n\talice\n', 'Group:\tother\nUsers:\n\tbob\n'])
            self.assertEqual((root / 'user.txt').read_text(), 'unrelated user file')
            self.assertEqual((root / 'group.txt').read_text(), 'unrelated group file')

    def test_group_removal_uses_distinct_files_for_previously_colliding_names(self):
        with self.working_directory() as root, prior.module('removeuserfromgroups', ['alice']) as mod:
            (root / 'a_2fb.group').write_text('unrelated group file')
            def command(args, **kwargs):
                if args[1] == 'groups':
                    return subprocess.CompletedProcess(args, 0, 'a/b\na_2fb\n')
                if args[1:3] == ['group', '-o']:
                    return subprocess.CompletedProcess(args, 0, 'Group: ' + args[-1] + '\nUsers:\n\talice\n\tbob\n')
                return subprocess.CompletedProcess(args, 0)
            with patch.object(sdputils, 'run_checked', side_effect=command) as run:
                mod.automain(['alice'])
            specs = [c.kwargs['input'] for c in run.call_args_list if 'input' in c.kwargs]
            self.assertEqual(len(specs), 2)
            self.assertIn('Group: a/b\n', specs[0])
            self.assertIn('Group: a_2fb\n', specs[1])
            self.assertTrue(all('\talice\n' not in spec and '\tbob\n' in spec for spec in specs))
            self.assertEqual((root / 'a_2fb.group').read_text(), 'unrelated group file')
            self.assertEqual(list(root.glob('.sdp-group-removal-*')), [])

    def test_maintain_users_keeps_existing_report_files(self):
        with self.working_directory() as root, prior.module('maintain_user_from_groups') as mod:
            for name in ['users.txt', 'groups.txt', 'ztagfiles.txt']:
                (root / name).write_text('existing ' + name)
            with patch.object(sdputils, 'run_checked', return_value=subprocess.CompletedProcess([], 0, '')):
                mod.main()
            for name in ['users.txt', 'groups.txt', 'ztagfiles.txt']:
                self.assertEqual((root / name).read_text(), 'existing ' + name)

    def test_failed_label_conversion_keeps_specs_files_owner_and_label_name(self):
        with self.working_directory() as root, prior.module('convert_label_to_autoreload', ['label']) as mod:
            (root / 'owner.txt').write_text('unrelated owner')
            def command(args, **kwargs):
                if '-x' in args:
                    raise subprocess.CalledProcessError(7, args)
                text = '//depot/file#1 - add change 1\n' if args[1] == 'files' else 'Label: label\nOptions: noautoreload\n'
                return subprocess.CompletedProcess(args, 0, text)
            def unlock(label, owner_file):
                Path(owner_file).write_text('original owner')
            with patch.object(mod, 'check_auto', return_value=0), patch.object(sdputils, 'run_checked', side_effect=command), patch.object(mod.p4unlock, 'unlock', side_effect=unlock), patch.object(mod.p4lock, 'lock') as lock, self.assertRaises(subprocess.CalledProcessError):
                mod.convertlabel('label')
            recovery, = root.glob('.sdp-label-recovery-*')
            self.assertEqual((recovery / 'owner.txt').read_text(), 'original owner')
            self.assertEqual((recovery / 'label-name.txt').read_text(), 'label\n')
            self.assertIn('autoreload', (recovery / 'label.labelspec').read_text())
            self.assertEqual((recovery / 'label.labelfiles').read_text(), '//depot/file#1\n')
            self.assertEqual((root / 'owner.txt').read_text(), 'unrelated owner')
            lock.assert_not_called()

    def test_successful_label_conversion_removes_only_private_workspace(self):
        with self.working_directory() as root, prior.module('convert_label_to_autoreload', ['label']) as mod:
            (root / 'owner.txt').write_text('unrelated owner')
            result = subprocess.CompletedProcess([], 0, 'Label: label\nOptions: noautoreload\n')
            with patch.object(mod, 'check_auto', return_value=0), patch.object(sdputils, 'run_checked', return_value=result), patch.object(mod.p4unlock, 'unlock') as unlock, patch.object(mod.p4lock, 'lock') as lock:
                mod.convertlabel('label')
            self.assertEqual(unlock.call_args.kwargs['owner_file'], lock.call_args.kwargs['owner_file'])
            self.assertNotEqual(unlock.call_args.kwargs['owner_file'], 'owner.txt')
            self.assertEqual(list(root.glob('.sdp-label-recovery-*')), [])
            self.assertEqual((root / 'owner.txt').read_text(), 'unrelated owner')


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