Fixing some tests to work on MacOS for development.

modules/videos/common.py

@@ -211,6 +211,9 @@ async def ffprobe_json(video_path: Union[Path, str]) -> dict:
         ],
     }
     """
+    if not FFPROBE_BIN:
+        raise RuntimeError('ffprobe was not found')
+
     cmd = (FFPROBE_BIN,
            '-print_format', 'json',
            '-loglevel', 'quiet',

wrolpi/cmd.py

@@ -92,6 +92,10 @@ YT_DLP_BIN = which(
 FFPROBE_BIN = which('ffprobe', '/usr/bin/ffprobe')
 FFMPEG_BIN = which('ffmpeg', '/usr/bin/ffmpeg')
 
+# Documents
+CATDOC_PATH = which('catdoc')
+TEXTUTIL_PATH = which('textutil')
+
 
 @dataclasses.dataclass
 class CommandResult:

wrolpi/common.py

@@ -370,7 +370,8 @@ URL_CHARS = string.ascii_lowercase + string.digits
 
 
 def find_file(directory: pathlib.Path, name: str, depth=1) -> Optional[pathlib.Path]:
-    """Recursively searches a directory for a file with the provided name."""
+    """Recursively searches a directory for a
+    file with the provided name."""
     if depth == 0:
         return
 
@@ -394,6 +395,17 @@ def set_test_media_directory(path):
         TEST_MEDIA_DIRECTORY = None
 
 
+def is_tempfile(path: pathlib.Path | str) -> bool:
+    path = str(path)
+    if path.startswith('/tmp'):
+        return True
+    elif path.startswith('/var/folders/'):
+        return True
+    elif path.startswith('/private/var/folders/'):
+        return True
+    return False
+
+
 def get_media_directory() -> Path:
     """Get the media directory.
 
@@ -407,8 +419,8 @@ def get_media_directory() -> Path:
         raise RuntimeError('No test media directory set during testing!!')
 
     if isinstance(TEST_MEDIA_DIRECTORY, pathlib.Path):
-        if not str(TEST_MEDIA_DIRECTORY).startswith('/tmp'):
-            raise RuntimeError('Refusing to run test outside tmp directory!')
+        if not is_tempfile(TEST_MEDIA_DIRECTORY):
+            raise RuntimeError(f'Refusing to run test outside tmp directory: {TEST_MEDIA_DIRECTORY}')
         return TEST_MEDIA_DIRECTORY
 
     return MEDIA_DIRECTORY
@@ -536,7 +548,7 @@ class ConfigFile:
                 raise RuntimeError(f'Refusing to overwrite newer config ({rel_path}): {version} > {self.version}')
 
         # Don't overwrite a real config while testing.
-        if PYTEST and not str(file).startswith('/tmp'):
+        if PYTEST and not is_tempfile(file):
             raise ValueError(f'Refusing to save config file while testing: {rel_path}')
 
         logger_.debug(f'Save config called for {rel_path}')
@@ -1132,7 +1144,12 @@ def get_relative_to_media_directory(path: Union[str, Path]) -> Path:
 def get_paths_in_parent_directory(paths: List[Union[str, Path]], parent_directory: pathlib.Path) -> List[Path]:
     """Return only those of the provided paths that are children of the provided parent directory."""
     paths = [i if isinstance(i, Path) else Path(i) for i in paths]
-    return [path for path in paths if parent_directory in path.resolve().parents and path != parent_directory]
+    parent_directory = parent_directory.resolve()
+    new_paths = []
+    for path in paths:
+        if parent_directory in path.resolve().parents and path != parent_directory:
+            new_paths.append(path)
+    return new_paths
 
 
 def get_paths_in_media_directory(paths: List[Union[str, Path]]) -> List[Path]:
