85 lines
2.1 KiB
Python
85 lines
2.1 KiB
Python
# JTFTP - Python/AsyncIO TFTP Server
|
|
# Copyright (C) 2022 Jeffrey C. Ollie
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
|
|
|
import time
|
|
from itertools import count
|
|
from itertools import islice
|
|
|
|
import pytest
|
|
from jtftp.util import iterlast
|
|
from jtftp.util import timed_caller
|
|
|
|
|
|
def test_iterlast_3():
|
|
assert list(iterlast([1, 2, 3])) == [(False, 1), (False, 2), (True, 3)]
|
|
|
|
|
|
def test_iterlast_0():
|
|
assert list(iterlast([])) == []
|
|
|
|
|
|
def test_iterlast_1():
|
|
assert list(iterlast([1])) == [(True, 1)]
|
|
|
|
|
|
def test_iterlast_infinite():
|
|
assert list(islice(iterlast(count(1)), 3)) == [(False, 1), (False, 2), (False, 3)]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_timed_caller():
|
|
latency = 250000
|
|
r = []
|
|
t = [time.monotonic_ns()]
|
|
|
|
async def c():
|
|
end = time.monotonic_ns()
|
|
r.append(("c", (end - t[-1]) < latency))
|
|
t.append(end)
|
|
|
|
async def l():
|
|
end = time.monotonic_ns()
|
|
r.append(("l", (end - t[-1]) < latency))
|
|
t.append(end)
|
|
|
|
await timed_caller((0, 0, 0), c, l)
|
|
assert r == [("c", True), ("c", True), ("l", True)]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_timed_caller_no_timings():
|
|
with pytest.raises(ValueError):
|
|
|
|
async def c():
|
|
pass
|
|
|
|
async def l():
|
|
pass
|
|
|
|
await timed_caller([], c, l)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_timed_caller_negative_timings():
|
|
with pytest.raises(ValueError):
|
|
|
|
async def c():
|
|
pass
|
|
|
|
async def l():
|
|
pass
|
|
|
|
await timed_caller([0, -1, 0], c, l)
|