Skip to content

Commit

Permalink
Merge pull request #119 from IronCore864/cloud-spec
Browse files Browse the repository at this point in the history
feat: add support for cloud spec
  • Loading branch information
PietroPasotti authored May 29, 2024
2 parents e758e78 + c56e267 commit 1df85e1
Show file tree
Hide file tree
Showing 7 changed files with 241 additions and 1 deletion.
40 changes: 40 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -948,6 +948,46 @@ assert out.model.name == "my-model"
assert out.model.uuid == state_in.model.uuid
```

## CloudSpec

You can set CloudSpec information in the state (only `type` and `name` are required).

Example:

```python
import scenario

state = scenario.State(
cloud_spec=scenario.CloudSpec(
type="lxd",
name="localhost",
endpoint="https://127.0.0.1:8443",
credential=scenario.CloudCredential(
auth_type="clientcertificate",
attributes={
"client-cert": "foo",
"client-key": "bar",
"server-cert": "baz",
},
),
),
model=scenario.Model(name="my-vm-model", type="lxd"),
)
```

Then you can access it by `Model.get_cloud_spec()`:

```python
# charm.py
class MyVMCharm(ops.CharmBase):
def __init__(self, framework: ops.Framework):
super().__init__(framework)
framework.observe(self.on.start, self._on_start)

