from contextlib import nullcontext
import unittest
from unittest.mock import Mock

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


class NoAllocationTests(unittest.TestCase):
    def test_existing_and_authored_paths_never_open_an_unavailable_allocation_lock(self):
        host = Host()
        host.networks.append(network('existing', '172.29.0.0/16'))
        prepare(config('created'), 'created', **host.arguments())
        lock = Mock(side_effect=NetworkFailure('allocation_lock_unavailable'))
        for project, settings in (('existing', {}), ('created', {}), ('external', {'external': True}),
                                  ('custom', {'name': 'custom-name'}),
                                  ('authored', {'ipam': {'config': [{'subnet': '10.23.0.0/24'}]}})):
            with self.subTest(project=project):
                result = prepare(config(project, settings), project, **dict(host.arguments(), lock=lock))
                self.assertTrue(result['ok'])
        lock.assert_not_called()
        self.assertEqual(len(host.calls), 1)

    def test_missing_eligible_network_still_requires_the_allocation_lock(self):
        host = Host()
        lock = Mock(side_effect=NetworkFailure('allocation_lock_unavailable'))
        with self.assertRaisesRegex(NetworkFailure, 'allocation_lock_unavailable'):
            prepare(config('missing'), 'missing', **dict(host.arguments(), lock=lock))
        lock.assert_called_once_with()
        self.assertEqual(host.calls, [])

    def test_network_appearing_before_lock_is_read_again_and_preserved_or_refused(self):
        for foreign in (False, True):
            row = network('first', '172.28.1.0/24')
            if foreign:
                row['Labels']['com.docker.compose.project'] = 'another'
            owner = Mock(side_effect=[{'networks': []}, {'networks': [row]}])
            command = Mock(side_effect=AssertionError('No create is authorized by the unlocked snapshot'))
            reader = Mock(side_effect=AssertionError('No allocation inventory is needed after the locked read'))
            arguments = {'owner_reader': owner, 'reader': reader, 'command': command, 'lock': nullcontext}
            if foreign:
                with self.assertRaisesRegex(ContractFailure, 'network_ownership_mismatch'):
                    prepare(config('first'), 'first', **arguments)
            else:
                result = prepare(config('first'), 'first', **arguments)
                self.assertEqual(result['networks'][0]['id'], row['Id'])
            self.assertEqual(owner.call_count, 2)
            command.assert_not_called()
            reader.assert_not_called()

    def test_unlocked_replay_still_verifies_the_owned_network_contract(self):
        host = Host()
        prepare(config('first'), 'first', **host.arguments())
        lock = Mock(side_effect=NetworkFailure('allocation_lock_unavailable'))
        with self.assertRaisesRegex(ContractFailure, 'network_contract_changed'):
            prepare(config('first', {'internal': True}), 'first', **dict(host.arguments(), lock=lock))
        lock.assert_not_called()
        self.assertEqual(len(host.calls), 1)


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