feat(tests): miscellaneous improvements

tests/conftest.py

@@ -11,6 +11,12 @@
 from server import FileActivityService
 
 
+# Declare files holding fixtures
+pytest_plugins = [
+    'test_editors.commons'
+]
+
+
 @pytest.fixture
 def monitored_dir():
     """
@@ -149,7 +155,7 @@ def test_container(request, docker_client, monitored_dir, ignored_dir):
     container.remove()
 
 
-@pytest.fixture
+@pytest.fixture(autouse=True)
 def fact(request, docker_client, fact_config, server, logs_dir, test_file):
     """
     Run the fact docker container for integration tests.

tests/containers/editors/Containerfile

@@ -0,0 +1,5 @@
+FROM quay.io/fedora/fedora:43
+
+RUN dnf install -y \
+    neovim \
+    vim

tests/event.py

@@ -1,4 +1,5 @@
 import os
+from re import Pattern
 import string
 from enum import Enum
 from typing import Any, override
@@ -104,6 +105,21 @@ def get_id(line: str, wanted_id: str) -> int | None:
                        container_id=container_id,
                        loginuid=loginuid)
 
+    @classmethod
+    def in_container(cls,
+                     exe_path: str,
+                     args: str,
+                     name: str,
+                     container_id: str):
+        return Process(pid=None,
+                       uid=0,
+                       gid=0,
+                       loginuid=pow(2, 32)-1,
+                       exe_path=exe_path,
+                       args=args,
+                       name=name,
+                       container_id=container_id)
+
     @property
     def uid(self) -> int:
         return self._uid
@@ -161,6 +177,13 @@ def __str__(self) -> str:
                 f'loginuid={self.loginuid})')
 
 
+def cmp_path(p1: str | Pattern[str], p2: str) -> bool:
+    if isinstance(p1, Pattern):
+        return bool(p1.match(p2))
+    else:
+        return p1 == p2
+
+
 class Event:
     """
     Represents a file activity event, associating a process with an
@@ -170,15 +193,15 @@ class Event:
     def __init__(self,
                  process: Process,
                  event_type: EventType,
-                 file: str,
-                 host_path: str = '',
+                 file: str | Pattern[str],
+                 host_path: str | Pattern[str] = '',
                  mode: int | None = None,
                  owner_uid: int | None = None,
                  owner_gid: int | None = None,):
         self._type: EventType = event_type
         self._process: Process = process
-        self._file: str = file
-        self._host_path: str = host_path
+        self._file: str | Pattern[str] = file
+        self._host_path: str | Pattern[str] = host_path
         self._mode: int | None = mode
         self._owner_uid: int | None = owner_uid
         self._owner_gid: int | None = owner_gid
@@ -192,11 +215,11 @@ def process(self) -> Process:
         return self._process
 
     @property
-    def file(self) -> str:
+    def file(self) -> str | Pattern[str]:
         return self._file
 
     @property
-    def host_path(self) -> str:
+    def host_path(self) -> str | Pattern[str]:
         return self._host_path
 
     @property
@@ -218,21 +241,21 @@ def __eq__(self, other: Any) -> bool:
                 return False
 
             if self.event_type == EventType.CREATION:
-                return self.file == other.creation.activity.path and \
-                    self.host_path == other.creation.activity.host_path
+                return cmp_path(self.file, other.creation.activity.path) and \
+                    cmp_path(self.host_path, other.creation.activity.host_path)
             elif self.event_type == EventType.OPEN:
-                return self.file == other.open.activity.path and \
-                    self.host_path == other.open.activity.host_path
+                return cmp_path(self.file, other.open.activity.path) and \
+                    cmp_path(self.host_path, other.open.activity.host_path)
             elif self.event_type == EventType.UNLINK:
-                return self.file == other.unlink.activity.path and \
-                    self.host_path == other.unlink.activity.host_path
+                return cmp_path(self.file, other.unlink.activity.path) and \
+                    cmp_path(self.host_path, other.unlink.activity.host_path)
             elif self.event_type == EventType.PERMISSION:
-                return self.file == other.permission.activity.path and \
-                    self.host_path == other.permission.activity.host_path and \
+                return cmp_path(self.file, other.permission.activity.path) and \
+                    cmp_path(self.host_path, other.permission.activity.host_path) and \
                     self.mode == other.permission.mode
             elif self.event_type == EventType.OWNERSHIP:
-                return self.file == other.ownership.activity.path and \
-                    self.host_path == other.ownership.activity.host_path and \
+                return cmp_path(self.file, other.ownership.activity.path) and \
+                    cmp_path(self.host_path, other.ownership.activity.host_path) and \
                     self.owner_uid == other.ownership.uid and \
                     self.owner_gid == other.ownership.gid
             return False

tests/server.py

@@ -81,34 +81,34 @@ def is_running(self):
         """
         return self.running.is_set()
 
-    def _wait_events(self, events: list[Event], ignored: list[Event]):
+    def _wait_events(self, events: list[Event], strict: bool):
         while self.is_running():
             msg = self.get_next()
             if msg is None:
                 sleep(0.5)
                 continue
 
             print(f'Got event: {msg}')
-            if msg in ignored:
-                raise ValueError(f'Caught ignored event: {msg}')
-
             if msg in events:
                 events.remove(msg)
                 if len(events) == 0:
                     break
+            elif strict:
+                raise ValueError(f'Encountered unexpected event: {msg}')
 
