diff --git a/tests/conftest.py b/tests/conftest.py index 3b74c36c..f23d4dbe 100644 --- a/tests/conftest.py +++ b/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. diff --git a/tests/containers/editors/Containerfile b/tests/containers/editors/Containerfile new file mode 100644 index 00000000..47c91b43 --- /dev/null +++ b/tests/containers/editors/Containerfile @@ -0,0 +1,5 @@ +FROM quay.io/fedora/fedora:43 + +RUN dnf install -y \ + neovim \ + vim diff --git a/tests/event.py b/tests/event.py index 389c01e0..7fee2cb8 100644 --- a/tests/event.py +++ b/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 diff --git a/tests/server.py b/tests/server.py index 2dd17b69..cd0a8c71 100644 --- a/tests/server.py +++ b/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) diff --git a/tests/test_editors/commons.py b/tests/test_editors/commons.py new file mode 100644 index 00000000..150e662d --- /dev/null +++ b/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) diff --git a/tests/test_editors/test_nvim.py b/tests/test_editors/test_nvim.py new file mode 100644 index 00000000..b61bdf1c --- /dev/null +++ b/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' -c x") + + process = Process.in_container( + exe_path='/usr/bin/nvim', + args=f'nvim {fut} +:normal iThis is a test -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' -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 -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' -c x") + + process = Process.in_container( + exe_path='/usr/bin/nvim', + args=f'nvim {fut} +:normal iThis is a test -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' -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 -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) diff --git a/tests/test_editors/test_sed.py b/tests/test_editors/test_sed.py new file mode 100644 index 00000000..ed979f66 --- /dev/null +++ b/tests/test_editors/test_sed.py @@ -0,0 +1,82 @@ +import re +from event import Event, EventType, Process + + +def test_sed(vi_container, server): + # File Under Test + fut = '/mounted/test.txt' + container_id = vi_container.id[:12] + + vi_container.exec_run(f"sh -c \"echo 'This is a test' > {fut}\"") + vi_container.exec_run(fr"sed -i -e 's/a test/not \0/' {fut}") + + shell = Process.in_container( + exe_path='/usr/bin/bash', + args=f"sh -c echo 'This is a test' > {fut}", + name='sh', + container_id=container_id, + ) + sed = Process.in_container( + exe_path='/usr/bin/sed', + args=fr'sed -i -e s/a test/not \0/ {fut}', + name='sed', + container_id=container_id, + ) + + sed_tmp_file = re.compile(r'\/mounted\/sed[0-9a-zA-Z]{6}') + + events = [ + Event(process=shell, event_type=EventType.CREATION, + file=fut, host_path=''), + Event(process=sed, event_type=EventType.CREATION, + file=sed_tmp_file, host_path=''), + Event(process=sed, event_type=EventType.OWNERSHIP, + file=sed_tmp_file, host_path='', owner_uid=0, owner_gid=0), + ] + + for e in events: + print(f'Waiting for event: {e}') + + server.wait_events(events, strict=True) + + +def test_sed_ovfs(vi_container, server): + # File Under Test + fut = '/container-dir/test.txt' + container_id = vi_container.id[:12] + + vi_container.exec_run(f"sh -c \"echo 'This is a test' > {fut}\"") + vi_container.exec_run(fr"sed -i -e 's/a test/not \0/' {fut}") + + shell = Process.in_container( + exe_path='/usr/bin/bash', + args=f"sh -c echo 'This is a test' > {fut}", + name='sh', + container_id=container_id, + ) + sed = Process.in_container( + exe_path='/usr/bin/sed', + args=fr'sed -i -e s/a test/not \0/ {fut}', + name='sed', + container_id=container_id, + ) + + sed_tmp_file = re.compile(r'\/container-dir\/sed[0-9a-zA-Z]{6}') + + events = [ + Event(process=shell, event_type=EventType.CREATION, + file=fut, host_path=''), + Event(process=shell, event_type=EventType.OPEN, + file=fut, host_path=''), + Event(process=sed, event_type=EventType.CREATION, + file=sed_tmp_file, host_path=''), + Event(process=sed, event_type=EventType.OPEN, + file=sed_tmp_file, host_path=''), + Event(process=sed, event_type=EventType.OWNERSHIP, + file=sed_tmp_file, host_path='', owner_uid=0, owner_gid=0), + ] + + for e in events: + print(f'Waiting for event: {e}') + + server.wait_events(events, strict=True) diff --git a/tests/test_editors/test_vi.py b/tests/test_editors/test_vi.py new file mode 100644 index 00000000..59013636 --- /dev/null +++ b/tests/test_editors/test_vi.py @@ -0,0 +1,226 @@ +from event import Event, EventType, Process +from test_editors.commons import get_vi_test_file + + +def test_new_file(vi_container, server): + fut = '/mounted/test.txt' + swap_file = '/mounted/.test.txt.swp' + swx_file = '/mounted/.test.txt.swx' + exe = '/usr/libexec/vi' + + vi_container.exec_run( + f"vi {fut} +':normal iThis is a test' -c x") + + process = Process.in_container( + exe_path=exe, + args=f'{exe} {fut} +:normal iThis is a test -c x', + name='vi', + container_id=vi_container.id[:12], + ) + + events = [ + Event(process=process, event_type=EventType.CREATION, + file=swap_file, host_path=''), + Event(process=process, event_type=EventType.CREATION, + file=swx_file, host_path=''), + Event(process=process, event_type=EventType.UNLINK, + file=swx_file, host_path=''), + Event(process=process, event_type=EventType.UNLINK, + file=swap_file, host_path=''), + Event(process=process, event_type=EventType.CREATION, + file=swap_file, host_path=''), + Event(process=process, event_type=EventType.CREATION, + file=fut, host_path=''), + Event(process=process, event_type=EventType.UNLINK, + file=swap_file, host_path=''), + ] + + for e in events: + print(f'Waiting for event: {e}') + + server.wait_events(events, strict=True) + + +def test_new_file_ovfs(vi_container, server): + fut = '/container-dir/test.txt' + swap_file = '/container-dir/.test.txt.swp' + swx_file = '/container-dir/.test.txt.swx' + exe = '/usr/libexec/vi' + + vi_container.exec_run( + f"vi {fut} +':normal iThis is a test' -c x") + + process = Process.in_container( + exe_path=exe, + args=f'{exe} {fut} +:normal iThis is a test -c x', + name='vi', + container_id=vi_container.id[:12], + ) + + events = [ + Event(process=process, event_type=EventType.CREATION, + file=swap_file, host_path=''), + Event(process=process, event_type=EventType.OPEN, + file=swap_file, host_path=''), + Event(process=process, event_type=EventType.CREATION, + file=swx_file, host_path=''), + Event(process=process, event_type=EventType.OPEN, + file=swx_file, host_path=''), + Event(process=process, event_type=EventType.UNLINK, + file=swx_file, host_path=''), + Event(process=process, event_type=EventType.UNLINK, + file=swap_file, host_path=''), + Event(process=process, event_type=EventType.CREATION, + file=swap_file, host_path=''), + Event(process=process, event_type=EventType.OPEN, + file=swap_file, host_path=''), + Event(process=process, event_type=EventType.CREATION, + file=fut, host_path=''), + Event(process=process, event_type=EventType.OPEN, + file=fut, host_path=''), + Event(process=process, event_type=EventType.UNLINK, + file=swap_file, host_path=''), + ] + + for e in events: + print(f'Waiting for event: {e}') + + server.wait_events(events, strict=True) + + +def test_open_file(vi_container, server): + fut = '/mounted/test.txt' + swap_file = '/mounted/.test.txt.swp' + swx_file = '/mounted/.test.txt.swx' + vi_test_file = get_vi_test_file('/mounted') + exe = '/usr/libexec/vi' + container_id = vi_container.id[:12] + + # We ensure the file exists before editing. + vi_container.exec_run(f'touch {fut}') + vi_container.exec_run( + f"vi {fut} +':normal iThis is a test' -c x") + + touch_process = Process.in_container( + exe_path='/usr/bin/touch', + args=f'touch {fut}', + name='touch', + container_id=container_id, + ) + vi_process = Process.in_container( + exe_path=exe, + args=f'{exe} {fut} +:normal iThis is a test -c x', + name='vi', + container_id=container_id, + ) + + events = [ + Event(process=touch_process, event_type=EventType.CREATION, + file=fut, host_path=''), + Event(process=vi_process, event_type=EventType.CREATION, + file=swap_file, host_path=''), + Event(process=vi_process, event_type=EventType.CREATION, + file=swx_file, host_path=''), + Event(process=vi_process, event_type=EventType.UNLINK, + file=swx_file, host_path=''), + Event(process=vi_process, event_type=EventType.UNLINK, + file=swap_file, host_path=''), + Event(process=vi_process, event_type=EventType.CREATION, + file=swap_file, host_path=''), + Event(process=vi_process, event_type=EventType.PERMISSION, + file=swap_file, host_path='', mode=0o644), + Event(process=vi_process, event_type=EventType.CREATION, + file=vi_test_file, host_path=''), + Event(process=vi_process, event_type=EventType.OWNERSHIP, + file=vi_test_file, host_path='', owner_uid=0, owner_gid=0), + Event(process=vi_process, event_type=EventType.UNLINK, + file=vi_test_file, host_path=''), + Event(process=vi_process, event_type=EventType.CREATION, + file=fut, host_path=''), + Event(process=vi_process, event_type=EventType.PERMISSION, + file=fut, host_path='', mode=0o100644), + Event(process=vi_process, event_type=EventType.UNLINK, + file=f'{fut}~', host_path=''), + Event(process=vi_process, event_type=EventType.UNLINK, + file=swap_file, host_path=''), + ] + + for e in events: + print(f'Waiting for event: {e}') + + server.wait_events(events, strict=True) + + +def test_open_file_ovfs(vi_container, server): + fut = '/container-dir/test.txt' + swap_file = '/container-dir/.test.txt.swp' + swx_file = '/container-dir/.test.txt.swx' + vi_test_file = get_vi_test_file('/container-dir') + exe = '/usr/libexec/vi' + container_id = vi_container.id[:12] + + # We ensure the file exists before editing. + vi_container.exec_run(f'touch {fut}') + vi_container.exec_run( + f"vi {fut} +':normal iThis is a test' -c x") + + touch_process = Process.in_container( + exe_path='/usr/bin/touch', + args=f'touch {fut}', + name='touch', + container_id=container_id, + ) + vi_process = Process.in_container( + exe_path=exe, + args=f'{exe} {fut} +:normal iThis is a test -c x', + name='vi', + container_id=container_id, + ) + + events = [ + Event(process=touch_process, event_type=EventType.CREATION, + file=fut, host_path=''), + Event(process=touch_process, event_type=EventType.OPEN, + file=fut, host_path=''), + Event(process=vi_process, event_type=EventType.CREATION, + file=swap_file, host_path=''), + Event(process=vi_process, event_type=EventType.OPEN, + file=swap_file, host_path=''), + Event(process=vi_process, event_type=EventType.CREATION, + file=swx_file, host_path=''), + Event(process=vi_process, event_type=EventType.OPEN, + file=swx_file, host_path=''), + Event(process=vi_process, event_type=EventType.UNLINK, + file=swx_file, host_path=''), + Event(process=vi_process, event_type=EventType.UNLINK, + file=swap_file, host_path=''), + Event(process=vi_process, event_type=EventType.CREATION, + file=swap_file, host_path=''), + Event(process=vi_process, event_type=EventType.OPEN, + file=swap_file, host_path=''), + Event(process=vi_process, event_type=EventType.PERMISSION, + file=swap_file, host_path='', mode=0o644), + Event(process=vi_process, event_type=EventType.CREATION, + file=vi_test_file, host_path=''), + Event(process=vi_process, event_type=EventType.OPEN, + file=vi_test_file, host_path=''), + Event(process=vi_process, event_type=EventType.OWNERSHIP, + file=vi_test_file, host_path='', owner_uid=0, owner_gid=0), + Event(process=vi_process, event_type=EventType.UNLINK, + file=vi_test_file, host_path=''), + Event(process=vi_process, event_type=EventType.CREATION, + file=fut, host_path=''), + Event(process=vi_process, event_type=EventType.OPEN, + file=fut, host_path=''), + Event(process=vi_process, event_type=EventType.PERMISSION, + file=fut, host_path='', mode=0o100644), + Event(process=vi_process, event_type=EventType.UNLINK, + file=f'{fut}~', host_path=''), + Event(process=vi_process, event_type=EventType.UNLINK, + file=swap_file, host_path=''), + ] + + for e in events: + print(f'Waiting for event: {e}') + + server.wait_events(events, strict=True) diff --git a/tests/test_editors/test_vim.py b/tests/test_editors/test_vim.py new file mode 100644 index 00000000..b3778c91 --- /dev/null +++ b/tests/test_editors/test_vim.py @@ -0,0 +1,220 @@ +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' + swap_file = '/mounted/.test.txt.swp' + swx_file = '/mounted/.test.txt.swx' + + editor_container.exec_run( + f"vim {fut} +':normal iThis is a test' -c x") + + process = Process.in_container( + exe_path='/usr/bin/vim', + args=f'vim {fut} +:normal iThis is a test -c x', + name='vim', + container_id=editor_container.id[:12], + ) + events = [ + Event(process=process, event_type=EventType.CREATION, + file=swap_file, host_path=''), + Event(process=process, event_type=EventType.CREATION, + file=swx_file, host_path=''), + Event(process=process, event_type=EventType.UNLINK, + file=swx_file, host_path=''), + Event(process=process, event_type=EventType.UNLINK, + file=swap_file, host_path=''), + Event(process=process, event_type=EventType.CREATION, + file=swap_file, host_path=''), + Event(process=process, event_type=EventType.CREATION, + file=fut, host_path=''), + Event(process=process, event_type=EventType.UNLINK, + file=swap_file, 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' + swap_file = '/container-dir/.test.txt.swp' + swx_file = '/container-dir/.test.txt.swx' + + editor_container.exec_run( + f"vim {fut} +':normal iThis is a test' -c x") + + process = Process.in_container( + exe_path='/usr/bin/vim', + args=f'vim {fut} +:normal iThis is a test -c x', + name='vim', + container_id=editor_container.id[:12], + ) + events = [ + Event(process=process, event_type=EventType.CREATION, + file=swap_file, host_path=''), + Event(process=process, event_type=EventType.OPEN, + file=swap_file, host_path=''), + Event(process=process, event_type=EventType.CREATION, + file=swx_file, host_path=''), + Event(process=process, event_type=EventType.OPEN, + file=swx_file, host_path=''), + Event(process=process, event_type=EventType.UNLINK, + file=swx_file, host_path=''), + Event(process=process, event_type=EventType.UNLINK, + file=swap_file, host_path=''), + Event(process=process, event_type=EventType.CREATION, + file=swap_file, host_path=''), + Event(process=process, event_type=EventType.OPEN, + file=swap_file, host_path=''), + Event(process=process, event_type=EventType.CREATION, + file=fut, host_path=''), + Event(process=process, event_type=EventType.OPEN, + file=fut, host_path=''), + Event(process=process, event_type=EventType.UNLINK, + file=swap_file, 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' + swap_file = '/mounted/.test.txt.swp' + swx_file = '/mounted/.test.txt.swx' + vi_test_file = get_vi_test_file('/mounted') + 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"vim {fut} +':normal iThis is a test' -c x") + + touch_process = Process.in_container( + exe_path='/usr/bin/touch', + args=f'touch {fut}', + name='touch', + container_id=container_id, + ) + vi_process = Process.in_container( + exe_path='/usr/bin/vim', + args=f'vim {fut} +:normal iThis is a test -c x', + name='vim', + container_id=container_id, + ) + + events = [ + Event(process=touch_process, event_type=EventType.CREATION, + file=fut, host_path=''), + Event(process=vi_process, event_type=EventType.CREATION, + file=swap_file, host_path=''), + Event(process=vi_process, event_type=EventType.CREATION, + file=swx_file, host_path=''), + Event(process=vi_process, event_type=EventType.UNLINK, + file=swx_file, host_path=''), + Event(process=vi_process, event_type=EventType.UNLINK, + file=swap_file, host_path=''), + Event(process=vi_process, event_type=EventType.CREATION, + file=swap_file, host_path=''), + Event(process=vi_process, event_type=EventType.PERMISSION, + file=swap_file, host_path='', mode=0o644), + Event(process=vi_process, event_type=EventType.CREATION, + file=vi_test_file, host_path=''), + Event(process=vi_process, event_type=EventType.OWNERSHIP, + file=vi_test_file, host_path='', owner_uid=0, owner_gid=0), + Event(process=vi_process, event_type=EventType.UNLINK, + file=vi_test_file, host_path=''), + Event(process=vi_process, event_type=EventType.CREATION, + file=fut, host_path=''), + Event(process=vi_process, event_type=EventType.PERMISSION, + file=fut, host_path='', mode=0o100644), + Event(process=vi_process, event_type=EventType.UNLINK, + file=f'{fut}~', host_path=''), + Event(process=vi_process, event_type=EventType.UNLINK, + file=swap_file, 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' + swap_file = '/container-dir/.test.txt.swp' + swx_file = '/container-dir/.test.txt.swx' + vi_test_file = get_vi_test_file('/container-dir') + 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"vim {fut} +':normal iThis is a test' -c x") + + touch_process = Process.in_container( + exe_path='/usr/bin/touch', + args=f'touch {fut}', + name='touch', + container_id=container_id, + ) + vi_process = Process.in_container( + exe_path='/usr/bin/vim', + args=f'vim {fut} +:normal iThis is a test -c x', + name='vim', + container_id=container_id, + ) + + events = [ + Event(process=touch_process, event_type=EventType.CREATION, + file=fut, host_path=''), + Event(process=touch_process, event_type=EventType.OPEN, + file=fut, host_path=''), + Event(process=vi_process, event_type=EventType.CREATION, + file=swap_file, host_path=''), + Event(process=vi_process, event_type=EventType.OPEN, + file=swap_file, host_path=''), + Event(process=vi_process, event_type=EventType.CREATION, + file=swx_file, host_path=''), + Event(process=vi_process, event_type=EventType.OPEN, + file=swx_file, host_path=''), + Event(process=vi_process, event_type=EventType.UNLINK, + file=swx_file, host_path=''), + Event(process=vi_process, event_type=EventType.UNLINK, + file=swap_file, host_path=''), + Event(process=vi_process, event_type=EventType.CREATION, + file=swap_file, host_path=''), + Event(process=vi_process, event_type=EventType.OPEN, + file=swap_file, host_path=''), + Event(process=vi_process, event_type=EventType.PERMISSION, + file=swap_file, host_path='', mode=0o644), + Event(process=vi_process, event_type=EventType.CREATION, + file=vi_test_file, host_path=''), + Event(process=vi_process, event_type=EventType.OPEN, + file=vi_test_file, host_path=''), + Event(process=vi_process, event_type=EventType.OWNERSHIP, + file=vi_test_file, host_path='', owner_uid=0, owner_gid=0), + Event(process=vi_process, event_type=EventType.UNLINK, + file=vi_test_file, host_path=''), + Event(process=vi_process, event_type=EventType.CREATION, + file=fut, host_path=''), + Event(process=vi_process, event_type=EventType.OPEN, + file=fut, host_path=''), + Event(process=vi_process, event_type=EventType.PERMISSION, + file=fut, host_path='', mode=0o100644), + Event(process=vi_process, event_type=EventType.UNLINK, + file=f'{fut}~', host_path=''), + Event(process=vi_process, event_type=EventType.UNLINK, + file=swap_file, host_path=''), + ] + + for e in events: + print(f'Waiting for event: {e}') + + server.wait_events(events, strict=True)