Hi Dean, thank you for your work! Some minor comments.
On 26/07/2024 15:15, Dean Marx wrote:
add dataclass for passing in flow rule creation arguments, as well as a
__str__ method for converting to a sendable testpmd command. Add
flow_create method to TestPmdShell class for initializing flow rules.
Signed-off-by: Dean Marx <dm...@iol.unh.edu>
---
dts/framework/remote_session/testpmd_shell.py | 58 ++++++++++++++++++-
1 file changed, 57 insertions(+), 1 deletion(-)
diff --git a/dts/framework/remote_session/testpmd_shell.py
b/dts/framework/remote_session/testpmd_shell.py
index eda6eb320f..d6c111da0a 100644
--- a/dts/framework/remote_session/testpmd_shell.py
+++ b/dts/framework/remote_session/testpmd_shell.py
@@ -19,7 +19,7 @@
from dataclasses import dataclass, field
from enum import Flag, auto
from pathlib import PurePath
-from typing import ClassVar
+from typing import ClassVar, Optional
from typing_extensions import Self, Unpack
@@ -577,6 +577,43 @@ class TestPmdPortStats(TextParser):
tx_bps: int = field(metadata=TextParser.find_int(r"Tx-bps:\s+(\d+)"))
+@dataclass
+class flow_func:
Class names should be UpperCamelCase, also should be a suitable name for
what it's describing. I believe FlowRule should work.
+ """Dataclass for setting flow rule parameters."""
+
+ #:
+ port_id: int
+ #:
+ ingress: bool
+ #:
+ pattern: str
+ #:
+ actions: str
+
+ #:
+ group_id: Optional[int] = None
+ #:
+ priority_level: Optional[int] = None
+ #:
+ user_id: Optional[int] = None
Optional[..] is an outdated notation. `int | None` is preferred instead.
See PEP 604[1].
+
+ def __str__(self) -> str:
+ """Returns the string representation of a flow_func instance.
+
+ In this case, a properly formatted flow create command that can be
sent to testpmd.
+ """
+ ret = []
+ ret.append(f"flow create {self.port_id} ")
+ ret.append(f"group {self.group_id} " if self.group_id is not None else
"")
+ ret.append(f"priority {self.priority_level} " if self.priority_level is not None
else "")
+ ret.append("ingress " if self.ingress else "egress ")
+ ret.append(f"user_id {self.user_id} " if self.user_id is not None else
"")
+ ret.append(f"pattern {self.pattern} ")
+ ret.append(" / end actions ")
+ ret.append(f"{self.actions} / end")
+ return "".join(ret)
Using a list with inline conditional appending is not particularly
readable. A regular string with conditional appending should do:
ret = f"flow create {self.port_id} "
if self.group_id is not None:
ret += f"group {self.group_id} "
...
Also the latest three append lines can all be in one, if you like the
separation you can just do a multi-line string:
ret += (
f"pattern {self.pattern} / end "
f"actions {self.actions} / end"
)
# or actually this may be just fine:
ret += f"pattern {self.pattern} / end "
ret += f"actions {self.actions} / end"
I guess the way it's split is more of a game changer.
If you really want to use a list (in a way that is similar to what I've
described here) then I'd take advantage of it... by omitting leading and
trailing whitespaces and then use the join to add them in between: "
".join(ret)
+
+
class TestPmdShell(DPDKShell):
"""Testpmd interactive shell.
@@ -804,6 +841,25 @@ def show_port_stats(self, port_id: int) -> TestPmdPortStats:
return TestPmdPortStats.parse(output)
+ def flow_create(self, cmd: str, verify: bool = True) -> None:
Not a comment, but a discussion point. Normally we'd want a function to
be read as an action such as:
create_flow
But I understand this is basically mirroring the command format... I
wonder which one would be the best. I am personally inclined in verb
first. Maybe others can give their opinion.
+ """Creates a flow rule in the testpmd session.
+
+ Args:
+ cmd: String from flow_func instance to send as a flow rule.
+ verify: If :data:`True`, the output of the command is scanned
+ to ensure the flow rule was created successfully.
+
+ Raises:
+ InteractiveCommandExecutionError: If flow rule is invalid.
+ """
+ flow_output = self.send_command(cmd)
+ if verify:
+ if "created" not in flow_output:
+ self._logger.debug(f"Failed to create flow
rule:\n{flow_output}")
+ raise InteractiveCommandExecutionError(
+ f"Failed to create flow rule:\n{flow_output}"
+ )
+
def _close(self) -> None:
"""Overrides :meth:`~.interactive_shell.close`."""
self.stop()
[1] https://peps.python.org/pep-0604/