-    def wait_events(self, events: list[Event], ignored: list[Event] = []):
+    def wait_events(self, events: list[Event], strict: bool = True):
         """
         Continuously checks the server for incoming events until the
         specified events are found.
 
         Args:
             server: The server instance to retrieve events from.
             event (list[Event]): The events to search for.
-            ignored (list[Event]): List of events that should not happen.
+            strict (bool): Fail if an unexpected event is detected.
 
         Raises:
             TimeoutError: If the required events are not found in 5 seconds.
         """
-        fs = self.executor.submit(self._wait_events, events, ignored)
+        print('Waiting for events:', *events, sep='\n')
+        fs = self.executor.submit(self._wait_events, events, strict)
         fs.result(timeout=5)

tests/test_config_hotreload.py

@@ -100,7 +100,6 @@ def test_output_grpc_address_change(fact, fact_config, monitored_dir, server, al
     process = Process.from_proc()
     e = Event(process=process, event_type=EventType.CREATION,
               file=fut, host_path='')
-    print(f'Waiting for event: {e}')
 
     server.wait_events([e])
 
@@ -114,7 +113,6 @@ def test_output_grpc_address_change(fact, fact_config, monitored_dir, server, al
 
     e = Event(process=process, event_type=EventType.OPEN,
               file=fut, host_path='')
-    print(f'Waiting for event on alternate server: {e}')
 
     alternate_server.wait_events([e])
 
@@ -127,20 +125,15 @@ def test_paths(fact, fact_config, monitored_dir, ignored_dir, server):
     with open(ignored_file, 'w') as f:
         f.write('This is to be ignored')
 
-    ignored_event = Event(process=p, event_type=EventType.CREATION,
-                          file=ignored_file, host_path='')
-    print(f'Ignoring: {ignored_event}')
-
     # File Under Test
     fut = os.path.join(monitored_dir, 'test2.txt')
     with open(fut, 'w') as f:
         f.write('This is a test')
 
     e = Event(process=p, event_type=EventType.CREATION,
               file=fut, host_path='')
-    print(f'Waiting for event: {e}')
 
-    server.wait_events([e], ignored=[ignored_event])
+    server.wait_events([e])
 
     config, config_file = fact_config
     config['paths'] = [ignored_dir]
@@ -153,17 +146,12 @@ def test_paths(fact, fact_config, monitored_dir, ignored_dir, server):
 
     e = Event(process=p, event_type=EventType.OPEN,
               file=ignored_file, host_path='')
-    print(f'Waiting for event: {e}')
 
     # File Under Test
     with open(fut, 'w') as f:
         f.write('This is another ignored event')
 
-    ignored_event = Event(
-        process=p, event_type=EventType.OPEN, file=fut, host_path='')
-    print(f'Ignoring: {ignored_event}')
-
-    server.wait_events([e], ignored=[ignored_event])
+    server.wait_events([e])
 
 
 def test_paths_addition(fact, fact_config, monitored_dir, ignored_dir, server):
@@ -174,20 +162,15 @@ def test_paths_addition(fact, fact_config, monitored_dir, ignored_dir, server):
     with open(ignored_file, 'w') as f:
         f.write('This is to be ignored')
 
-    ignored_event = Event(process=p, event_type=EventType.CREATION,
-                          file=ignored_file, host_path='')
-    print(f'Ignoring: {ignored_event}')
-
     # File Under Test
     fut = os.path.join(monitored_dir, 'test2.txt')
     with open(fut, 'w') as f:
         f.write('This is a test')
 
     e = Event(process=p, event_type=EventType.CREATION,
               file=fut, host_path='')
-    print(f'Waiting for event: {e}')
 
-    server.wait_events([e], ignored=[ignored_event])
+    server.wait_events([e])
 
     config, config_file = fact_config
     config['paths'] = [monitored_dir, ignored_dir]
@@ -205,6 +188,5 @@ def test_paths_addition(fact, fact_config, monitored_dir, ignored_dir, server):
               file=ignored_file, host_path=''),
         Event(process=p, event_type=EventType.OPEN, file=fut, host_path='')
     ]
-    print(f'Waiting for events: {events}')
 
     server.wait_events(events)

tests/test_editors/commons.py

@@ -0,0 +1,49 @@
+import os
+
+import pytest
+
+
+def get_vi_test_file(dir):
+    return os.path.join(dir, '4913')
+
+
+@pytest.fixture(scope='session')
+def build_editor_image(docker_client):
+    image, _ = docker_client.images.build(
+        path='containers/editors',
+        tag='editors:latest',
+        dockerfile='Containerfile',
+    )
+    return image
+
+
+def run_editor_container(image, docker_client, ignored_dir):
+    container = docker_client.containers.run(
+        image,
+        detach=True,
+        tty=True,
+        name='editors',
+        volumes={
+            ignored_dir: {
+                'bind': '/mounted',
+                'mode': 'z',
+            },
+        },
+    )
+    container.exec_run('mkdir /container-dir')
+
+    yield container
+
+    container.kill()
+    container.remove()
+
+
+@pytest.fixture
+def vi_container(docker_client, ignored_dir):
+    yield from run_editor_container('quay.io/fedora/fedora:43', docker_client, ignored_dir)
+
+
+@pytest.fixture
+def editor_container(build_editor_image, docker_client, ignored_dir):
+    image = build_editor_image.tags[0]
+    yield from run_editor_container(image, docker_client, ignored_dir)