feat(Model): support credential-get hook tool in both model and harness

docs/requirements.txt

@@ -1,5 +1,5 @@
 #
-# This file is autogenerated by pip-compile with Python 3.8
+# This file is autogenerated by pip-compile with Python 3.12
 # by the following command:
 #
 #    pip-compile --extra=docs --output-file=docs/requirements.txt pyproject.toml
@@ -29,8 +29,6 @@ idna==3.6
     # via requests
 imagesize==1.4.1
     # via sphinx
-importlib-metadata==7.0.1
-    # via sphinx
 jinja2==3.1.3
     # via sphinx
 lxd-sphinx-extensions==0.0.16
@@ -44,8 +42,6 @@ pygments==2.17.2
     #   furo
     #   sphinx
     #   sphinx-tabs
-pytz==2023.3.post1
-    # via babel
 pyyaml==6.0.1
     # via ops (pyproject.toml)
 requests==2.31.0
@@ -89,5 +85,3 @@ urllib3==2.1.0
     # via requests
 websocket-client==1.7.0
     # via ops (pyproject.toml)
-zipp==3.17.0
-    # via importlib-metadata

ops/__init__.py

@@ -129,6 +129,7 @@
     'BindingMapping',
     'BlockedStatus',
     'CheckInfoMapping',
+    'CloudSpec',
     'ConfigData',
     'Container',
     'ContainerMapping',
@@ -270,6 +271,7 @@
     BindingMapping,
     BlockedStatus,
     CheckInfoMapping,
+    CloudSpec,
     ConfigData,
     Container,
     ContainerMapping,

ops/model.py

@@ -281,6 +281,20 @@ def get_secret(self, *, id: Optional[str] = None, label: Optional[str] = None) -
         content = self._backend.secret_get(id=id, label=label)
         return Secret(self._backend, id=id, label=label, content=content)
 
+    def get_cloud_spec(self) -> 'CloudSpec':
+        """Get cloud credential information.
+
+        Access the cloud credential information and return the cloud specification
+        used by the model.
+
+        Note: It does not work in a "CaaS" model (Container-as-a-Service,
+        i.e., containerized environment, Kubernetes charm).
+
+        Raises:
+            :class:`ModelError`: if called in a "CaaS" model.
+        """
+        return self._backend.credential_get()
+
 
 if typing.TYPE_CHECKING:
     # (entity type, name): instance.
@@ -3507,6 +3521,17 @@ def reboot(self, now: bool = False):
         else:
             self._run("juju-reboot")
 
+    def credential_get(self) -> 'CloudSpec':
+        """Access cloud credentials by running the credential-get hook tool.
+
+        Returns the cloud specification used by the unit's model.
+        """
+        try:
+            result = self._run('credential-get', return_output=True, use_json=True)
+            return CloudSpec.from_dict(typing.cast(Dict[str, Any], result))
+        except ModelError:
+            raise
+
 
 class _ModelBackendValidator:
     """Provides facilities for validating inputs and formatting them for model backends."""
@@ -3596,3 +3621,60 @@ def _ensure_loaded(self):
         self._notice = self._container.get_notice(self.id)
         assert self._notice.type == self.type
         assert self._notice.key == self.key
+
+
+class CloudSpec:
+    """Cloud specification information (metadata) including credentials."""
+
+    def __init__(self,
+                 type: str,
+                 name: str,
+                 region: Optional[str],
+                 endpoint: Optional[str],
+                 is_controller_cloud: Optional[str],
+                 credential: Optional[Dict[str, Any]],
+                 identity_endpoint: Optional[str],
+                 storage_endpoint: Optional[str],
+                 ca_certificates: Optional[List[str]],
+                 skip_tls_verify: Optional[bool],
+                 ):
+        self.type = type
+        self.name = name
+        self.region = region
+        self.endpoint = endpoint
+        self.is_controller_cloud = is_controller_cloud
+        self.credential = credential
+        self.identity_endpoint = identity_endpoint
+        self.storage_endpoint = storage_endpoint
+        self.ca_certificates = ca_certificates
+        self.skip_tls_verify = skip_tls_verify
+
+    @classmethod
+    def from_dict(cls, d: Dict[str, Any]) -> 'CloudSpec':
+        """Create new CloudSpec object from a dict parsed from JSON."""
+        return cls(
+            type=typing.cast(str, d.get('type')),
+            name=typing.cast(str, d.get('name')),
+            region=d.get('region'),
+            endpoint=d.get('endpoint'),
+            is_controller_cloud=d.get('isControllerCloud'),
+            credential=typing.cast(Optional[Dict[str, Any]], d.get('credential')),
+            identity_endpoint=d.get('identityEndpoint'),
+            storage_endpoint=d.get('storageEndpoint'),
+            ca_certificates=d.get('caACertificates'),
+            skip_tls_verify=d.get('skipTLSVerify'),
+        )
+
+    def __repr__(self):
+        return ('CloudSpec('
+                f'type={self.type!r}, '
+                f'name={self.name!r}, '
+                f'region={self.region!r}, '
+                f'endpoint={self.endpoint!r}, '
+                f'is_controller_cloud={self.is_controller_cloud!r}, '
+                f'credential={self.credential!r}, '
+                f'identity_endpoint={self.identity_endpoint!r}, '
+                f'storage_endpoint={self.storage_endpoint!r}, '
+                f'ca_certificates={self.ca_certificates!r}, '
+                f'skip_tls_verify={self.skip_tls_verify!r})'
+                )

