import copy
from concurrent.futures import ThreadPoolExecutor
from functools import partial
from pathlib import Path
import tempfile
import unittest

from network_contract import ContractFailure, OWNER
from network_prepare import prepare
from network_runtime import NetworkFailure, allocation_lock
from network_test_fixture import Host, config, network


class PreparationTests(unittest.TestCase):
    def test_new_network_is_small_owned_and_replayed_without_mutation(self):
        host = Host()
        baseline = copy.deepcopy(host.networks[0])
        result = prepare(config('first'), 'first', **host.arguments())
        self.assertTrue(result['ok'])
        self.assertEqual(result['networks'][0]['subnets'], ['172.28.1.0/24'])
        self.assertEqual(host.networks[0], baseline)
        self.assertIn(OWNER, host.networks[-1]['Labels'])
        self.assertNotIn('com.docker.compose.config-hash', host.networks[-1]['Labels'])
        replay = prepare(config('first'), 'first', **host.arguments())
        self.assertEqual(replay['networks'][0]['id'], result['networks'][0]['id'])
        self.assertEqual(len(host.calls), 1)

    def test_existing_network_is_not_resized_even_with_exhausted_unknown_pools(self):
        host = Host()
        host.networks.append(network('existing', '172.29.0.0/16'))
        host.pool_failure = 'pool_provenance_unverified'
        result = prepare(config('existing'), 'existing', **host.arguments())
        self.assertEqual(result['networks'][0]['subnets'], ['172.29.0.0/16'])
        self.assertEqual(host.calls, [])

    def test_lost_create_reply_is_read_back_without_second_create(self):
        host = Host()
        host.lose_reply = True
        result = prepare(config('first'), 'first', **host.arguments())
        self.assertTrue(result['ok'])
        self.assertEqual(len(host.calls), 1)

    def test_foreign_or_ambiguous_names_refuse_before_create(self):
        for duplicate in (False, True):
            host = Host()
            host.networks[0]['Name'] = 'first_default'
            if duplicate:
                host.networks.append(network('first', '172.28.1.0/24'))
            with self.assertRaises(ContractFailure):
                prepare(config('first'), 'first', **host.arguments())
            self.assertEqual(host.calls, [])

    def test_changed_allocated_network_contract_refuses_without_replacement(self):
        host = Host()
        prepare(config('first'), 'first', **host.arguments())
        with self.assertRaisesRegex(ContractFailure, 'network_contract_changed'):
            prepare(config('first', {'internal': True}), 'first', **host.arguments())
        self.assertEqual(len(host.calls), 1)
        self.assertEqual(len(host.networks), 2)

    def test_authored_network_contracts_are_left_to_compose(self):
        cases = ({'name': 'custom'}, {'external': True}, {'ipam': {'config': [{'subnet': '10.20.0.0/24'}]}},
                 {'driver': 'overlay'}, {'enable_ipv6': True}, {'driver_opts': {'fixture': 'value'}})
        for value in cases:
            host = Host()
            result = prepare(config('first', value), 'first', **host.arguments())
            self.assertTrue(result['ok'])
            self.assertEqual(host.calls, [])

    def test_internal_and_authored_labels_survive_creation(self):
        host = Host()
        prepare(config('first', {'internal': True, 'labels': {'example.test': 'retained'}}),
                'first', **host.arguments())
        self.assertTrue(host.networks[-1]['Internal'])
        self.assertEqual(host.networks[-1]['Labels']['example.test'], 'retained')

    def test_concurrent_projects_use_distinct_subnets_under_real_host_lock(self):
        host = Host()
        with tempfile.TemporaryDirectory() as root:
            lock = partial(allocation_lock, path=Path(root) / 'allocation.lock')
            with ThreadPoolExecutor(max_workers=2) as executor:
                futures = [executor.submit(prepare, config(app), app, **host.arguments(lock))
                           for app in ('first', 'second')]
                values = [future.result()['networks'][0]['subnets'][0] for future in futures]
        self.assertEqual(set(values), {'172.28.1.0/24', '172.28.2.0/24'})
        self.assertEqual(len(host.calls), 2)

    def test_reviewed_compose_majors_allocate_and_others_refuse_before_create(self):
        for version in ('2.39.4', 'v2.40.3', '5.0.0', '5.3.1', 'v5.5.1'):
            host = Host()
            host.compose_version = version
            result = prepare(config('first'), 'first', **host.arguments())
            self.assertTrue(result['ok'])
            self.assertEqual(result['compose_version'], version)
            self.assertEqual(host.networks[-1]['Labels']['com.docker.compose.version'], version)
            self.assertNotIn('com.docker.compose.config-hash', host.networks[-1]['Labels'])
            self.assertEqual(len(host.calls), 1)
        for version in ('1.29.2', '3.0.0', '4.1.0', '6.0.0', '5.3', '25.0.0'):
            host = Host()
            host.compose_version = version
            with self.assertRaisesRegex(NetworkFailure, 'compose_version_unsupported'):
                prepare(config('first'), 'first', **host.arguments())
            self.assertEqual(host.calls, [])
            host.networks.append(network('existing', '172.29.0.0/16'))
            preserved = prepare(config('existing'), 'existing', **host.arguments())
            self.assertEqual(preserved['networks'][0]['state'], 'preserved')
            self.assertEqual(host.calls, [])

    def test_full_effective_pool_and_unknown_provenance_refuse_expansion(self):
        for unknown in (False, True):
            host = Host()
            host.networks = [network('unrelated', '172.28.0.0/22')]
            host.pool_failure = 'pool_provenance_unverified' if unknown else ''
            with self.assertRaises(NetworkFailure):
                prepare(config('first'), 'first', **host.arguments())
            self.assertEqual(host.calls, [])


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