modules/libjuju/tests/unit/test_controller.py

   1 import asynctest
   2 import mock
   3 from pathlib import Path
   4 from tempfile import NamedTemporaryFile
   5
   6 from juju.controller import Controller
   7 from juju.client import client
   8
   9 from .. import base
  10
  11
  12 class TestControllerConnect(asynctest.TestCase):
  13     @asynctest.patch('juju.client.connector.Connector.connect_controller')
  14     async def test_no_args(self, mock_connect_controller):
  15         c = Controller()
  16         await c.connect()
  17         mock_connect_controller.assert_called_once_with(None)
  18
  19     @asynctest.patch('juju.client.connector.Connector.connect_controller')
  20     async def test_with_controller_name(self, mock_connect_controller):
  21         c = Controller()
  22         await c.connect(controller_name='foo')
  23         mock_connect_controller.assert_called_once_with('foo')
  24
  25     @asynctest.patch('juju.client.connector.Connector.connect')
  26     async def test_with_endpoint_and_no_auth(self, mock_connect):
  27         c = Controller()
  28         with self.assertRaises(TypeError):
  29             await c.connect(endpoint='0.1.2.3:4566')
  30         self.assertEqual(mock_connect.call_count, 0)
  31
  32     @asynctest.patch('juju.client.connector.Connector.connect')
  33     async def test_with_endpoint_and_userpass(self, mock_connect):
  34         c = Controller()
  35         with self.assertRaises(TypeError):
  36             await c.connect(endpoint='0.1.2.3:4566', username='dummy')
  37         await c.connect(endpoint='0.1.2.3:4566',
  38                         username='user',
  39                         password='pass')
  40         mock_connect.assert_called_once_with(endpoint='0.1.2.3:4566',
  41                                              username='user',
  42                                              password='pass')
  43
  44     @asynctest.patch('juju.client.connector.Connector.connect')
  45     async def test_with_endpoint_and_bakery_client(self, mock_connect):
  46         c = Controller()
  47         await c.connect(endpoint='0.1.2.3:4566', bakery_client='bakery')
  48         mock_connect.assert_called_once_with(endpoint='0.1.2.3:4566',
  49                                              bakery_client='bakery')
  50
  51     @asynctest.patch('juju.client.connector.Connector.connect')
  52     async def test_with_endpoint_and_macaroons(self, mock_connect):
  53         c = Controller()
  54         await c.connect(endpoint='0.1.2.3:4566',
  55                         macaroons=['macaroon'])
  56         mock_connect.assert_called_with(endpoint='0.1.2.3:4566',
  57                                         macaroons=['macaroon'])
  58         await c.connect(endpoint='0.1.2.3:4566',
  59                         bakery_client='bakery',
  60                         macaroons=['macaroon'])
  61         mock_connect.assert_called_with(endpoint='0.1.2.3:4566',
  62                                         bakery_client='bakery',
  63                                         macaroons=['macaroon'])
  64
  65     @asynctest.patch('juju.client.connector.Connector.connect_controller')
  66     @asynctest.patch('juju.client.connector.Connector.connect')
  67     async def test_with_posargs(self, mock_connect, mock_connect_controller):
  68         c = Controller()
  69         await c.connect('foo')
  70         mock_connect_controller.assert_called_once_with('foo')
  71         with self.assertRaises(TypeError):
  72             await c.connect('endpoint', 'user')
  73         await c.connect('endpoint', 'user', 'pass')
  74         mock_connect.assert_called_once_with(endpoint='endpoint',
  75                                              username='user',
  76                                              password='pass')
  77         await c.connect('endpoint', 'user', 'pass', 'cacert', 'bakery',
  78                         'macaroons', 'loop', 'max_frame_size')
  79         mock_connect.assert_called_with(endpoint='endpoint',
  80                                         username='user',
  81                                         password='pass',
  82                                         cacert='cacert',
  83                                         bakery_client='bakery',
  84                                         macaroons='macaroons',
  85                                         loop='loop',
  86                                         max_frame_size='max_frame_size')
  87
  88     @asynctest.patch('juju.client.client.CloudFacade')
  89     async def test_file_cred_v2(self, mock_cf):
  90         with NamedTemporaryFile() as tempfile:
  91             tempfile.close()
  92             temppath = Path(tempfile.name)
  93             temppath.write_text('cred-test')
  94             cred = client.CloudCredential(auth_type='jsonfile',
  95                                           attrs={'file': tempfile.name})
  96             jujudata = mock.MagicMock()
  97             c = Controller(jujudata=jujudata)
  98             c._connector = base.AsyncMock()
  99             up_creds = base.AsyncMock()
 100             cloud_facade = mock_cf.from_connection()
 101             cloud_facade.version = 2
 102             cloud_facade.UpdateCredentials = up_creds
 103             await c.add_credential(
 104                 name='name',
 105                 credential=cred,
 106                 cloud='cloud',
 107                 owner='owner',
 108             )
 109             assert up_creds.called
 110             new_cred = up_creds.call_args[0][0][0].credential
 111             assert cred.attrs['file'] == tempfile.name
 112             assert new_cred.attrs['file'] == 'cred-test'
 113
 114     @asynctest.patch('juju.client.client.CloudFacade')
 115     async def test_file_cred_v3(self, mock_cf):
 116         with NamedTemporaryFile() as tempfile:
 117             tempfile.close()
 118             temppath = Path(tempfile.name)
 119             temppath.write_text('cred-test')
 120             cred = client.CloudCredential(auth_type='jsonfile',
 121                                           attrs={'file': tempfile.name})
 122             jujudata = mock.MagicMock()
 123             c = Controller(jujudata=jujudata)
 124             c._connector = base.AsyncMock()
 125             up_creds = base.AsyncMock()
 126             cloud_facade = mock_cf.from_connection()
 127             cloud_facade.version = 3
 128             cloud_facade.UpdateCredentialsCheckModels = up_creds
 129             await c.add_credential(
 130                 name='name',
 131                 credential=cred,
 132                 cloud='cloud',
 133                 owner='owner',
 134                 force=True,
 135             )
 136             assert up_creds.called
 137             assert up_creds.call_args[1]['force']
 138             new_cred = up_creds.call_args[1]['credentials'][0].credential
 139             assert cred.attrs['file'] == tempfile.name
 140             assert new_cred.attrs['file'] == 'cred-test'