137 lines
3.3 KiB
Python
137 lines
3.3 KiB
Python
import logging
|
|
import pytest
|
|
|
|
from django.utils import timezone as tz
|
|
|
|
from aircox.models import Sound
|
|
from aircox.controllers import sound_monitor
|
|
from aircox.test import Interface
|
|
|
|
|
|
now = tz.datetime.now()
|
|
|
|
|
|
@pytest.fixture
|
|
def event():
|
|
return Interface(src_path="/tmp/src_path", dest_path="/tmp/dest_path")
|
|
|
|
|
|
@pytest.fixture
|
|
def logger():
|
|
logger = Interface(
|
|
logging, {"info": None, "debug": None, "error": None, "warning": None}
|
|
)
|
|
print("logger", logger)
|
|
return logger
|
|
|
|
|
|
@pytest.fixture
|
|
def interfaces(logger):
|
|
return {
|
|
"SoundFile": Interface.inject(
|
|
sound_monitor,
|
|
"SoundFile",
|
|
{
|
|
"sync": None,
|
|
},
|
|
),
|
|
"time": Interface.inject(
|
|
sound_monitor,
|
|
"time",
|
|
{
|
|
"sleep": None,
|
|
},
|
|
),
|
|
"datetime": Interface.inject(sound_monitor, "datetime", {now: now}),
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def task(interfaces):
|
|
return sound_monitor.Task()
|
|
|
|
|
|
@pytest.fixture
|
|
def delete_task(interfaces):
|
|
return sound_monitor.DeleteTask()
|
|
|
|
|
|
@pytest.fixture
|
|
def move_task(interfaces):
|
|
return sound_monitor.MoveTask()
|
|
|
|
|
|
@pytest.fixture
|
|
def modified_task(interfaces):
|
|
return sound_monitor.ModifiedTask()
|
|
|
|
|
|
class TestTask:
|
|
def test___init__(self, task):
|
|
assert task.timestamp is not None
|
|
|
|
def test_ping(self, task):
|
|
task.timestamp = None
|
|
task.ping()
|
|
print("---", task.timestamp, now)
|
|
assert task.timestamp >= now
|
|
|
|
@pytest.mark.django_db
|
|
def test___call__(self, logger, task, event):
|
|
task.log_msg = "--{event.src_path}--"
|
|
sound_file = task(event, logger=logger, kw=13)
|
|
assert sound_file._trace("sync", kw=True) == {"kw": 13}
|
|
assert logger._trace("info", args=True) == (
|
|
task.log_msg.format(event=event),
|
|
)
|
|
|
|
|
|
class TestDeleteTask:
|
|
@pytest.mark.django_db
|
|
def test___call__(self, delete_task, logger, task, event):
|
|
sound_file = delete_task(event, logger=logger)
|
|
assert sound_file._trace("sync", kw=True) == {"deleted": True}
|
|
|
|
|
|
class TestMoveTask:
|
|
@pytest.mark.django_db
|
|
def test__call___with_sound(self, move_task, sound, event, logger):
|
|
event.src_path = sound.file.name
|
|
sound_file = move_task(event, logger=logger)
|
|
assert isinstance(sound_file._trace("sync", kw="sound"), Sound)
|
|
assert sound_file.path == sound.file.name
|
|
|
|
@pytest.mark.django_db
|
|
def test__call___no_sound(self, move_task, event, logger):
|
|
sound_file = move_task(event, logger=logger)
|
|
assert sound_file._trace("sync", kw=True) == {}
|
|
assert sound_file.path == event.dest_path
|
|
|
|
|
|
class TestModifiedTask:
|
|
def test_wait(self, modified_task):
|
|
dt_now = now + modified_task.timeout_delta - tz.timedelta(hours=10)
|
|
datetime = Interface.inject(sound_monitor, "datetime", {"now": dt_now})
|
|
|
|
def sleep(n):
|
|
datetime._imeta.funcs[
|
|
"now"
|
|
] = modified_task.timestamp + tz.timedelta(hours=10)
|
|
|
|
time = Interface.inject(sound_monitor, "time", {"sleep": sleep})
|
|
modified_task.wait()
|
|
assert time._trace("sleep", args=True)
|
|
|
|
datetime._imeta.release()
|
|
|
|
def test__call__(self):
|
|
pass
|
|
|
|
|
|
class TestMonitorHandler:
|
|
pass
|
|
|
|
|
|
class SoundMonitor:
|
|
pass
|