1import pytest
2
3from ansible.module_utils.source_control.bitbucket import BitbucketHelper
4from ansible.modules.source_control.bitbucket import bitbucket_pipeline_known_host
5from ansible.modules.source_control.bitbucket.bitbucket_pipeline_known_host import HAS_PARAMIKO
6from units.compat import unittest
7from units.compat.mock import patch
8from units.modules.utils import AnsibleExitJson, ModuleTestCase, set_module_args
9
10
11class TestBucketPipelineKnownHostModule(ModuleTestCase):
12    def setUp(self):
13        super(TestBucketPipelineKnownHostModule, self).setUp()
14        self.module = bitbucket_pipeline_known_host
15
16    @pytest.mark.skipif(not HAS_PARAMIKO, reason='paramiko must be installed to test key creation')
17    @patch.object(BitbucketHelper, 'fetch_access_token', return_value='token')
18    @patch.object(bitbucket_pipeline_known_host, 'get_existing_known_host', return_value=None)
19    def test_create_known_host(self, *args):
20        with patch.object(self.module, 'create_known_host') as create_known_host_mock:
21            with self.assertRaises(AnsibleExitJson) as exec_info:
22                set_module_args({
23                    'client_id': 'ABC',
24                    'client_secret': 'XXX',
25                    'username': 'name',
26                    'repository': 'repo',
27                    'name': 'bitbucket.org',
28                    'state': 'present',
29                })
30                self.module.main()
31
32            self.assertEqual(create_known_host_mock.call_count, 1)
33            self.assertEqual(exec_info.exception.args[0]['changed'], True)
34
35    @patch.object(BitbucketHelper, 'fetch_access_token', return_value='token')
36    @patch.object(BitbucketHelper, 'request', return_value=(dict(status=201), dict()))
37    @patch.object(bitbucket_pipeline_known_host, 'get_existing_known_host', return_value=None)
38    def test_create_known_host_with_key(self, *args):
39        with patch.object(self.module, 'get_host_key') as get_host_key_mock:
40            with self.assertRaises(AnsibleExitJson) as exec_info:
41                set_module_args({
42                    'client_id': 'ABC',
43                    'client_secret': 'XXX',
44                    'username': 'name',
45                    'repository': 'repo',
46                    'name': 'bitbucket.org',
47                    'key': 'ssh-rsa public',
48                    'state': 'present',
49                })
50                self.module.main()
51
52            self.assertEqual(get_host_key_mock.call_count, 0)
53            self.assertEqual(exec_info.exception.args[0]['changed'], True)
54
55    @pytest.mark.skipif(not HAS_PARAMIKO, reason='paramiko must be installed to test key creation')
56    @patch.object(BitbucketHelper, 'fetch_access_token', return_value='token')
57    @patch.object(bitbucket_pipeline_known_host, 'get_existing_known_host', return_value={
58        'type': 'pipeline_known_host',
59        'uuid': '{21cc0590-bebe-4fae-8baf-03722704119a7}',
60        'hostname': 'bitbucket.org',
61        'public_key': {
62            'type': 'pipeline_ssh_public_key',
63            'md5_fingerprint': 'md5:97:8c:1b:f2:6f:14:6b:4b:3b:ec:aa:46:46:74:7c:40',
64            'sha256_fingerprint': 'SHA256:zzXQOXSFBEiUtuE8AikoYKwbHaxvSc0ojez9YXaGp1A',
65            'key_type': 'ssh-rsa',
66            'key': 'AAAAB3NzaC1yc2EAAAABIwAAAQEAubiN81eDcafrgMeLzaFPsw2kN...seeFVBoGqzHM9yXw=='
67        }
68    })
69    def test_dont_create_same_value(self, *args):
70        with patch.object(self.module, 'create_known_host') as create_known_host_mock:
71            with self.assertRaises(AnsibleExitJson) as exec_info:
72                set_module_args({
73                    'client_id': 'ABC',
74                    'client_secret': 'XXX',
75                    'username': 'name',
76                    'repository': 'repo',
77                    'name': 'bitbucket.org',
78                    'state': 'present',
79                })
80                self.module.main()
81
82            self.assertEqual(create_known_host_mock.call_count, 0)
83            self.assertEqual(exec_info.exception.args[0]['changed'], False)
84
85    @pytest.mark.skipif(not HAS_PARAMIKO, reason='paramiko must be installed to test key creation')
86    @patch.object(BitbucketHelper, 'fetch_access_token', return_value='token')
87    @patch.object(bitbucket_pipeline_known_host, 'get_existing_known_host', return_value=None)
88    def test_create_known_host_check_mode(self, *args):
89        with patch.object(self.module, 'create_known_host') as create_known_host_mock:
90            with self.assertRaises(AnsibleExitJson) as exec_info:
91                set_module_args({
92                    'client_id': 'ABC',
93                    'client_secret': 'XXX',
94                    'username': 'name',
95                    'repository': 'repo',
96                    'name': 'bitbucket.org',
97                    'state': 'present',
98                    '_ansible_check_mode': True,
99                })
100                self.module.main()
101
102            self.assertEqual(create_known_host_mock.call_count, 0)
103            self.assertEqual(exec_info.exception.args[0]['changed'], True)
104
105    @pytest.mark.skipif(not HAS_PARAMIKO, reason='paramiko must be installed to test key creation')
106    @patch.object(BitbucketHelper, 'fetch_access_token', return_value='token')
107    @patch.object(bitbucket_pipeline_known_host, 'get_existing_known_host', return_value={
108        'type': 'pipeline_known_host',
109        'uuid': '{21cc0590-bebe-4fae-8baf-03722704119a7}',
110        'hostname': 'bitbucket.org',
111        'public_key': {
112            'type': 'pipeline_ssh_public_key',
113            'md5_fingerprint': 'md5:97:8c:1b:f2:6f:14:6b:4b:3b:ec:aa:46:46:74:7c:40',
114            'sha256_fingerprint': 'SHA256:zzXQOXSFBEiUtuE8AikoYKwbHaxvSc0ojez9YXaGp1A',
115            'key_type': 'ssh-rsa',
116            'key': 'AAAAB3NzaC1yc2EAAAABIwAAAQEAubiN81eDcafrgMeLzaFPsw2kN...seeFVBoGqzHM9yXw=='
117        }
118    })
119    def test_delete_known_host(self, *args):
120        with patch.object(self.module, 'delete_known_host') as delete_known_host_mock:
121            with self.assertRaises(AnsibleExitJson) as exec_info:
122                set_module_args({
123                    'client_id': 'ABC',
124                    'client_secret': 'XXX',
125                    'username': 'name',
126                    'repository': 'repo',
127                    'name': 'bitbucket.org',
128                    'state': 'absent',
129                })
130                self.module.main()
131
132            self.assertEqual(delete_known_host_mock.call_count, 1)
133            self.assertEqual(exec_info.exception.args[0]['changed'], True)
134
135    @pytest.mark.skipif(not HAS_PARAMIKO, reason='paramiko must be installed to test key creation')
136    @patch.object(BitbucketHelper, 'fetch_access_token', return_value='token')
137    @patch.object(bitbucket_pipeline_known_host, 'get_existing_known_host', return_value=None)
138    def test_delete_absent_known_host(self, *args):
139        with patch.object(self.module, 'delete_known_host') as delete_known_host_mock:
140            with self.assertRaises(AnsibleExitJson) as exec_info:
141                set_module_args({
142                    'client_id': 'ABC',
143                    'client_secret': 'XXX',
144                    'username': 'name',
145                    'repository': 'repo',
146                    'name': 'bitbucket.org',
147                    'state': 'absent',
148                })
149                self.module.main()
150
151            self.assertEqual(delete_known_host_mock.call_count, 0)
152            self.assertEqual(exec_info.exception.args[0]['changed'], False)
153
154    @pytest.mark.skipif(not HAS_PARAMIKO, reason='paramiko must be installed to test key creation')
155    @patch.object(BitbucketHelper, 'fetch_access_token', return_value='token')
156    @patch.object(bitbucket_pipeline_known_host, 'get_existing_known_host', return_value={
157        'type': 'pipeline_known_host',
158        'uuid': '{21cc0590-bebe-4fae-8baf-03722704119a7}',
159        'hostname': 'bitbucket.org',
160        'public_key': {
161            'type': 'pipeline_ssh_public_key',
162            'md5_fingerprint': 'md5:97:8c:1b:f2:6f:14:6b:4b:3b:ec:aa:46:46:74:7c:40',
163            'sha256_fingerprint': 'SHA256:zzXQOXSFBEiUtuE8AikoYKwbHaxvSc0ojez9YXaGp1A',
164            'key_type': 'ssh-rsa',
165            'key': 'AAAAB3NzaC1yc2EAAAABIwAAAQEAubiN81eDcafrgMeLzaFPsw2kN...seeFVBoGqzHM9yXw=='
166        }
167    })
168    def test_delete_known_host_check_mode(self, *args):
169        with patch.object(self.module, 'delete_known_host') as delete_known_host_mock:
170            with self.assertRaises(AnsibleExitJson) as exec_info:
171                set_module_args({
172                    'client_id': 'ABC',
173                    'client_secret': 'XXX',
174                    'username': 'name',
175                    'repository': 'repo',
176                    'name': 'bitbucket.org',
177                    'state': 'absent',
178                    '_ansible_check_mode': True,
179                })
180                self.module.main()
181
182            self.assertEqual(delete_known_host_mock.call_count, 0)
183            self.assertEqual(exec_info.exception.args[0]['changed'], True)
184
185
186if __name__ == '__main__':
187    unittest.main()
188