def _on_start(self, event: ops.StartEvent):
self.cloud_spec = self.model.get_cloud_spec()
```

# Actions

An action is a special sort of event, even though `ops` handles them almost identically.
Expand Down
4 changes: 4 additions & 0 deletions scenario/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,8 @@
Action,
Address,
BindAddress,
CloudCredential,
CloudSpec,
Container,
DeferredEvent,
Event,
Expand All @@ -29,6 +31,8 @@
__all__ = [
"Action",
"ActionOutput",
"CloudCredential",
"CloudSpec",
"Context",
"deferred",
"StateValidationError",
Expand Down
21 changes: 21 additions & 0 deletions scenario/consistency_checker.py
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,7 @@ def check_consistency(
check_storages_consistency,
check_relation_consistency,
check_network_consistency,
check_cloudspec_consistency,
):
results = check(
state=state,
Expand Down Expand Up @@ -565,3 +566,23 @@ def check_containers_consistency(
errors.append(f"Duplicate container name(s): {dupes}.")

return Results(errors, [])


def check_cloudspec_consistency(
*,
state: "State",
event: "Event",
charm_spec: "_CharmSpec",
**_kwargs, # noqa: U101
) -> Results:
"""Check that Kubernetes charms/models don't have `state.cloud_spec`."""

errors = []
warnings = []

if state.model.type == "kubernetes" and state.cloud_spec:
errors.append(
"CloudSpec is only available for machine charms, not Kubernetes charms. Tell Scenario to simulate a machine substrate with: `scenario.State(..., model=scenario.Model(type='lxd'))`.",
)

return Results(errors, warnings)
9 changes: 8 additions & 1 deletion scenario/mocking.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
cast,
)

from ops import JujuVersion, pebble
from ops import CloudSpec, JujuVersion, pebble
from ops.model import ModelError, RelationNotFoundError
from ops.model import Secret as Secret_Ops # lol
from ops.model import (
Expand Down Expand Up @@ -629,6 +629,13 @@ def resource_get(self, resource_name: str) -> str:
f"resource {resource_name} not found in State. please pass it.",
)

def credential_get(self) -> CloudSpec:
if not self._state.cloud_spec:
raise ModelError(
"ERROR cloud spec is empty, initialise it with `scenario.State(cloud_spec=scenario.CloudSpec(...))`",
)
return self._state.cloud_spec._to_ops()


class _MockPebbleClient(_TestingPebbleClient):
def __init__(
Expand Down
74 changes: 74 additions & 0 deletions scenario/state.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
)
from uuid import uuid4

import ops
import yaml
from ops import pebble
from ops.charm import CharmBase, CharmEvents
Expand Down Expand Up @@ -140,6 +141,77 @@ def copy(self) -> "Self":
return copy.deepcopy(self)


@dataclasses.dataclass(frozen=True)
class CloudCredential:
auth_type: str
"""Authentication type."""

attributes: Dict[str, str] = dataclasses.field(default_factory=dict)
"""A dictionary containing cloud credentials.
For example, for AWS, it contains `access-key` and `secret-key`;
for Azure, `application-id`, `application-password` and `subscription-id`
can be found here.
"""

redacted: List[str] = dataclasses.field(default_factory=list)
"""A list of redacted generic cloud API secrets."""

def _to_ops(self) -> ops.CloudCredential:
return ops.CloudCredential(
auth_type=self.auth_type,
attributes=self.attributes,
redacted=self.redacted,
)


@dataclasses.dataclass(frozen=True)
class CloudSpec:
type: str
"""Type of the cloud."""

name: str = "localhost"
"""Juju cloud name."""

region: Optional[str] = None
"""Region of the cloud."""

endpoint: Optional[str] = None
"""Endpoint of the cloud."""

identity_endpoint: Optional[str] = None
"""Identity endpoint of the cloud."""

storage_endpoint: Optional[str] = None
"""Storage endpoint of the cloud."""

credential: Optional[CloudCredential] = None
"""Cloud credentials with key-value attributes."""

ca_certificates: List[str] = dataclasses.field(default_factory=list)
"""A list of CA certificates."""

skip_tls_verify: bool = False
"""Whether to skip TLS verfication."""

is_controller_cloud: bool = False
"""If this is the cloud used by the controller."""

def _to_ops(self) -> ops.CloudSpec:
return ops.CloudSpec(
type=self.type,
name=self.name,
region=self.region,
endpoint=self.endpoint,
identity_endpoint=self.identity_endpoint,
storage_endpoint=self.storage_endpoint,
credential=self.credential._to_ops() if self.credential else None,
ca_certificates=self.ca_certificates,
skip_tls_verify=self.skip_tls_verify,
is_controller_cloud=self.is_controller_cloud,
)


@dataclasses.dataclass(frozen=True)
class Secret(_DCBase):
id: str
Expand Down Expand Up @@ -919,6 +991,8 @@ class State(_DCBase):
"""Status of the unit."""
workload_version: str = ""
"""Workload version."""
cloud_spec: Optional[CloudSpec] = None
"""Cloud specification information (metadata) including credentials."""

def __post_init__(self):
for name in ["app_status", "unit_status"]:
Expand Down
36 changes: 36 additions & 0 deletions tests/test_consistency_checker.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,14 @@
import pytest
from ops.charm import CharmBase

from scenario import Model
from scenario.consistency_checker import check_consistency
from scenario.runtime import InconsistentScenarioError
from scenario.state import (
RELATION_EVENTS_SUFFIX,
Action,
CloudCredential,
CloudSpec,
Container,
Event,
Network,
Expand Down Expand Up @@ -562,3 +565,36 @@ def test_networks_consistency():
},
),
)


def test_cloudspec_consistency():
cloud_spec = CloudSpec(
type="lxd",
endpoint="https://127.0.0.1:8443",
credential=CloudCredential(
auth_type="clientcertificate",
attributes={
"client-cert": "foo",
"client-key": "bar",
"server-cert": "baz",
},
),
)

assert_consistent(
State(cloud_spec=cloud_spec, model=Model(name="lxd-model", type="lxd")),
Event("start"),
_CharmSpec(
MyCharm,
meta={"name": "MyVMCharm"},
),
)

assert_inconsistent(
State(cloud_spec=cloud_spec, model=Model(name="k8s-model", type="kubernetes")),
Event("start"),
_CharmSpec(
MyCharm,
meta={"name": "MyK8sCharm"},
),
)
58 changes: 58 additions & 0 deletions tests/test_e2e/test_cloud_spec.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,58 @@
import ops
import pytest

import scenario


class MyCharm(ops.CharmBase):
def __init__(self, framework: ops.Framework):
super().__init__(framework)
for evt in self.on.events().values():
self.framework.observe(evt, self._on_event)

def _on_event(self, event):
pass


def test_get_cloud_spec():
scenario_cloud_spec = scenario.CloudSpec(
type="lxd",
name="localhost",
endpoint="https://127.0.0.1:8443",
credential=scenario.CloudCredential(
auth_type="clientcertificate",
attributes={
"client-cert": "foo",
"client-key": "bar",
"server-cert": "baz",
},
),
)
expected_cloud_spec = ops.CloudSpec(
type="lxd",
name="localhost",
endpoint="https://127.0.0.1:8443",
credential=ops.CloudCredential(
auth_type="clientcertificate",
attributes={
"client-cert": "foo",
"client-key": "bar",
"server-cert": "baz",
},
),
)
ctx = scenario.Context(MyCharm, meta={"name": "foo"})
state = scenario.State(
cloud_spec=scenario_cloud_spec,
model=scenario.Model(name="lxd-model", type="lxd"),
)
with ctx.manager("start", state=state) as mgr:
assert mgr.charm.model.get_cloud_spec() == expected_cloud_spec


def test_get_cloud_spec_error():
ctx = scenario.Context(MyCharm, meta={"name": "foo"})
state = scenario.State(model=scenario.Model(name="lxd-model", type="lxd"))
with ctx.manager("start", state) as mgr:
with pytest.raises(ops.ModelError):
mgr.charm.model.get_cloud_spec()

0 comments on commit 1df85e1

Please sign in to comment.