DO NOT MERGE: example AI development of space commands (main instructions)

jubilant/__init__.py

@@ -1,6 +1,6 @@
 """Jubilant is a Pythonic wrapper around the Juju CLI."""
 
-from . import modeltypes, secrettypes, statustypes
+from . import modeltypes, secrettypes, spacetypes, statustypes
 from ._all_any import (
     all_active,
     all_agents_idle,
@@ -20,6 +20,7 @@
 from ._version import Version
 from .modeltypes import ModelInfo
 from .secrettypes import RevealedSecret, Secret, SecretURI
+from .spacetypes import ShowSpaceInfo, Space, SpaceInfo, SpaceSubnet, SubnetInfo
 from .statustypes import Status
 
 __all__ = [
@@ -30,7 +31,12 @@
     'RevealedSecret',
     'Secret',
     'SecretURI',
+    'ShowSpaceInfo',
+    'Space',
+    'SpaceInfo',
+    'SpaceSubnet',
     'Status',
+    'SubnetInfo',
     'Task',
     'TaskError',
     'Version',
@@ -48,6 +54,7 @@
     'any_waiting',
     'modeltypes',
     'secrettypes',
+    'spacetypes',
     'statustypes',
     'temp_model',
 ]

jubilant/_juju.py

@@ -19,6 +19,7 @@
 from ._version import Version
 from .modeltypes import ModelInfo
 from .secrettypes import RevealedSecret, Secret, SecretURI
+from .spacetypes import ShowSpaceInfo, Space
 from .statustypes import Status
 
 logger = logging.getLogger('jubilant')
@@ -240,6 +241,16 @@ def add_secret(
 
         return SecretURI(output.strip())
 
+    def add_space(self, name: str, *cidrs: str) -> None:
+        """Add a new network space.
+
+        Args:
+            name: Name of the space to add.
+            cidrs: Optional subnet CIDRs to associate with the space,
+                for example ``'172.31.0.0/20'``.
+        """
+        self.cli('add-space', name, *cidrs)
+
     def add_ssh_key(self, *keys: str) -> None:
         """Add one or more SSH keys to the model.
 
@@ -849,6 +860,19 @@ def model_constraints(
         args.extend(_format_config(k, v) for k, v in constraints.items())
         self.cli(*args)
 
+    def move_to_space(self, name: str, *cidrs: str, force: bool = False) -> None:
+        """Move subnets to a network space.
+
+        Args:
+            name: Name of the space to move subnets to.
+            cidrs: Subnet CIDRs to move to the space, for example ``'172.31.0.0/20'``.
+            force: Allow moving subnets even if they are in use on another machine.
+        """
+        args = ['move-to-space', name, *cidrs]
+        if force:
+            args.append('--force')
+        self.cli(*args)
+
     def offer(
         self,
         app: str,
@@ -939,6 +963,10 @@ def refresh(
 
             self.cli(*args)
 
+    def reload_spaces(self) -> None:
+        """Reload spaces and subnets from the substrate."""
+        self.cli('reload-spaces')
+
     def remove_application(
         self,
         *app: str,
@@ -991,6 +1019,20 @@ def remove_secret(self, identifier: str | SecretURI, *, revision: int | None = N
             args.extend(['--revision', str(revision)])
         self.cli(*args)
 
+    def remove_space(self, name: str, *, force: bool = False) -> None:
+        """Remove a network space.
+
+        Any subnets associated with the space will be transferred to the default space.
+
+        Args:
+            name: Name of the space to remove.
+            force: Remove the space even if there are existing bindings, constraints, or settings.
+        """
+        args = ['remove-space', '--yes', name]
+        if force:
+            args.append('--force')
+        self.cli(*args)
+
     def remove_ssh_key(self, *ids: str) -> None:
         """Remove one or more SSH keys from the model.
 
@@ -1044,6 +1086,15 @@ def remove_unit(
 
         self.cli(*args)
 
+    def rename_space(self, name: str, new_name: str) -> None:
+        """Rename a network space.
+
+        Args:
+            name: Current name of the space.
+            new_name: New name for the space.
+        """
+        self.cli('rename-space', name, new_name)
+
     def run(
         self,
         unit: str,
@@ -1277,6 +1328,27 @@ def show_secret(
             return RevealedSecret._from_dict(secret)
         return Secret._from_dict(secret)
 
+    def show_space(self, name: str) -> ShowSpaceInfo:
+        """Get information about a network space.
+
+        Args:
+            name: Name of the space.
+        """
+        stdout = self.cli('show-space', name, '--format', 'json')
+        output = json.loads(stdout)
+        return ShowSpaceInfo._from_dict(output)
+
+    def spaces(self) -> list[Space]:
+        """Get all network spaces in the model.
+
+        Returns:
+            A list of all spaces in the model.
+        """
+        stdout = self.cli('spaces', '--format', 'json')
+        output = json.loads(stdout)
+        spaces_list: list[dict[str, Any]] = output.get('spaces') or []
+        return [Space._from_dict(s) for s in spaces_list]
+
     def ssh(
         self,
         target: str | int,

jubilant/spacetypes.py

@@ -0,0 +1,105 @@
+"""Dataclasses that contain parsed output from ``juju spaces`` and ``juju show-space``."""
+
+from __future__ import annotations
+
+import dataclasses
+from typing import Any
+
+
+@dataclasses.dataclass(frozen=True)
+class SubnetInfo:
+    """Information about a subnet in a space (from ``juju show-space``)."""
+
+    cidr: str
+    vlan_tag: int
+
+    provider_id: str = ''
+    provider_space_id: str = ''
+    provider_network_id: str = ''
+    zones: tuple[str, ...] = ()
+
+    @classmethod
+    def _from_dict(cls, d: dict[str, Any]) -> SubnetInfo:
+        return cls(
+            cidr=d['cidr'],
+            vlan_tag=d.get('vlan-tag', 0),
+            provider_id=d.get('provider-id') or '',
+            provider_space_id=d.get('provider-space-id') or '',
+            provider_network_id=d.get('provider-network-id') or '',
+            zones=tuple(d.get('zones') or ()),
+        )
+
+
+@dataclasses.dataclass(frozen=True)
+class SpaceInfo:
+    """Information about a space (from ``juju show-space``)."""
+
+    id: str
+    name: str
+
+    subnets: tuple[SubnetInfo, ...] = ()
+
+    @classmethod
+    def _from_dict(cls, d: dict[str, Any]) -> SpaceInfo:
+        subnets_list: list[dict[str, Any]] = d.get('subnets') or []
+        return cls(
+            id=d['id'],
+            name=d['name'],
+            subnets=tuple(SubnetInfo._from_dict(s) for s in subnets_list),
+        )
+
+
+@dataclasses.dataclass(frozen=True)
+class ShowSpaceInfo:
+    """Parsed version of the object returned by ``juju show-space --format=json``."""
+
+    space: SpaceInfo
+    applications: tuple[str, ...] = ()
+    machine_count: int = 0
+
+    @classmethod
+    def _from_dict(cls, d: dict[str, Any]) -> ShowSpaceInfo:
+        return cls(
+            space=SpaceInfo._from_dict(d['space']),
+            applications=tuple(d.get('applications') or []),
+            machine_count=d.get('machine-count', 0),
+        )
+
+
+@dataclasses.dataclass(frozen=True)
+class SpaceSubnet:
+    """Information about a subnet in a space (from ``juju spaces``)."""
+
+    type: str
+    status: str
+    zones: tuple[str, ...] = ()
+    provider_id: str = ''
+
+    @classmethod
+    def _from_dict(cls, d: dict[str, Any]) -> SpaceSubnet:
+        return cls(
+            type=d.get('type') or '',
+            status=d.get('status') or '',
+            zones=tuple(d.get('zones') or ()),
+            provider_id=d.get('provider-id') or '',
+        )
+
+
+@dataclasses.dataclass(frozen=True)
+class Space:
+    """Parsed version of a single space from ``juju spaces --format=json``."""
+
+    id: str
+    name: str
+    subnets: dict[str, SpaceSubnet] = dataclasses.field(default_factory=dict)  # type: ignore
+
+    @classmethod
+    def _from_dict(cls, d: dict[str, Any]) -> Space:
+        subnets_dict: dict[str, Any] = d.get('subnets') or {}
+        return cls(
+            id=d['id'],
+            name=d['name'],
+            subnets={
+                cidr: SpaceSubnet._from_dict(subnet) for cidr, subnet in subnets_dict.items()
+            },
+        )

tests/unit/fake_spaces.py

@@ -0,0 +1,69 @@
+"""Fake data for space-related unit tests."""
+
+from __future__ import annotations
+
+from typing import Any
+
+SPACES_LIST: dict[str, Any] = {
+    'spaces': [
+        {
+            'id': '0',
+            'name': 'alpha',
+            'subnets': {
+                '172.31.0.0/20': {
+                    'type': 'ipv4',
+                    'provider-id': 'subnet-abc123',
+                    'status': 'in-use',
+                    'zones': ['us-east-1a', 'us-east-1b'],
+                },
+                '10.0.0.0/24': {
+                    'type': 'ipv4',
+                    'status': 'in-use',
+                    'zones': ['us-east-1a'],
+                },
+            },
+        },
+        {
+            'id': '1',
+            'name': 'db-space',
+            'subnets': {},
+        },
+    ],
+}
+
+SPACES_LIST_EMPTY: dict[str, Any] = {
+    'spaces': [],
+}
+
+SHOW_SPACE: dict[str, Any] = {
+    'space': {
+        'id': '0',
+        'name': 'alpha',
+        'subnets': [
+            {
+                'cidr': '172.31.0.0/20',
+                'provider-id': 'subnet-abc123',
+                'provider-space-id': 'prov-space-1',
+                'provider-network-id': 'vpc-123',
+                'vlan-tag': 0,
+                'zones': ['us-east-1a', 'us-east-1b'],
+            },
+            {
+                'cidr': '10.0.0.0/24',
+                'vlan-tag': 42,
+                'zones': ['us-east-1a'],
+            },
+        ],
+    },
+    'applications': ['mysql', 'wordpress'],
+    'machine-count': 3,
+}
+
+SHOW_SPACE_MINIMAL: dict[str, Any] = {
+    'space': {
+        'id': '2',
+        'name': 'empty-space',
+    },
+    'applications': [],
+    'machine-count': 0,
+}

tests/unit/test_add_space.py

@@ -0,0 +1,24 @@
+import jubilant
+
+from . import mocks
+
+
+def test_name_only(run: mocks.Run):
+    run.handle(['juju', 'add-space', 'my-space'])
+    juju = jubilant.Juju()
+
+    juju.add_space('my-space')
+
+
+def test_with_cidrs(run: mocks.Run):
+    run.handle(['juju', 'add-space', 'my-space', '172.31.0.0/20', '10.0.0.0/24'])
+    juju = jubilant.Juju()
+
+    juju.add_space('my-space', '172.31.0.0/20', '10.0.0.0/24')
+
+
+def test_with_model(run: mocks.Run):
+    run.handle(['juju', 'add-space', '--model', 'mdl', 'my-space'])
+    juju = jubilant.Juju(model='mdl')
+
+    juju.add_space('my-space')

tests/unit/test_move_to_space.py

@@ -0,0 +1,31 @@
+import jubilant
+
+from . import mocks
+
+
+def test_move(run: mocks.Run):
+    run.handle(['juju', 'move-to-space', 'db-space', '172.31.0.0/20'])
+    juju = jubilant.Juju()
+
+    juju.move_to_space('db-space', '172.31.0.0/20')
+
+
+def test_multiple_cidrs(run: mocks.Run):
+    run.handle(['juju', 'move-to-space', 'db-space', '172.31.0.0/20', '10.0.0.0/24'])
+    juju = jubilant.Juju()
+
+    juju.move_to_space('db-space', '172.31.0.0/20', '10.0.0.0/24')
+
+
+def test_force(run: mocks.Run):
+    run.handle(['juju', 'move-to-space', 'db-space', '172.31.0.0/20', '--force'])
+    juju = jubilant.Juju()
+
+    juju.move_to_space('db-space', '172.31.0.0/20', force=True)
+
+
+def test_with_model(run: mocks.Run):
+    run.handle(['juju', 'move-to-space', '--model', 'mdl', 'db-space', '172.31.0.0/20'])
+    juju = jubilant.Juju(model='mdl')
+
+    juju.move_to_space('db-space', '172.31.0.0/20')