@@ -1213,10 +1230,11 @@ def chdir(directory: Union[pathlib.Path, str, None] = None, with_home: bool = Fa
     This also changes the $HOME environment variable to match.
     """
     home_exists = 'HOME' in os.environ
-    home = os.environ.get('HOME')
+    original_home = os.environ.get('HOME')
     cwd = os.getcwd()
     if directory is None:
         with tempfile.TemporaryDirectory() as d:
+            d = os.path.realpath(d)
             try:
                 os.chdir(d)
                 if with_home:
@@ -1225,7 +1243,7 @@ def chdir(directory: Union[pathlib.Path, str, None] = None, with_home: bool = Fa
             finally:
                 os.chdir(cwd)
                 if home_exists:
-                    os.environ['HOME'] = home
+                    os.environ['HOME'] = original_home
                 else:
                     del os.environ['HOME']
             return
@@ -1238,7 +1256,7 @@ def chdir(directory: Union[pathlib.Path, str, None] = None, with_home: bool = Fa
     finally:
         os.chdir(cwd)
         if home_exists:
-            os.environ['HOME'] = home
+            os.environ['HOME'] = original_home
         else:
             del os.environ['HOME']
     return

wrolpi/files/indexers.py

@@ -7,6 +7,8 @@ from zipfile import ZipFile
 
 import docx
 
+from wrolpi import cmd
+from wrolpi.cmd import CATDOC_PATH, TEXTUTIL_PATH
 from wrolpi.vars import PYTEST, FILE_MAX_TEXT_SIZE
 
 try:
@@ -148,9 +150,16 @@ class DocIndexer(Indexer, ABC):
     def create_index(cls, path: pathlib.Path):
         a = cls.get_title(path)
 
-        cmd = ('catdoc', str(path.absolute()))
-        proc = subprocess.run(cmd, capture_output=True)
-        text = proc.stdout.decode()
+        if CATDOC_PATH:
+            # Use catdoc on Linux
+            cmd = (CATDOC_PATH, str(path.absolute()))
+            proc = subprocess.run(cmd, capture_output=True)
+            text = proc.stdout.decode()
+        elif TEXTUTIL_PATH:
+            # Use textutil on macOS.
+            cmd = (TEXTUTIL_PATH, '-stdout', '-cat', 'txt', str(path.absolute()))
+            proc = subprocess.run(cmd, capture_output=True)
+            text = proc.stdout.decode()
         text = truncate_object_bytes(text, FILE_MAX_TEXT_SIZE)
         words = split_lines_by_length(text)
 

wrolpi/files/test/test_ebooks.py

@@ -56,7 +56,7 @@ async def test_index(async_client, test_session, test_directory, example_epub, e
 
 
 @pytest.mark.asyncio
-async def test_discover_local_cover(test_session, test_directory, example_epub, image_bytes_factory):
+async def test_discover_local_cover(test_session, test_directory, example_epub, image_bytes_factory, await_switches):
     cover_path = example_epub.with_suffix('.jpg')
     cover_path.write_bytes(image_bytes_factory())
     await refresh_files()
@@ -68,7 +68,7 @@ async def test_discover_local_cover(test_session, test_directory, example_epub,
 
 
 @pytest.mark.asyncio
-async def test_extract_cover(test_session, test_directory, example_epub):
+async def test_extract_cover(test_session, test_directory, example_epub, await_switches):
     """First image is extracted from the Ebook and used as the cover."""
     await refresh_files()
 

wrolpi/files/test/test_lib.py

@@ -22,7 +22,7 @@ from wrolpi.errors import InvalidFile, UnknownDirectory, FileGroupIsTagged, NoPr
 from wrolpi.files import lib, indexers
 from wrolpi.files.models import FileGroup
 from wrolpi.tags import TagFile
-from wrolpi.vars import PROJECT_DIR
+from wrolpi.vars import PROJECT_DIR, IS_MACOS
 
 
 @pytest.mark.asyncio
@@ -1051,7 +1051,10 @@ async def test_doc_indexer(async_client, test_session, example_doc):
     doc = test_session.query(FileGroup).one()
     assert doc.title == 'example word.doc'
     assert doc.a_text == 'example word doc'
-    assert doc.d_text == 'Example Word Document\n\nSecond page\n'
+    if IS_MACOS:
+        assert doc.d_text == 'Example Word Document\nSecond page'
+    else:
+        assert doc.d_text == 'Example Word Document\n\nSecond page\n'
 
 
 @pytest.mark.asyncio

wrolpi/status.py

@@ -2,6 +2,7 @@
 import asyncio
 import json
 import multiprocessing
+import os
 import pathlib
 import re
 import statistics
@@ -56,10 +57,13 @@ LOAD_AVG_FILE = pathlib.Path('/proc/loadavg')
 
 async def get_load_stats() -> SystemLoad:
     """Read loadavg file and return the 1, 5, and 15 minute system loads."""
-    # Read load information from proc file.
-    # Example:  0.99 1.06 1.02 1/2305 43879
-    loadavg = LOAD_AVG_FILE.read_text()
-    load_1, load_5, load_15, *_ = loadavg.split(' ')
+    if LOAD_AVG_FILE.exists():
+        # Read load information from proc file.
+        # Example:  0.99 1.06 1.02 1/2305 43879
+        loadavg = LOAD_AVG_FILE.read_text()
+        load_1, load_5, load_15, *_ = loadavg.split(' ')
+    else:
+        load_1, load_5, load_15 = os.getloadavg()
     load = SystemLoad(Decimal(load_1), Decimal(load_5), Decimal(load_15))
     return load
 
@@ -317,7 +321,7 @@ DRIVE_REGEX = re.compile(r'^.+?'  # filesystem
                          r'\s+(.*)',  # mount
                          re.MULTILINE)
 IGNORED_DRIVES = ['/boot', '/etc']
-VALID_FORMATS = {'btrfs', 'ext4', 'ext3', 'ext2', 'vfat', 'exfat'}
+VALID_FORMATS = {'btrfs', 'ext4', 'ext3', 'ext2', 'vfat', 'exfat', 'apfs'}
 
 
 def get_drives_info_psutil() -> List[DriveInfo]:

wrolpi/test/common.py

@@ -1,4 +1,5 @@
 import json
+import os
 import pathlib
 import tempfile
 from contextlib import contextmanager
@@ -13,11 +14,12 @@ from wrolpi.api_utils import CustomJSONEncoder
 from wrolpi.common import get_media_directory
 from wrolpi.conftest import test_db, test_client  # noqa
 from wrolpi.db import postgres_engine
-from wrolpi.vars import CIRCLECI
+from wrolpi.vars import CIRCLECI, IS_MACOS
 
 TEST_CONFIG_PATH = tempfile.NamedTemporaryFile(mode='rt', delete=False)
 
 skip_circleci = pytest.mark.skipif(CIRCLECI, reason='This test is not supported in Circle CI')
+skip_macos = pytest.mark.skipif(IS_MACOS, reason='This test is not supported on MacOS')
 
 
 def wrap_test_db(func):

wrolpi/test/test_admin.py

@@ -5,10 +5,11 @@ from unittest.mock import call
 from wrolpi import admin
 from wrolpi.admin import HotspotStatus
 from wrolpi.common import WROLPI_CONFIG
-from wrolpi.test.common import skip_circleci
+from wrolpi.test.common import skip_circleci, skip_macos
 
 
 @skip_circleci
+@skip_macos
 def test_hotspot_status():
     with mock.patch('wrolpi.admin.subprocess') as mock_subprocess:
         mock_subprocess.check_output.return_value = b'wlan0: unavailable'
@@ -113,6 +114,7 @@ def test_throttle_off():
 
 
 @skip_circleci
+@skip_macos
 def test_hotspot_device(async_client):
     """Changing the hotspot device changes what device is turned on."""
     from wrolpi.common import WROLPI_CONFIG

wrolpi/test/test_cmd.py

@@ -6,8 +6,11 @@ from wrolpi import cmd
 
 
 def test_which(test_directory):
-    # Default location in Debian Linux.
-    assert cmd.which('ls') == Path('/usr/bin/ls')
+    if os.uname()[0] == "Darwin":
+        assert cmd.which('ls') == Path('/bin/ls')
+    else:
+        # Default location in Debian Linux.
+        assert cmd.which('ls') == Path('/usr/bin/ls')
     # This executable does not exist.
     assert cmd.which('asdf') is None
     # The alternative path does not exist either.

wrolpi/test/test_common.py

@@ -19,9 +19,10 @@ import pytz
 
 import wrolpi.vars
 from wrolpi import common
-from wrolpi.common import cum_timer, TIMERS, print_timer, limit_concurrent, run_after, get_wrolpi_config, TRACE_LEVEL
+from wrolpi.common import cum_timer, TIMERS, print_timer, limit_concurrent, run_after, get_wrolpi_config, TRACE_LEVEL, \
+    is_tempfile
 from wrolpi.errors import InvalidConfig
-from wrolpi.test.common import build_test_directories, skip_circleci
+from wrolpi.test.common import build_test_directories, skip_circleci, skip_macos
 
 
 def test_build_video_directories(test_directory):
@@ -220,16 +221,17 @@ def test_chdir():
 
     with common.chdir():
         assert os.getcwd() != original
-        assert str(os.getcwd()).startswith('/tmp')
+        assert is_tempfile(os.getcwd())
         assert os.environ['HOME'] != os.getcwd()
     # Replace $HOME
     with common.chdir(with_home=True):
         assert os.getcwd() != original
-        assert str(os.getcwd()).startswith('/tmp')
+        assert is_tempfile(os.getcwd())
         assert os.environ['HOME'] == os.getcwd()
 
     with tempfile.TemporaryDirectory() as d:
         # Without replacing $HOME
+        d = os.path.realpath(d)
         with common.chdir(pathlib.Path(d), with_home=True):
             assert os.getcwd() == d
             assert os.environ['HOME'] == os.getcwd()
@@ -457,8 +459,8 @@ def test_timer():
     with common.timer(name='test_timer'):
         time.sleep(0.1)
 
-
 @pytest.mark.asyncio
+@skip_macos
 async def test_limit_concurrent_async():
     """`limit_concurrent` can throw an error when the limit is reached."""
 
@@ -482,6 +484,7 @@ async def test_limit_concurrent_async():
     assert 'concurrent limit' in str(e)
 
 
+@skip_macos
 def test_limit_concurrent_sync():
     """`limit_concurrent` can throw an error when the limit is reached."""
 

wrolpi/test/test_status.py

@@ -4,7 +4,8 @@ from unittest import mock
 import pytest
 
 from wrolpi import status
-from wrolpi.test.common import skip_circleci
+from wrolpi.test.common import skip_circleci, skip_macos
+from wrolpi.vars import IS_MACOS
 
 
 @pytest.mark.asyncio
@@ -25,10 +26,12 @@ async def test_get_cpu_stats(async_client):
     info = await status.get_cpu_stats()
     assert isinstance(info, status.CPUInfo)
     assert isinstance(info.percent, int)
-    assert isinstance(info.temperature, int)
+    if not IS_MACOS:
+        assert isinstance(info.temperature, int)
 
 
 @pytest.mark.asyncio
+@skip_macos
 async def test_iostat_stats(async_client):
     info = await status.get_iostat_stats()
     assert isinstance(info, status.IostatInfo)
@@ -44,6 +47,7 @@ async def test_power_stats(async_client):
 
 
 @skip_circleci
+@skip_macos
 @pytest.mark.asyncio
 async def test_get_drives_stats(async_client):
     """Minimum drives info testing because this will fail in docker, etc."""

wrolpi/vars.py

@@ -111,5 +111,7 @@ IS_RPI4 = 'Raspberry Pi 4' in RPI_DEVICE_MODEL_CONTENTS if RPI_DEVICE_MODEL_CONT
 IS_RPI5 = 'Raspberry Pi 5' in RPI_DEVICE_MODEL_CONTENTS if RPI_DEVICE_MODEL_CONTENTS else False
 IS_RPI = IS_RPI4 or IS_RPI5
 
+IS_MACOS = os.uname()[0] == 'Darwin'
+
 SIMULTANEOUS_DOWNLOAD_DOMAINS = int(os.environ.get('SIMULTANEOUS_DOWNLOAD_DOMAINS',
                                                    2 if IS_RPI else 4))