Make xattr dependency conditional, add stub xattr library

examples/test_examples.py

@@ -2,7 +2,11 @@
 import subprocess
 
 import pytest
-import xattr
+
+try:
+    import xattr
+except ImportError:
+    import ofrak_core.ofrak.core.xattr_stub as xattr  # type: ignore[no-redef]
 
 from examples.ex5_binary_extension import SEVEN_KITTEH
 from examples.ex8_recursive_unpacking import KITTEH as KITTEH_ASCII

ofrak_core/Makefile

@@ -12,10 +12,15 @@ develop: ofrak/gui/public
 .PHONY: inspect
 inspect:
 	mypy
+
+.PHONY: serial
+serial:
+	$(PYTHON) -m pytest -n0 -m "serial" test_ofrak --cov=ofrak --cov-report=term-missing
+	fun-coverage 
 
 .PHONY: test
-test: inspect
-	$(PYTHON) -m pytest -n auto test_ofrak --cov=ofrak --cov-report=term-missing
+test: inspect serial
+	$(PYTHON) -m pytest -n auto -m "not serial" test_ofrak --cov=ofrak --cov-report=term-missing --cov-append
 	fun-coverage --cov-fail-under=100
 
 .PHONY: dependencies

ofrak_core/ofrak/core/filesystem.py

@@ -4,7 +4,10 @@
 from dataclasses import dataclass
 from typing import Dict, Iterable, Optional, Type, Union
 
-import xattr
+try:
+    import xattr
+except ImportError:
+    import ofrak.core.xattr_stub as xattr  # type: ignore[no-redef]
 
 from ofrak.model.viewable_tag_model import AttributesType
 from ofrak.resource import Resource

ofrak_core/ofrak/core/xattr_stub.py

@@ -0,0 +1,151 @@
+import inspect
+import logging
+
+
+LOGGER = logging.getLogger(__name__)
+
+
+class xattr:
+    """
+    Stub library to support OFRAK on Windows and other platforms where xattr is not available.
+    """
+
+    def __init__(self, obj, options=0):
+        frame = inspect.currentframe()
+        _warn_user_no_xattr(inspect.getframeinfo(frame).function)
+
+    def __repr__(self):
+        frame = inspect.currentframe()
+        _warn_user_no_xattr(inspect.getframeinfo(frame).function)
+        return ""
+
+    def _call(self, name_func, fd_func, *args):
+        frame = inspect.currentframe()
+        _warn_user_no_xattr(inspect.getframeinfo(frame).function)
+
+    def get(self, name, options=0):
+        frame = inspect.currentframe()
+        _warn_user_no_xattr(inspect.getframeinfo(frame).function)
+        return b""
+
+    def set(self, name, value, options=0):
+        frame = inspect.currentframe()
+        _warn_user_no_xattr(inspect.getframeinfo(frame).function)
+        return None
+
+    def remove(self, name, options=0):
+        frame = inspect.currentframe()
+        _warn_user_no_xattr(inspect.getframeinfo(frame).function)
+        return None
+
+    def list(self, options=0):
+        frame = inspect.currentframe()
+        _warn_user_no_xattr(inspect.getframeinfo(frame).function)
+        return []
+
+    def __len__(self):
+        frame = inspect.currentframe()
+        _warn_user_no_xattr(inspect.getframeinfo(frame).function)
+        return 0
+
+    def __delitem__(self, item):
+        frame = inspect.currentframe()
+        _warn_user_no_xattr(inspect.getframeinfo(frame).function)
+        return None
+
+    def __setitem__(self, item, value):
+        frame = inspect.currentframe()
+        _warn_user_no_xattr(inspect.getframeinfo(frame).function)
+        return None
+
+    def __getitem__(self, item):
+        frame = inspect.currentframe()
+        _warn_user_no_xattr(inspect.getframeinfo(frame).function)
+        return b""
+
+    def iterkeys(self):
+        frame = inspect.currentframe()
+        _warn_user_no_xattr(inspect.getframeinfo(frame).function)
+        return iter(list())
+
+    def has_key(self, item):
+        frame = inspect.currentframe()
+        _warn_user_no_xattr(inspect.getframeinfo(frame).function)
+        return False
+
+    def clear(self):
+        frame = inspect.currentframe()
+        _warn_user_no_xattr(inspect.getframeinfo(frame).function)
+        return None
+
+    def update(self, seq):
+        frame = inspect.currentframe()
+        _warn_user_no_xattr(inspect.getframeinfo(frame).function)
+        return None
+
+    def copy(self):
+        frame = inspect.currentframe()
+        _warn_user_no_xattr(inspect.getframeinfo(frame).function)
+        return dict()
+
+    def setdefault(self, k, d=""):
+        frame = inspect.currentframe()
+        _warn_user_no_xattr(inspect.getframeinfo(frame).function)
+        return b""
+
+    def keys(self):
+        frame = inspect.currentframe()
+        _warn_user_no_xattr(inspect.getframeinfo(frame).function)
+        return []
+
+    def itervalues(self):
+        frame = inspect.currentframe()
+        _warn_user_no_xattr(inspect.getframeinfo(frame).function)
+        yield b""
+
+    def values(self):
+        frame = inspect.currentframe()
+        _warn_user_no_xattr(inspect.getframeinfo(frame).function)
+        return []
+
+    def iteritems(self):
+        frame = inspect.currentframe()
+        _warn_user_no_xattr(inspect.getframeinfo(frame).function)
+        yield tuple()
+
+    def items(self):
+        frame = inspect.currentframe()
+        _warn_user_no_xattr(inspect.getframeinfo(frame).function)
+        return []
+
+
+def listxattr(f, symlink=False):
+    frame = inspect.currentframe()
+    _warn_user_no_xattr(inspect.getframeinfo(frame).function)
+    return tuple()
+
+
+def getxattr(f, attr, symlink=False):
+    frame = inspect.currentframe()
+    _warn_user_no_xattr(inspect.getframeinfo(frame).function)
+    return b""
+
+
+def setxattr(f, attr, value, options=0, symlink=False):
+    frame = inspect.currentframe()
+    _warn_user_no_xattr(inspect.getframeinfo(frame).function)
+    return None
+
+
+def removexattr(f, attr, symlink=False):
+    frame = inspect.currentframe()
+    _warn_user_no_xattr(inspect.getframeinfo(frame).function)
+    return None
+
+
+def _warn_user_no_xattr(function_name: str) -> None:
+    logging.warning(
+        f"Function {function_name} not found. Library xattr is not available on Windows platforms. \
+        Extended attributes will not be properly handled while using OFRAK on this platform. \
+        If you require extended attributes, please use a platform that supports xattr."
+    )

