107 lines
2.9 KiB
Python
107 lines
2.9 KiB
Python
import json
|
|
import os
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from spectra.trigger import (
|
|
_trigger_path,
|
|
cache_dir,
|
|
clear_trigger,
|
|
read_and_clear_trigger,
|
|
read_trigger,
|
|
write_trigger,
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def isolate_cache(monkeypatch, tmp_path):
|
|
cache = tmp_path / "cache"
|
|
cache.mkdir()
|
|
monkeypatch.setenv("SPECTRA_CACHE_DIR", str(cache))
|
|
yield
|
|
monkeypatch.delenv("SPECTRA_CACHE_DIR", raising=False)
|
|
|
|
|
|
class TestWriteRead:
|
|
def test_write_then_read(self):
|
|
write_trigger({"action": "refresh"})
|
|
data = read_trigger()
|
|
assert data == {"action": "refresh"}
|
|
|
|
def test_read_empty(self):
|
|
assert read_trigger() is None
|
|
|
|
def test_write_overwrites(self):
|
|
write_trigger({"action": "refresh"})
|
|
write_trigger({"action": "clear"})
|
|
data = read_trigger()
|
|
assert data == {"action": "clear"}
|
|
|
|
def test_write_complex_data(self):
|
|
write_trigger({"action": "show_upload", "path": "/tmp/test.png", "image_id": 42})
|
|
data = read_trigger()
|
|
assert data["action"] == "show_upload"
|
|
assert data["path"] == "/tmp/test.png"
|
|
assert data["image_id"] == 42
|
|
|
|
|
|
class TestClear:
|
|
def test_clear_trigger(self):
|
|
write_trigger({"action": "refresh"})
|
|
clear_trigger()
|
|
assert read_trigger() is None
|
|
|
|
def test_clear_missing_trigger(self):
|
|
clear_trigger()
|
|
|
|
def test_clear_then_read(self):
|
|
write_trigger({"action": "refresh"})
|
|
clear_trigger()
|
|
assert read_trigger() is None
|
|
|
|
|
|
class TestReadAndClear:
|
|
def test_read_and_clear_returns_data(self):
|
|
write_trigger({"action": "refresh"})
|
|
data = read_and_clear_trigger()
|
|
assert data == {"action": "refresh"}
|
|
|
|
def test_read_and_clear_removes_file(self):
|
|
write_trigger({"action": "refresh"})
|
|
read_and_clear_trigger()
|
|
assert read_trigger() is None
|
|
|
|
def test_read_and_clear_nonexistent(self):
|
|
assert read_and_clear_trigger() is None
|
|
|
|
def test_read_and_clear_is_atomic(self):
|
|
write_trigger({"action": "refresh"})
|
|
data = read_and_clear_trigger()
|
|
assert data == {"action": "refresh"}
|
|
assert read_trigger() is None
|
|
|
|
|
|
class TestCacheDir:
|
|
def test_cache_dir_uses_env(self, monkeypatch):
|
|
monkeypatch.setenv("SPECTRA_CACHE_DIR", "/tmp/spectra-test-cache")
|
|
d = cache_dir()
|
|
assert d == "/tmp/spectra-test-cache"
|
|
|
|
def test_trigger_path_uses_cache_dir(self):
|
|
d = cache_dir()
|
|
path = _trigger_path()
|
|
assert path.startswith(d)
|
|
assert path.endswith("trigger.json")
|
|
|
|
|
|
class TestCorruptFile:
|
|
def test_corrupt_json_returns_none(self):
|
|
path = _trigger_path()
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
with open(path, "w") as f:
|
|
f.write("{invalid json")
|
|
assert read_trigger() is None
|
|
assert read_and_clear_trigger() is None
|