60 lines
2.1 KiB
Python
60 lines
2.1 KiB
Python
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
from viano_bug_analysis.lifecycle import FileLock, LifecycleError, LockBusy, Workspace
|
|
from viano_bug_analysis.lifecycle.storage import atomic_json
|
|
|
|
|
|
def test_atomic_json_failure_keeps_old_file(tmp_path, monkeypatch):
|
|
path = tmp_path / "index.json"
|
|
atomic_json(path, {"old": True})
|
|
def fail(*args):
|
|
raise OSError("replace failure")
|
|
monkeypatch.setattr(os, "replace", fail)
|
|
with pytest.raises(LifecycleError):
|
|
atomic_json(path, {"new": True})
|
|
assert json.loads(path.read_text()) == {"old": True}
|
|
assert not list(tmp_path.glob(".json-*"))
|
|
|
|
|
|
def test_case_ids_do_not_become_paths(tmp_path):
|
|
workspace = Workspace(tmp_path / "service")
|
|
path = workspace.case_directory("../../case", "C:/package")
|
|
assert path.is_relative_to(workspace.cases)
|
|
assert len(path.name) == 64
|
|
assert len(path.parent.name) == 64
|
|
|
|
|
|
def test_competing_lock_and_release(tmp_path):
|
|
path = tmp_path / "lock"
|
|
with FileLock(path):
|
|
with pytest.raises(LockBusy):
|
|
with FileLock(path):
|
|
pass
|
|
with FileLock(path):
|
|
pass
|
|
|
|
|
|
def test_lock_released_on_process_exit(tmp_path):
|
|
path = tmp_path / "process.lock"
|
|
script = "from pathlib import Path; from viano_bug_analysis.lifecycle import FileLock; import os,sys; lock=FileLock(Path(sys.argv[1])); lock.__enter__(); print('locked',flush=True); sys.stdin.readline(); os._exit(17)"
|
|
env = dict(os.environ, PYTHONPATH=str(__import__('pathlib').Path(__file__).resolve().parents[1] / "src"))
|
|
process = subprocess.Popen([sys.executable, "-c", script, str(path)], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=env)
|
|
try:
|
|
assert process.stdout.readline().strip() == "locked"
|
|
with pytest.raises(LockBusy):
|
|
with FileLock(path):
|
|
pass
|
|
process.communicate("exit\n", timeout=10)
|
|
assert process.returncode == 17
|
|
with FileLock(path):
|
|
pass
|
|
finally:
|
|
if process.poll() is None:
|
|
process.kill()
|
|
process.wait()
|