Tests: lttngtest: add methods to control session rotations

[lttng-tools.git] / tests / utils / lttngtest / lttngctl.py

diff --git a/tests/utils/lttngtest/lttngctl.py b/tests/utils/lttngtest/lttngctl.py
index 9b9658ec626086ef280d8acc9ff19c26894f60d0..64c78b83909a91a079019bcdecd78a804b3bb369 100644 (file)
--- a/tests/utils/lttngtest/lttngctl.py
+++ b/tests/utils/lttngtest/lttngctl.py
@@ -89,6 +89,17 @@ class TracingDomain(enum.Enum):
         return "<%s.%s>" % (self.__class__.__name__, self.name)
 
 
+@enum.unique
+class BufferSharingPolicy(enum.Enum):
+    """Buffer sharing policy."""
+
+    PerUID = "Per-UID buffering"
+    PerPID = "Per-PID buffering"
+
+    def __repr__(self):
+        return "<%s.%s>" % (self.__class__.__name__, self.name)
+
+
 class EventRule(abc.ABC):
     """Event rule base class, see LTTNG-EVENT-RULE(7)."""
 
@@ -351,8 +362,13 @@ class Session(abc.ABC):
         pass
 
     @abc.abstractmethod
-    def add_channel(self, domain, channel_name=None):
-        # type: (TracingDomain, Optional[str]) -> Channel
+    def add_channel(
+        self,
+        domain,
+        channel_name=None,
+        buffer_sharing_policy=BufferSharingPolicy.PerUID,
+    ):
+        # type: (TracingDomain, Optional[str], BufferSharingPolicy) -> Channel
         """Add a channel with default attributes to the session."""
         pass
 
@@ -376,6 +392,11 @@ class Session(abc.ABC):
         # type: () -> bool
         pass
 
+    @abc.abstractmethod
+    def rotate(self):
+        # type: () -> None
+        raise NotImplementedError
+
     @abc.abstractproperty
     def kernel_pid_process_attribute_tracker(self):
         # type: () -> Type[ProcessIDProcessAttributeTracker]
@@ -527,3 +548,27 @@ class Controller(abc.ABC):
         List all sessions visible to the current user.
         """
         pass
+
+    @abc.abstractmethod
+    def rotate_session_by_name(self, name, wait=True):
+        # type: (str, bool) -> None
+        """
+        Rotate a session
+        """
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def schedule_size_based_rotation(self, name, size_bytes):
+        # type: (str, int) -> None
+        """
+        Schedule automatic size-based rotations.
+        """
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def schedule_time_based_rotation(self, name, period_seconds):
+        # type: (str, int) -> None
+        """
+        Schedule automatic time-based rotations.
+        """
+        raise NotImplementedError