L'électron rare
43 lines
2.0 KiB
Python
43 lines
2.0 KiB
Python
import os, tempfile, time, json
|
|
os.environ["LISAEL_BASE"] = tempfile.mkdtemp(prefix="transfers_test_")
|
|
os.environ["LISAEL_STAGING"] = os.path.join(os.environ["LISAEL_BASE"], "podcasts")
|
|
os.environ["LISAEL_STAGING_ROUTINES"] = os.path.join(os.environ["LISAEL_BASE"], "routines")
|
|
os.environ["LISAEL_TRANSFERS"] = os.path.join(os.environ["LISAEL_BASE"], "transfers.json")
|
|
import lisael_content as C
|
|
|
|
def test_begin_mark_end_cycle():
|
|
C.transfers_begin("192.168.0.250", [("a.mp3", "routines", 1000),
|
|
("routines.json", "routines", 50)])
|
|
a = C.transfers_active()
|
|
assert a["box_ip"] == "192.168.0.250" and a["total"] == 2 and a["current"] == 0
|
|
assert all(f["status"] == "pending" for f in a["files"])
|
|
C.transfers_mark("a.mp3", "sending")
|
|
assert C.transfers_active()["current"] == 1
|
|
time.sleep(0.05)
|
|
C.transfers_mark("a.mp3", "done")
|
|
f = next(x for x in C.transfers_active()["files"] if x["name"] == "a.mp3")
|
|
assert f["status"] == "done" and f["duration_s"] > 0 and f["rate_bps"] > 0
|
|
C.transfers_mark("routines.json", "sending")
|
|
C.transfers_mark("routines.json", "failed", "boom")
|
|
C.transfers_end()
|
|
assert C.transfers_active() is None
|
|
hist = C.transfers_history()
|
|
names = {h["name"]: h for h in hist}
|
|
assert names["a.mp3"]["status"] == "done"
|
|
assert names["routines.json"]["status"] == "failed" and names["routines.json"]["error"] == "boom"
|
|
|
|
def test_history_truncates_200():
|
|
for i in range(250):
|
|
C.transfers_begin("1.2.3.4", [(f"f{i}.mp3", "routines", 10)])
|
|
C.transfers_mark(f"f{i}.mp3", "sending"); C.transfers_mark(f"f{i}.mp3", "done")
|
|
C.transfers_end()
|
|
assert len(C.transfers_history()) == 200
|
|
# the most recent survive
|
|
assert any(h["name"] == "f249.mp3" for h in C.transfers_history())
|
|
|
|
if __name__ == "__main__":
|
|
for name, fn in sorted(globals().items()):
|
|
if name.startswith("test_"):
|
|
fn(); print("ok", name)
|
|
print("ALL OK")
|