mirror of
https://github.com/blw1138/Zordon.git
synced 2026-06-09 13:39:24 -05:00
brett
24eb7b5616
* Add tests and new github workflow * Add new unit tests * Add Github CI workflow * Workflow fix * Add pytest install to workflow file * More CI / test updates * More test cleanup * Whitespace cleanup and link complexity override * More whitespace cleanup * Make lint less strict * More lint tweaks
135 lines
4.5 KiB
Python
135 lines
4.5 KiB
Python
from unittest.mock import patch
|
||
from pathlib import Path
|
||
import pytest
|
||
|
||
from src.render_queue import RenderQueue
|
||
from src.engines.engine_manager import EngineManager
|
||
from src.api.preview_manager import PreviewManager
|
||
from src.utilities.config import Config
|
||
from src.distributed_job_manager import DistributedJobManager
|
||
from src.utilities.zeroconf_server import ZeroconfServer
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Global pubsub patch – real pubsub can fire callbacks across tests.
|
||
# Each service module does `from pubsub import pub` at import time, so we
|
||
# must patch each module-level reference individually.
|
||
# ---------------------------------------------------------------------------
|
||
_PUBSUB_TARGETS = [
|
||
'pubsub.pub',
|
||
'src.render_queue.pub',
|
||
'src.distributed_job_manager.pub',
|
||
'src.utilities.zeroconf_server.pub',
|
||
'src.engines.core.base_worker.pub',
|
||
]
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _patch_pubsub():
|
||
mocks = {}
|
||
patchers = []
|
||
for target in _PUBSUB_TARGETS:
|
||
p = patch(target)
|
||
patchers.append(p)
|
||
mocks[target] = p.start()
|
||
yield mocks.get('pubsub.pub')
|
||
for p in reversed(patchers):
|
||
p.stop()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Temp directory for file-system-dependent tests
|
||
# ---------------------------------------------------------------------------
|
||
@pytest.fixture
|
||
def tmp_workspace(tmp_path: Path) -> Path:
|
||
ws = tmp_path / 'zordon_test'
|
||
ws.mkdir()
|
||
return ws
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Config fixture
|
||
# ---------------------------------------------------------------------------
|
||
@pytest.fixture
|
||
def config_instance(tmp_workspace: Path) -> Config:
|
||
orig = Config._default_instance
|
||
cfg = Config()
|
||
cfg.upload_folder = str(tmp_workspace / 'uploads')
|
||
Config._default_instance = cfg
|
||
Config._sync_class()
|
||
yield cfg
|
||
Config._default_instance = orig
|
||
Config._sync_class()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# EngineManager fixture
|
||
# ---------------------------------------------------------------------------
|
||
@pytest.fixture
|
||
def engine_manager_instance(tmp_workspace: Path) -> EngineManager:
|
||
orig = EngineManager._default_instance
|
||
em = EngineManager()
|
||
em.engines_path = str(tmp_workspace / 'engines')
|
||
EngineManager._default_instance = em
|
||
EngineManager._sync_class()
|
||
yield em
|
||
EngineManager._default_instance = orig
|
||
EngineManager._sync_class()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# RenderQueue fixture
|
||
# ---------------------------------------------------------------------------
|
||
@pytest.fixture
|
||
def render_queue_instance() -> RenderQueue:
|
||
orig = RenderQueue._default_instance
|
||
rq = RenderQueue()
|
||
RenderQueue._default_instance = rq
|
||
RenderQueue._sync_class()
|
||
yield rq
|
||
RenderQueue._default_instance = orig
|
||
RenderQueue._sync_class()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# DistributedJobManager fixture
|
||
# ---------------------------------------------------------------------------
|
||
@pytest.fixture
|
||
def distributed_job_manager_instance() -> DistributedJobManager:
|
||
orig = DistributedJobManager._default_instance
|
||
djm = DistributedJobManager()
|
||
DistributedJobManager._default_instance = djm
|
||
DistributedJobManager._sync_class()
|
||
yield djm
|
||
DistributedJobManager._default_instance = orig
|
||
DistributedJobManager._sync_class()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# PreviewManager fixture
|
||
# ---------------------------------------------------------------------------
|
||
@pytest.fixture
|
||
def preview_manager_instance(tmp_workspace: Path) -> PreviewManager:
|
||
orig = PreviewManager._default_instance
|
||
pm = PreviewManager()
|
||
pm.storage_path = str(tmp_workspace / 'previews')
|
||
PreviewManager._default_instance = pm
|
||
PreviewManager._sync_class()
|
||
yield pm
|
||
PreviewManager._default_instance = orig
|
||
PreviewManager._sync_class()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# ZeroconfServer fixture
|
||
# ---------------------------------------------------------------------------
|
||
@pytest.fixture
|
||
def zeroconf_server_instance() -> ZeroconfServer:
|
||
orig = ZeroconfServer._default_instance
|
||
zs = ZeroconfServer()
|
||
ZeroconfServer._default_instance = zs
|
||
ZeroconfServer._sync_class()
|
||
yield zs
|
||
ZeroconfServer._default_instance = orig
|
||
ZeroconfServer._sync_class()
|