ops/testing.py

@@ -1890,6 +1890,42 @@ def run_action(self, action_name: str,
                 output=action_under_test.output)
         return action_under_test.output
 
+    def set_cloud_spec(self, content: Dict[str, str]):
+        """Set cloud specification (metadata) including credentials.
+
+        Call this method before trying to call `Harness.Model.get_cloud_spec`.
+
+        Example usage::
+
+            class TestCharm(unittest.TestCase):
+                def setUp(self):
+                    self.harness = ops.testing.Harness(MyVMCharm)
+                    self.addCleanup(self.harness.cleanup)
+
+                def test_start(self):
+                    self.harness.set_cloud_spec(
+                        {"name": "lxd", "type": "lxd", "endpoint": "https://127.0.0.1:8443"}
+                    )
+                    self.harness.begin_with_initial_hooks()
+                    cloud_spec = self.harness.model.get_cloud_spec()
+                    expected = ops.CloudSpec(
+                        name="lxd",
+                        type="lxd",
+                        region=None,
+                        endpoint="https://127.0.0.1:8443",
+                        is_controller_cloud=None,
+                        credential=None,
+                        identity_endpoint=None,
+                        storage_endpoint=None,
+                        ca_certificates=None,
+                        skip_tls_verify=None,
+                    )
+                    self.assertEqual(repr(cloud_spec), repr(expected))
+                    self.assertEqual(self.harness.model.unit.status, ops.ActiveStatus())
+
+        """
+        self._backend._cloud_spec = model.CloudSpec.from_dict(typing.cast(Dict[str, Any], content))
+
 
 def _get_app_or_unit_name(app_or_unit: AppUnitOrName) -> str:
     """Return name of given application or unit (return strings directly)."""
@@ -2105,6 +2141,8 @@ def __init__(self, unit_name: str, meta: charm.CharmMeta, config: '_RawConfig'):
         self._networks: Dict[Tuple[Optional[str], Optional[int]], _NetworkDict] = {}
         self._reboot_count = 0
         self._running_action: Optional[_RunningAction] = None
