mirror of
https://git.zavage.net/Zavage-Software/wabot.git
synced 2026-07-21 13:06:08 -06:00
70 lines
2.7 KiB
Python
70 lines
2.7 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
|
|
from wabot.sessions import SessionRecord, SessionStore
|
|
|
|
|
|
def make_record(name="scraper1", created_at=None):
|
|
return SessionRecord(
|
|
name=name,
|
|
executor_url="http://127.0.0.1:9515",
|
|
session_id="abc123",
|
|
browser="chromium",
|
|
created_at=(created_at or datetime.now(timezone.utc)).isoformat(),
|
|
service_pid=4242,
|
|
service_port=9515,
|
|
)
|
|
|
|
|
|
class TestSessionStore:
|
|
def test_get_missing_returns_none(self, tmp_path):
|
|
store = SessionStore(path=tmp_path / "sessions.json")
|
|
assert store.get("nope") is None
|
|
|
|
def test_save_and_get_round_trip(self, tmp_path):
|
|
store = SessionStore(path=tmp_path / "sessions.json")
|
|
store.save(make_record())
|
|
rec = store.get("scraper1")
|
|
assert rec == make_record(created_at=datetime.fromisoformat(rec.created_at))
|
|
assert rec.session_id == "abc123"
|
|
assert rec.service_pid == 4242
|
|
|
|
def test_save_is_readable_by_a_fresh_store_instance(self, tmp_path):
|
|
path = tmp_path / "sessions.json"
|
|
SessionStore(path=path).save(make_record())
|
|
assert SessionStore(path=path).get("scraper1") is not None
|
|
|
|
def test_names_lists_sessions_sorted(self, tmp_path):
|
|
store = SessionStore(path=tmp_path / "sessions.json")
|
|
store.save(make_record(name="zeta"))
|
|
store.save(make_record(name="alpha"))
|
|
assert store.names() == ["alpha", "zeta"]
|
|
|
|
def test_remove(self, tmp_path):
|
|
store = SessionStore(path=tmp_path / "sessions.json")
|
|
store.save(make_record())
|
|
store.remove("scraper1")
|
|
assert store.get("scraper1") is None
|
|
store.remove("scraper1") # removing twice must not raise
|
|
|
|
def test_stale_records_are_evicted_on_load(self, tmp_path):
|
|
store = SessionStore(path=tmp_path / "sessions.json", max_age=timedelta(days=3))
|
|
old = datetime.now(timezone.utc) - timedelta(days=4)
|
|
store.save(make_record(name="old", created_at=old))
|
|
store.save(make_record(name="fresh"))
|
|
assert store.names() == ["fresh"]
|
|
|
|
def test_corrupt_file_is_moved_aside_not_crashed_on(self, tmp_path):
|
|
path = tmp_path / "sessions.json"
|
|
path.write_text("{ this is not json")
|
|
store = SessionStore(path=path)
|
|
assert store.load() == {}
|
|
assert (tmp_path / "sessions.json.bad").exists()
|
|
store.save(make_record()) # store is usable again
|
|
assert store.get("scraper1") is not None
|
|
|
|
def test_default_path_is_under_platformdirs(self):
|
|
from wabot.sessions import default_store_path
|
|
|
|
assert default_store_path().name == "sessions.json"
|
|
assert "wabot" in str(default_store_path())
|