Move prompt_user() into utils.py

src/feditest/nodedrivers/__init__.py

@@ -7,8 +7,8 @@
 from typing import Any, cast, final
 
 from feditest.testplan import TestPlanConstellationNode, TestPlanNodeParameter, TestPlanNodeAccountField, TestPlanNodeNonExistingAccountField
-from feditest.reporting import info, warning
-from feditest.utils import hostname_validate, appname_validate, appversion_validate
+from feditest.reporting import info
+from feditest.utils import appname_validate, appversion_validate, hostname_validate, prompt_user
 
 
 APP_PAR = TestPlanNodeParameter(
@@ -386,9 +386,13 @@ class Node(ABC):
     Node, one for the client, and one for the server.
 
     Any application that wishes to benefit from automated test execution with FediTest
-    needs to define for itself a subclass of each protocol-specific subclass of Node
-    so FediTest can control and observe what it needs to when attempting to
+    needs to define for itself a class that inherits from each protocol-specific subclass
+    of Node it supports so FediTest can control and observe what it needs to when attempting to
     participate with the respective protocol.
+
+    Subclasses of Node that have the string "Diag" in them are "diagnostic Nodes" that
+    allow FediTest to control and observe in a more fine-grained manner than could be
+    reasonably expected from an implementation of the respective protocol.
     """
     def __init__(self, rolename: str, config: NodeConfiguration, account_manager: AccountManager | None = None):
         """
@@ -450,11 +454,11 @@ def provision_non_existing_account_for_role(self, role: str | None = None) -> No
 
 
     def add_cert_to_trust_store(self, root_cert: str) -> None:
-        self.prompt_user(f'Please add this temporary certificate to the trust root of node { self } and hit return when done:\n' + root_cert)
+        prompt_user(f'Please add this temporary certificate to the trust root of node { self } and hit return when done:\n' + root_cert)
 
 
     def remove_cert_from_trust_store(self, root_cert: str) -> None:
-        self.prompt_user(f'Please remove this previously-added temporary certificate from the trust store of node { self } and hit return when done:\n' + root_cert)
+        prompt_user(f'Please remove this previously-added temporary certificate from the trust store of node { self } and hit return when done:\n' + root_cert)
 
 
     def __str__(self) -> str:
@@ -463,21 +467,6 @@ def __str__(self) -> str:
         return f'"{ type(self).__name__}" in constellation role "{self.rolename}"'
 
 
-    def prompt_user(self, question: str, value_if_known: Any | None = None, parse_validate: Callable[[str],Any] | None = None) -> Any | None:
-        """
-        If an Node does not natively implement support for a particular method,
-        this method is invoked as a fallback. It prompts the user to enter information
-        at the console.
-
-        question: the text to be emitted to the user as a prompt
-        value_if_known: if given, that value can be used instead of asking the user
-        parse_validate: optional function that attempts to parse and validate the provided user input.
-        If the value is valid, it parses the value and returns the parsed version. If not valid, it returns None.
-        return: the value entered by the user, parsed, or None
-        """
-        return self.node_driver.prompt_user(question, value_if_known, parse_validate)
-
-
 class NodeDriver(ABC):
     """
     This is an abstract superclass for all objects that know how to instantiate Nodes of some kind.
@@ -573,39 +562,6 @@ def _unprovision_node(self, node: Node) -> None:
         pass # pylint: disable=unnecessary-pass
 
 
-    def prompt_user(self, question: str, value_if_known: Any | None = None, parse_validate: Callable[[str],Any] | None = None) -> Any | None:
-        """
-        If an NodeDriver does not natively implement support for a particular method,
-        this method is invoked as a fallback. It prompts the user to enter information
-        at the console.
-
-        This is implemented on NodeDriver rather than Node, so we can also ask
-        provisioning-related questions.
-
-        question: the text to be emitted to the user as a prompt
-        value_if_known: if given, that value can be used instead of asking the user
-        parse_validate: optional function that attempts to parse and validate the provided user input.
-        If the value is valid, it parses the value and returns the parsed version. If not valid, it returns None.
-        return: the value entered by the user, parsed, or None
-        """
-        if value_if_known:
-            if parse_validate is None:
-                return value_if_known
-            ret_parsed = parse_validate(value_if_known)
-            if ret_parsed is not None:
-                return ret_parsed
-            warning(f'Preconfigured value "{ value_if_known }" is invalid, ignoring.')
-
-        while True:
-            ret = input(f'TESTER ACTION REQUIRED: { question }')
-            if parse_validate is None:
-                return ret
-            ret_parsed = parse_validate(ret)
-            if ret_parsed is not None:
-                return ret_parsed
-            print(f'INPUT ERROR: invalid input, try again. Was: "{ ret }"')
-
-
     def __str__(self) -> str:
         return self.__class__.__name__
 

src/feditest/nodedrivers/fallback/fediverse.py

@@ -19,7 +19,7 @@
 )
 from feditest.protocols.fediverse import FediverseNode
 from feditest.testplan import TestPlanConstellationNode, TestPlanNodeAccountField, TestPlanNodeNonExistingAccountField
-from feditest.utils import appname_validate, boolean_parse_validate, hostname_validate, http_https_acct_uri_validate, https_uri_validate
+from feditest.utils import appname_validate, boolean_parse_validate, hostname_validate, http_https_acct_uri_validate, https_uri_validate, prompt_user
 
 
 ROLE_ACCOUNT_FIELD = TestPlanNodeAccountField(
@@ -151,11 +151,11 @@ class FallbackFediverseNode(FediverseNode):
     # Python 3.12 @override
     def provision_account_for_role(self, role: str | None = None) -> Account | None:
         context_msg = f'Node { self }:'
-        uri = cast(str, self.prompt_user(
+        uri = cast(str, prompt_user(
                 context_msg
                 + f' provision an account for account role "{ role }" and enter its URI here (with https: or acct: scheme) (node account field "{ URI_ACCOUNT_FIELD.name }"): ',
                 parse_validate=http_https_acct_uri_validate))
-        actor_uri = cast(str, self.prompt_user(
+        actor_uri = cast(str, prompt_user(
                 context_msg
                 + f' for the account with account role "{ role }", enter its Actor URI here (with https: scheme) (node account field "{ ACTOR_URI_ACCOUNT_FIELD.name }"): ',
                 parse_validate=https_uri_validate))
@@ -165,11 +165,11 @@ def provision_account_for_role(self, role: str | None = None) -> Account | None:
 
     def provision_non_existing_account_for_role(self, role: str | None = None) -> NonExistingAccount | None:
         context_msg = f'Node { self }:'
-        uri = cast(str, self.prompt_user(
+        uri = cast(str, prompt_user(
                 context_msg
                 + f' provide the URI of a non-existing account for account role "{ role }" (with https: or acct: scheme) (node non_existing_account field "{ URI_NON_EXISTING_ACCOUNT_FIELD.name }"): ',
                 parse_validate=http_https_acct_uri_validate))
-        actor_uri = cast(str, self.prompt_user(
+        actor_uri = cast(str, prompt_user(
                 context_msg
                 + f' provide the Actor URI of a non-existing account with account role "{ role }" (with https: scheme) (node non_existing_account field "{ ACTOR_URI_NON_EXISTING_ACCOUNT_FIELD.name }"): ',
                 parse_validate=https_uri_validate))
@@ -204,12 +204,12 @@ def obtain_non_existing_account_identifier(self, rolename: str | None = None ) -
     # Python 3.12 @override
     def make_create_note(self, actor_uri: str, content: str, deliver_to: list[str] | None = None) -> str:
         if deliver_to :
-            return cast(str, self.prompt_user(
+            return cast(str, prompt_user(
                     f'On FediverseNode "{ self.hostname }", make actor "{ actor_uri }" create a Note'
                     + ' to be delivered to ' + ", ".join(deliver_to)
                     + ' and enter its URI when created.'
                     + f' Note content:"""\n{ content }\n"""' ))
-        return cast(str, self.prompt_user(
+        return cast(str, prompt_user(
                 f'On FediverseNode "{ self.hostname }", make actor "{ actor_uri }" create a Note'
                 + ' and enter its URI when created.'
                 + f' Note content:"""\n{ content }\n"""' ))
@@ -218,23 +218,23 @@ def make_create_note(self, actor_uri: str, content: str, deliver_to: list[str] |
 
     # Python 3.12 @override
     def make_announce_object(self, actor_uri, to_be_announced_object_uri: str) -> str:
-        return cast(str, self.prompt_user(
+        return cast(str, prompt_user(
                 f'On FediverseNode "{ self.hostname }", make actor "{ actor_uri }" boost "{ to_be_announced_object_uri }"'
                 + ' and enter the Announce object\'s local URI:',
                 parse_validate=https_uri_validate))
 
 
     # Python 3.12 @override
     def make_reply_note(self, actor_uri, to_be_replied_to_object_uri: str, reply_content: str) -> str:
-        return cast(str, self.prompt_user(
+        return cast(str, prompt_user(
                 f'On FediverseNode "{ self.hostname }", make actor "{ actor_uri }" reply to object with "{ to_be_replied_to_object_uri }"'
                 + ' and enter the Announce object\'s URI when created.'
                 + f' Reply content:"""\n{ reply_content }\n"""' ))
 
 
     # Python 3.12 @override
     def make_follow(self, actor_uri: str, to_follow_actor_uri: str) -> None:
-        self.prompt_user(
+        prompt_user(
                 f'On FediverseNode "{ self.hostname }", make actor "{ actor_uri }" follow actor "{ to_follow_actor_uri }"'
                 + ' and hit return when done.')
 
@@ -244,7 +244,7 @@ def make_follow(self, actor_uri: str, to_follow_actor_uri: str) -> None:
 
     # Python 3.12 @override
     def wait_until_actor_is_following_actor(self, actor_uri: str, to_be_followed_uri: str, max_wait: float = 5.) -> None:
-        answer = self.prompt_user(
+        answer = prompt_user(
                 f'On FediverseNode "{ self.hostname }", wait until actor "{ actor_uri }" is following actor "{ to_be_followed_uri }"'
                 + ' and enter "true"; "false" if it didn\'t happen.',
                 parse_validate=boolean_parse_validate)
@@ -254,7 +254,7 @@ def wait_until_actor_is_following_actor(self, actor_uri: str, to_be_followed_uri
 
     # Python 3.12 @override
     def wait_until_actor_is_followed_by_actor(self, actor_uri: str, to_be_following_uri: str, max_wait: float = 5.) -> None:
-        answer = self.prompt_user(
+        answer = prompt_user(
                 f'On FediverseNode "{ self.hostname }", wait until actor "{ actor_uri }" is followed by actor "{ to_be_following_uri }"'
                 + ' and enter "true"; "false" if it didn\'t happen.',
                 parse_validate=boolean_parse_validate)
@@ -264,7 +264,7 @@ def wait_until_actor_is_followed_by_actor(self, actor_uri: str, to_be_following_
 
     # Python 3.12 @override
     def wait_until_actor_is_unfollowing_actor(self, actor_uri: str, to_be_unfollowed_uri: str, max_wait: float = 5.) -> None:
-        answer = self.prompt_user(
+        answer = prompt_user(
                 f'On FediverseNode "{ self.hostname }", wait until actor "{ actor_uri }" is not following any more actor "{ to_be_unfollowed_uri }"'
                 + ' and enter "true"; "false" if it didn\'t happen.',
                 parse_validate=boolean_parse_validate)
@@ -274,7 +274,7 @@ def wait_until_actor_is_unfollowing_actor(self, actor_uri: str, to_be_unfollowed
 
     # Python 3.12 @override
     def wait_until_actor_is_unfollowed_by_actor(self, actor_uri: str, to_be_unfollowing_uri: str, max_wait: float = 5.) -> None:
-        answer = self.prompt_user(
+        answer = prompt_user(
                 f'On FediverseNode "{ self.hostname }", wait until in actor "{ actor_uri }" is not followed any more by actor "{ to_be_unfollowing_uri }"'
                 + ' and enter "true"; "false" if it didn\'t happen.',
                 parse_validate=boolean_parse_validate)
@@ -306,10 +306,10 @@ def create_configuration_account_manager(self, rolename: str, test_plan_node: Te
         hostname = test_plan_node.parameter(HOSTNAME_PAR)
 
         if not hostname:
-            hostname = self.prompt_user(f'Enter the hostname for the Node of constellation role "{ rolename }" (node parameter "hostname"): ',
+            hostname = prompt_user(f'Enter the hostname for the Node of constellation role "{ rolename }" (node parameter "hostname"): ',
                                         parse_validate=hostname_validate)
         if not app:
-            app = self.prompt_user(f'Enter the name of the app at constellation role "{ rolename }" and hostname "{ hostname }" (node parameter "app"): ',
+            app = prompt_user(f'Enter the name of the app at constellation role "{ rolename }" and hostname "{ hostname }" (node parameter "app"): ',
                                    parse_validate=appname_validate)
 
         accounts : list[Account] = []

src/feditest/nodedrivers/manual/__init__.py

@@ -5,6 +5,7 @@
 from feditest.nodedrivers import AccountManager, Node, NodeConfiguration
 from feditest.nodedrivers.fallback.fediverse import AbstractFallbackFediverseNodeDriver, FallbackFediverseNode
 from feditest.protocols.fediverse import FediverseNode
+from feditest.utils import prompt_user
 
 
 class FediverseManualNodeDriver(AbstractFallbackFediverseNodeDriver):
@@ -13,11 +14,11 @@ class FediverseManualNodeDriver(AbstractFallbackFediverseNodeDriver):
     """
     # Python 3.12 @override
     def _provision_node(self, rolename: str, config: NodeConfiguration, account_manager: AccountManager | None) -> FediverseNode:
-        self.prompt_user(f'Manually provision the Node for constellation role { rolename }'
-                         + f' at host { config.hostname } with app { config.app } and hit return when done.')
+        prompt_user(f'Manually provision the Node for constellation role { rolename }'
+                    + f' at host { config.hostname } with app { config.app } and hit return when done.')
         return FallbackFediverseNode(rolename, config, account_manager)
 
 
     # Python 3.12 @override
     def _unprovision_node(self, node: Node) -> None:
-        self.prompt_user(f'Manually unprovision the Node for constellation role { node.rolename } and hit return when done.')
+        prompt_user(f'Manually unprovision the Node for constellation role { node.rolename } and hit return when done.')

src/feditest/nodedrivers/mastodon/__init__.py

@@ -29,7 +29,7 @@
 from feditest.protocols.fediverse import FediverseNode
 from feditest.reporting import is_trace_active, trace
 from feditest.testplan import InvalidAccountSpecificationException, TestPlanConstellationNode, TestPlanNodeAccountField, TestPlanNodeNonExistingAccountField, TestPlanNodeParameter
-from feditest.utils import boolean_parse_validate, email_validate, find_first_in_array, hostname_validate
+from feditest.utils import boolean_parse_validate, email_validate, find_first_in_array, hostname_validate, prompt_user
 
 
 # We use the Mastodon.py module primarily because of its built-in support for rate limiting.
@@ -578,15 +578,15 @@ def obtain_non_existing_account_identifier(self, rolename: str | None = None ) -
     # Python 3.12 @override
     def provision_account_for_role(self, role: str | None = None) -> Account | None:
         context_msg = f'Mastodon Node { self }: '
-        userid = cast(str, self.prompt_user(
+        userid = cast(str, prompt_user(
                 context_msg
                 + f' provide the userid of an existing account for account role "{ role }" (node account field "{ USERID_ACCOUNT_FIELD.name }"): ',
                 parse_validate=_userid_validate))
-        password = cast(str, self.prompt_user(
+        password = cast(str, prompt_user(
                 context_msg
                 + f' provide the password for account "{ userid }", account role "{ role }" (node account field "{ PASSWORD_ACCOUNT_FIELD.name }"): ',
                 parse_validate=_password_validate))
-        email = cast(str, self.prompt_user(
+        email = cast(str, prompt_user(
                 context_msg
                 + f' provide the email for account "{ userid }", account role "{ role }" (node account field "{ EMAIL_ACCOUNT_FIELD.name }"): ',
                 parse_validate=_password_validate))
@@ -596,7 +596,7 @@ def provision_account_for_role(self, role: str | None = None) -> Account | None:
 
     def provision_non_existing_account_for_role(self, role: str | None = None) -> NonExistingAccount | None:
         context_msg = f'Mastodon Node { self }: '
-        userid = cast(str, self.prompt_user(
+        userid = cast(str, prompt_user(
                 context_msg
                 + f' provide the userid of a non-existing account for account role "{ role }" (node non_existing_account field "{ USERID_NON_EXISTING_ACCOUNT_FIELD.name }"): ',
                 parse_validate=_userid_validate))
@@ -764,7 +764,7 @@ def create_configuration_account_manager(self, rolename: str, test_plan_node: Te
         verify_tls_certificate = test_plan_node.parameter_or_raise(VERIFY_API_TLS_CERTIFICATE_PAR, { VERIFY_API_TLS_CERTIFICATE_PAR.name: 'true' })
 
         if not hostname:
-            hostname = self.prompt_user(f'Enter the hostname for the Mastodon Node of constellation role "{ rolename }" (node parameter "hostname"): ',
+            hostname = prompt_user(f'Enter the hostname for the Mastodon Node of constellation role "{ rolename }" (node parameter "hostname"): ',
                                         parse_validate=hostname_validate)
 
         accounts : list[Account] = []

src/feditest/nodedrivers/wordpress/__init__.py

@@ -25,7 +25,7 @@
 from feditest.protocols.fediverse import FediverseNode
 from feditest.reporting import is_trace_active, trace
 from feditest.testplan import TestPlanConstellationNode, TestPlanNodeAccountField, TestPlanNodeNonExistingAccountField, TestPlanNodeParameter
-from feditest.utils import boolean_parse_validate, hostname_validate
+from feditest.utils import boolean_parse_validate, hostname_validate, prompt_user
 
 
 VERIFY_API_TLS_CERTIFICATE_PAR = TestPlanNodeParameter(
@@ -177,7 +177,7 @@ def _actor_uri_to_userid(self, actor_uri: str) -> str:
 
 
     def _provision_oauth_token_for(self, account: WordPressAccount, oauth_client_id: str) -> str:
-        ret = cast(str, self.prompt_user(f'Enter the OAuth token for the Mastodon API for user "{ account.userid  }"'
+        ret = cast(str, prompt_user(f'Enter the OAuth token for the Mastodon API for user "{ account.userid  }"'
                               + f' on constellation role "{ self.rolename }", OAuth client id "{ oauth_client_id }" (user field "{ OAUTH_TOKEN_ACCOUNT_FIELD }"): ',
                               parse_validate=_oauth_token_validate))
         return ret
@@ -213,7 +213,7 @@ def create_configuration_account_manager(self, rolename: str, test_plan_node: Te
         verify_tls_certificate = test_plan_node.parameter_or_raise(VERIFY_API_TLS_CERTIFICATE_PAR, { VERIFY_API_TLS_CERTIFICATE_PAR.name: 'true' })
 
         if not hostname:
-            hostname = self.prompt_user(f'Enter the hostname for the WordPress + ActivityPub plugin Node of constellation role "{ rolename }"'
+            hostname = prompt_user(f'Enter the hostname for the WordPress + ActivityPub plugin Node of constellation role "{ rolename }"'
                                         + f' (node parameter "{ HOSTNAME_PAR }"): ',
                                         parse_validate=hostname_validate)