+        # for `Model.get_cloud_spec`, initialized to None
+        self._cloud_spec: Optional[model.CloudSpec] = None
 
     def _validate_relation_access(self, relation_name: str, relations: List[model.Relation]):
         """Ensures that the named relation exists/has been added.
@@ -2678,6 +2716,12 @@ def reboot(self, now: bool = False):
         # to handle everything after the exit.
         raise SystemExit()
 
+    def credential_get(self) -> model.CloudSpec:
+        if not self._cloud_spec:
+            raise model.ModelError(
+                'ERROR cloud spec is empty, set it with Harness.set_cloud_spec first')
+        return self._cloud_spec
+
 
 @_copy_docstrings(pebble.ExecProcess)
 class _TestingExecProcess:

test/test_model.py

@@ -3791,5 +3791,90 @@ def test_repr(self):
         )
 
 
+class TestCloudSpec(unittest.TestCase):
+    def setUp(self) -> None:
+        self.credential = {
+            'auth-type': 'certificate',
+            'attrs': {
+                'client-cert': 'foo',
+                'client-key': 'bar',
+                'server-cert': 'baz'
+            },
+        }
+
+    def test_init(self):
+        cloud_spec = ops.CloudSpec(
+            type="localhost",
+            name="lxd",
+            region='localhost',
+            endpoint='https://127.0.0.1:8443',
+            is_controller_cloud=None,
+            credential=self.credential,
+            identity_endpoint='http://127.0.0.1',
+            storage_endpoint='http://127.0.0.1',
+            ca_certificates=['foo', 'bar'],
+            skip_tls_verify=True,
+        )
+
+        self.assertEqual(cloud_spec.type, 'localhost')
+        self.assertEqual(cloud_spec.name, 'lxd')
+        self.assertEqual(cloud_spec.region, 'localhost')
+        self.assertEqual(cloud_spec.endpoint, 'https://127.0.0.1:8443')
+        self.assertEqual(cloud_spec.credential, self.credential)
+        self.assertEqual(cloud_spec.is_controller_cloud, None)
+        self.assertEqual(cloud_spec.identity_endpoint, 'http://127.0.0.1')
+        self.assertEqual(cloud_spec.storage_endpoint, 'http://127.0.0.1')
+        self.assertEqual(cloud_spec.ca_certificates, ['foo', 'bar'])
+        self.assertEqual(cloud_spec.skip_tls_verify, True)
+
+        self.assertTrue(repr(cloud_spec).startswith('CloudSpec('))
+        self.assertTrue(repr(cloud_spec).endswith(')'))
+
+    def test_from_dict(self):
+        cloud_spec = ops.CloudSpec.from_dict(
+            {
+                'type': 'lxd',
+                'name': 'localhost',
+                'region': 'localhost',
+                'endpoint': 'https://10.76.251.1:8443',
+                'isControllerCloud': None,
+                'credential': self.credential,
+                'identityEndpoint': None,
+                'storageEndpoint': None,
+                'caACertificates': None,
+                'skipTLSVerify': None
+            }
+        )
+        self.assertEqual(cloud_spec.type, 'lxd')
+        self.assertEqual(cloud_spec.name, 'localhost')
+        self.assertEqual(cloud_spec.region, 'localhost')
+        self.assertEqual(cloud_spec.endpoint, 'https://10.76.251.1:8443')
+        self.assertEqual(cloud_spec.is_controller_cloud, None)
+        self.assertEqual(cloud_spec.identity_endpoint, None)
+        self.assertEqual(cloud_spec.storage_endpoint, None)
+        self.assertEqual(cloud_spec.ca_certificates, None)
+        self.assertEqual(cloud_spec.skip_tls_verify, None)
+        self.assertEqual(cloud_spec.credential, self.credential)
+
+
+class TestGetCloudSpec(unittest.TestCase):
+    def setUp(self):
+        self.model = ops.Model(ops.CharmMeta(), _ModelBackend('myapp/0'))
+
+    def test_get_cloud_spec(self):
+        fake_script(self, 'credential-get', """echo '{"type": "lxd", "name": "localhost"}'""")
+        cloud_spec = self.model.get_cloud_spec()
+        self.assertEqual(cloud_spec.type, 'lxd')
+        self.assertEqual(cloud_spec.name, 'localhost')
+        self.assertEqual(fake_script_calls(self, clear=True),
+                         [['credential-get', '--format=json']])
+
+    @patch("ops.model._ModelBackend.credential_get")
+    def test_get_cloud_spec_error(self, credential_get: MagicMock):
+        credential_get.side_effect = ops.ModelError
+        with self.assertRaises(ops.ModelError):
+            self.model.get_cloud_spec()
+
+
 if __name__ == "__main__":
     unittest.main()

test/test_testing.py

@@ -5735,3 +5735,26 @@ def test_get_notices(self):
 class TestNotices(unittest.TestCase, _TestingPebbleClientMixin, PebbleNoticesMixin):
     def setUp(self):
         self.client = self.get_testing_client()
+
+
+class TestCloudSpec(unittest.TestCase):
+    def test_get_set_cloud_spec(self):
+        harness = ops.testing.Harness(EventRecorder, meta='name: myapp')
+        self.addCleanup(harness.cleanup)
+        cloud_spec_dict = {
+            "name": "localhost",
+            "type": "lxd",
+            "endpoint": "https://127.0.0.1:8443"
+        }
+        harness.set_cloud_spec(cloud_spec_dict)
+        harness.begin()
+        result = harness.model.get_cloud_spec()
+        expected = ops.model.CloudSpec.from_dict(cloud_spec_dict)
+        self.assertEqual(repr(result), repr(expected))
+
+    def test_get_cloud_spec_without_set_error(self):
+        harness = ops.testing.Harness(EventRecorder, meta='name: myapp')
+        self.addCleanup(harness.cleanup)
+        harness.begin()
+        with self.assertRaises(ops.ModelError):
+            harness.model.get_cloud_spec()