Offer pip install ops[testing] and if available expose Scenario in op…

…s.testing.

ops/testing.py

@@ -14,6 +14,10 @@
 
 """Infrastructure to build unit tests for charms using the ops library."""
 
+import typing as _typing
+from importlib.metadata import PackageNotFoundError as _PackageNotFoundError
+from importlib.metadata import version as _get_package_version
+
 from ._private.harness import (
     ActionFailed,
     ActionOutput,
@@ -38,8 +42,144 @@
     storage,
 )
 
-# The Harness testing framework.
-_ = ActionFailed
+# If the 'ops.testing' optional extra is installed, make those
+# names available in this namespace.
+try:
+    _version = _get_package_version('ops-scenario')
+except _PackageNotFoundError:
+    pass
+else:
+    if _version and int(_version.split('.', 1)[0]) >= 7:
+        from scenario import (
+            ActionFailed as _ScenarioActionFailed,
+        )
+        from scenario import (
+            ActiveStatus,
+            Address,
+            AnyJson,
+            BindAddress,
+            BlockedStatus,
+            CheckInfo,
+            CloudCredential,
+            CloudSpec,
+            Container,
+            Context,
+            DeferredEvent,
+            ErrorStatus,
+            Exec,
+            ICMPPort,
+            JujuLogLine,
+            MaintenanceStatus,
+            Manager,
+            Model,
+            Mount,
+            Network,
+            Notice,
+            PeerRelation,
+            Port,
+            RawDataBagContents,
+            RawSecretRevisionContents,
+            Relation,
+            RelationBase,
+            Resource,
+            Secret,
+            State,
+            StateValidationError,
+            Storage,
+            StoredState,
+            SubordinateRelation,
+            TCPPort,
+            UDPPort,
+            UnitID,
+            UnknownStatus,
+            WaitingStatus,
+        )
+
+        # The Scenario unit testing framework.
+        _ = ActiveStatus
+        _ = Address
+        _ = AnyJson
+        _ = BindAddress
+        _ = BlockedStatus
+        _ = CheckInfo
+        _ = CloudCredential
+        _ = CloudSpec
+        _ = Container
+        _ = Context
+        _ = DeferredEvent
+        _ = ErrorStatus
+        _ = Exec
+        _ = ICMPPort
+        _ = JujuLogLine
+        _ = MaintenanceStatus
+        _ = Manager
+        _ = Model
+        _ = Mount
+        _ = Network
+        _ = Notice
+        _ = PeerRelation
+        _ = Port
+        _ = RawDataBagContents
+        _ = RawSecretRevisionContents
+        _ = Relation
+        _ = RelationBase
+        _ = Resource
+        _ = Secret
+        _ = State
+        _ = StateValidationError
+        _ = Storage
+        _ = StoredState
+        _ = SubordinateRelation
+        _ = TCPPort
+        _ = UDPPort
+        _ = UnitID
+        _ = UnknownStatus
+        _ = WaitingStatus
+
+        # Handle the name clash between Harness's and Scenario's ActionFailed.
+        class _MergedActionFailed(ActionFailed, _ScenarioActionFailed):
+            """Raised when :code:`event.fail()` is called during an action handler."""
+
+            message: str
+            """Optional details of the failure, as provided by :meth:`ops.ActionEvent.fail`."""
+
+            output: ActionOutput
+            """Any logs and results set by the Charm.
+
+            When using Context.run, both logs and results will be empty - these
+            can be found in Context.action_logs and Context.action_results.
+            """
+
+            state: _typing.Optional[State]
+            """The Juju state after the action has been run.
+
+            When using Harness.run_action, this will be None.
+            """
+
+            def __init__(
+                self,
+                message: str,
+                output: _typing.Optional[ActionOutput] = None,
+                *,
+                state: _typing.Optional[State] = None,
+            ):
+                self.message = message
+                self.output = ActionOutput([], {}) if output is None else output
+                self.state = state
+
+        ActionFailed = _MergedActionFailed
+
+        # Also monkeypatch this merged one in so that isinstance checks work.
+        import ops._private.harness as _harness
+
+        _harness.ActionFailed = _MergedActionFailed
+        import scenario.context as _context
+
+        _context.ActionFailed = _MergedActionFailed
+
+
+# The Harness unit testing framework.
+_ = ActionFailed  # If Scenario has been installed, then this will be the merged ActionFailed.
 _ = ActionOutput
 _ = AppUnitOrName
 _ = CharmType
@@ -53,7 +193,7 @@
 # Names exposed for backwards compatibility
 _ = CharmBase
 _ = CharmMeta
-_ = Container
+_ = Container  # If Scenario has been installed, then this will be scenario.Container.
 _ = ExecProcess
 _ = RelationNotFoundError
 _ = RelationRole

pyproject.toml

@@ -37,6 +37,9 @@ docs = [
     "sphinxcontrib-jquery",
     "sphinxext-opengraph"
 ]
+testing = [
+    "ops-scenario>=7.0.5,<8",
+]
 
 [project.urls]
 "Homepage" = "https://juju.is/docs/sdk"

test/test_testing.py

@@ -7022,3 +7022,42 @@ def test_get_cloud_spec_without_set_error(self, request: pytest.FixtureRequest):
         harness.begin()
         with pytest.raises(ops.ModelError):
             harness.model.get_cloud_spec()
+
+
+@pytest.mark.skipif(
+    not hasattr(ops.testing, 'Context'), reason='requires optional ops[testing] install'
+)
+def test_scenario_available():
+    ctx = ops.testing.Context(ops.CharmBase, meta={'name': 'foo'})
+    state = ctx.run(ctx.on.start(), ops.testing.State())
+    assert isinstance(state.unit_status, ops.testing.UnknownStatus)
+
+
+@pytest.mark.skipif(
+    not hasattr(ops.testing, 'Context'), reason='requires optional ops[testing] install'
+)
+def test_merged_actionfailed():
+    class MyCharm(ops.CharmBase):
+        def __init__(self, framework):
+            super().__init__(framework)
+            framework.observe(self.on.go_action, self._on_go)
+
+        def _on_go(self, event):
+            event.log('\U0001f680')
+            event.set_results({'interim': '\U0001f97a'})
+            event.fail('\U0001f61e')
+
+    harness = ops.testing.Harness(MyCharm, actions="""go:\n description: go""")
+    harness.begin()
+    with pytest.raises(ops.testing.ActionFailed) as excinfo:
+        harness.run_action('go')
+    assert excinfo.value.message == '\U0001f61e'
+    assert excinfo.value.output.logs == ['\U0001f680']
+    assert excinfo.value.output.results == {'interim': '\U0001f97a'}
+
+    ctx = ops.testing.Context(MyCharm, meta={'name': 'foo'}, actions={'go': {}})
+    with pytest.raises(ops.testing.ActionFailed) as excinfo:
+        ctx.run(ctx.on.action('go'), ops.testing.State())
+    assert excinfo.value.message == '\U0001f61e'
+    assert ctx.action_logs == ['\U0001f680']
+    assert ctx.action_results == {'interim': '\U0001f97a'}

tox.ini

@@ -80,6 +80,7 @@ deps =
     pyright==1.1.377
     pytest~=7.2
     typing_extensions~=4.2
+    ops-scenario>=7.0.5,<8.0
 commands =
     pyright {posargs}
 
@@ -94,6 +95,7 @@ deps =
     pytest~=7.2
     pytest-xdist~=3.6
     typing_extensions~=4.2
+    ops-scenario>=7.0.5,<8.0
 commands =
     pytest -n auto --ignore={[vars]tst_path}smoke -v --tb native {posargs}