test: add tests for files being created and modified with text editors

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,7 +81,7 @@ 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], ignored: list[Event], strict: bool):
         while self.is_running():
             msg = self.get_next()
             if msg is None:
@@ -96,8 +96,10 @@ def _wait_events(self, events: list[Event], ignored: list[Event]):
                 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], ignored: list[Event] = [], strict: bool = False):
         """
         Continuously checks the server for incoming events until the
         specified events are found.
@@ -106,9 +108,10 @@ def wait_events(self, events: list[Event], ignored: list[Event] = []):
             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)
+        fs = self.executor.submit(self._wait_events, events, ignored, strict)
         fs.result(timeout=5)

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)

tests/test_editors/test_nvim.py

@@ -0,0 +1,151 @@
+from event import Event, EventType, Process
+
+from test_editors.commons import get_vi_test_file
+
+
+def test_new_file(editor_container, server):
+    fut = '/mounted/test.txt'
+
+    editor_container.exec_run(
+        f"nvim {fut} +':normal iThis is a test<CR>' -c x")
+
+    process = Process.in_container(
+        exe_path='/usr/bin/nvim',
+        args=f'nvim {fut} +:normal iThis is a test<CR> -c x',
+        name='nvim',
+        container_id=editor_container.id[:12],
+    )
+    events = [
+        Event(process=process, event_type=EventType.CREATION,
+              file=fut, host_path=''),
+    ]
+
+    for e in events:
+        print(f'Waiting for event: {e}')
+
+    server.wait_events(events, strict=True)
+
+
+def test_open_file(editor_container, server):
+    fut = '/mounted/test.txt'
+    container_id = editor_container.id[:12]
+
+    # We ensure the file exists before editing.
+    editor_container.exec_run(f'touch {fut}')
+    editor_container.exec_run(
+        f"nvim {fut} +':normal iThis is a test<CR>' -c x")
+
+    touch = Process.in_container(
+        exe_path='/usr/bin/touch',
+        args=f'touch {fut}',
+        name='touch',
+        container_id=container_id,
+    )
+    nvim = Process.in_container(
+        exe_path='/usr/bin/nvim',
+        args=f'nvim {fut} +:normal iThis is a test<CR> -c x',
+        name='nvim',
+        container_id=container_id,
+    )
+
+    vi_test_file = get_vi_test_file('/mounted')
+
+    events = [
+        Event(process=touch, event_type=EventType.CREATION,
+              file=fut, host_path=''),
+        Event(process=nvim, event_type=EventType.CREATION,
+              file=vi_test_file, host_path=''),
+        Event(process=nvim, event_type=EventType.OWNERSHIP,
+              file=vi_test_file, host_path='', owner_uid=0, owner_gid=0),
+        Event(process=nvim, event_type=EventType.UNLINK,
+              file=vi_test_file, host_path=''),
+        Event(process=nvim, event_type=EventType.CREATION,
+              file=fut, host_path=''),
+        Event(process=nvim, event_type=EventType.PERMISSION,
+              file=fut, host_path='', mode=0o100644),
+        Event(process=nvim, event_type=EventType.UNLINK,
+              file=f'{fut}~', host_path=''),
+    ]
+
+    for e in events:
+        print(f'Waiting for event: {e}')
+
+    server.wait_events(events, strict=True)
+
+
+def test_new_file_ovfs(editor_container, server):
+    fut = '/container-dir/test.txt'
+
+    editor_container.exec_run(
+        f"nvim {fut} +':normal iThis is a test<CR>' -c x")
+
+    process = Process.in_container(
+        exe_path='/usr/bin/nvim',
+        args=f'nvim {fut} +:normal iThis is a test<CR> -c x',
+        name='nvim',
+        container_id=editor_container.id[:12],
+    )
+    events = [
+        Event(process=process, event_type=EventType.CREATION,
+              file=fut, host_path=''),
+        Event(process=process, event_type=EventType.OPEN,
+              file=fut, host_path=''),
+    ]
+
+    for e in events:
+        print(f'Waiting for event: {e}')
+
+    server.wait_events(events, strict=True)
+
+
+def test_open_file_ovfs(editor_container, server):
+    fut = '/container-dir/test.txt'
+    container_id = editor_container.id[:12]
+
+    # We ensure the file exists before editing.
+    editor_container.exec_run(f'touch {fut}')
+    editor_container.exec_run(
+        f"nvim {fut} +':normal iThis is a test<CR>' -c x")
+
+    touch = Process.in_container(
+        exe_path='/usr/bin/touch',
+        args=f'touch {fut}',
+        name='touch',
+        container_id=container_id,
+    )
+    nvim = Process.in_container(
+        exe_path='/usr/bin/nvim',
+        args=f'nvim {fut} +:normal iThis is a test<CR> -c x',
+        name='nvim',
+        container_id=container_id,
+    )
+
+    vi_test_file = get_vi_test_file('/container-dir')
+
+    events = [
+        Event(process=touch, event_type=EventType.CREATION,
+              file=fut, host_path=''),
+        Event(process=touch, event_type=EventType.OPEN,
+              file=fut, host_path=''),
+        Event(process=nvim, event_type=EventType.CREATION,
+              file=vi_test_file, host_path=''),
+        Event(process=nvim, event_type=EventType.OPEN,
+              file=vi_test_file, host_path=''),
+        Event(process=nvim, event_type=EventType.OWNERSHIP,
+              file=vi_test_file, host_path='', owner_uid=0, owner_gid=0),
+        Event(process=nvim, event_type=EventType.UNLINK,
+              file=vi_test_file, host_path=''),
+        Event(process=nvim, event_type=EventType.CREATION,
+              file=fut, host_path=''),
+        Event(process=nvim, event_type=EventType.OPEN,
+              file=fut, host_path=''),
+        Event(process=nvim, event_type=EventType.PERMISSION,
+              file=fut, host_path='', mode=0o100644),
+        Event(process=nvim, event_type=EventType.UNLINK,
+              file=f'{fut}~', host_path=''),
+    ]
+
+    for e in events:
+        print(f'Waiting for event: {e}')
+
+    server.wait_events(events, strict=True)