ofrak_core/pytest.ini

@@ -1,2 +1,4 @@
 [pytest]
 asyncio_mode = auto
+markers = 
+    serial: force tests to run in serial

ofrak_core/pytest_ofrak/patterns/pack_unpack_filesystem.py

@@ -4,7 +4,10 @@
 from abc import ABC, abstractmethod
 from subprocess import CalledProcessError
 
-import xattr
+try:
+    import xattr
+except ImportError:
+    import ofrak.core.xattr_stub as xattr  # type: ignore[no-redef]
 
 from ofrak import OFRAKContext
 from ofrak.component.abstract import ComponentSubprocessError

ofrak_core/requirements.txt

@@ -14,4 +14,4 @@ sortedcontainers==2.2.2
 synthol~=0.1.1
 typeguard~=2.13.3
 ubi-reader==0.8.5
-xattr==0.10.1
+xattr==0.10.1;platform_system!="Windows"

ofrak_core/test_ofrak/components/test_filesystem_component.py

@@ -71,6 +71,7 @@ async def filesystem_root(ofrak_context: OFRAKContext) -> Resource:
         yield filesystem_root
 
 
+@pytest.mark.serial
 class TestFilesystemRoot:
     """
     Test FilesystemRoot methods.
@@ -164,6 +165,7 @@ async def test_remove_file(self, filesystem_root: FilesystemRoot):
         assert CHILD_TEXTFILE_NAME not in updated_list_dir_output
 
 
+@pytest.mark.serial
 class TestFilesystemEntry:
     """
     Test FilesystemEntry methods.
@@ -189,6 +191,7 @@ async def test_modify_xattr_attribute(self, filesystem_root: FilesystemRoot):
         assert child_textfile.xattrs == {"user.foo": b"bar"}
 
 
+@pytest.mark.serial
 class TestFolder:
     async def test_get_entry(self, filesystem_root: FilesystemRoot):
         """
@@ -201,6 +204,7 @@ async def test_get_entry(self, filesystem_root: FilesystemRoot):
         assert subchild.name == SUBCHILD_TEXTFILE_NAME
 
 
+@pytest.mark.serial
 class TestSymbolicLinkUnpackPack(FilesystemPackUnpackVerifyPattern):
     def setup(self):
         super().setup()
@@ -338,6 +342,7 @@ async def extract(self, root_resource: Resource, extract_dir: str):
             subprocess.run(command, check=True, capture_output=True)
 
 
+@pytest.mark.serial
 class TestLoadInMemoryFilesystem(TestSymbolicLinkUnpackPack):
     async def create_root_resource(self, ofrak_context: OFRAKContext, directory: str) -> Resource:
         with tempfile.TemporaryDirectory() as archive_dir:
@@ -363,6 +368,7 @@ async def extract(self, root_resource: Resource, extract_dir: str):
         await super().extract(child, extract_dir)
 
 
+@pytest.mark.serial
 def diff_directories(dir_1, dir_2, extra_diff_flags):
     """
     Diff two directories and assert that their contents are equal.

ofrak_core/test_ofrak/components/test_filesystem_without_xattr.py

@@ -0,0 +1,16 @@
+import os
+import subprocess
+
+
+import pytest
+
+
+@pytest.mark.serial
+async def test_filesystem_without_xattr():
+    subprocess.run(["pip", "uninstall", "xattr", "-y"])
+    cwd = os.getcwd()
+    os.chdir(os.path.dirname(__file__))
+    result = subprocess.run(["pytest", "./test_filesystem_component.py"])
+    assert result.returncode == 0
+    os.chdir(cwd)
+    subprocess.run(["pip", "install", "xattr"])