wabot/tests/unit/test_sessions.py

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())

    def test_naive_created_at_is_treated_as_utc_not_crash(self, tmp_path):
        # legacy/hand-written records may carry naive timestamps
        import json

        path = tmp_path / "sessions.json"
        record = make_record()
        data = {"scraper1": {**record.__dict__, "created_at": "2999-01-01T00:00:00"}}
        path.write_text(json.dumps(data))
        store = SessionStore(path=path)
        assert store.get("scraper1") is not None  # fresh: kept, assumed UTC

    def test_garbage_created_at_evicts_only_that_record(self, tmp_path):
        import json

        path = tmp_path / "sessions.json"
        good, bad = make_record(name="good"), make_record(name="bad")
        data = {
            "good": good.__dict__,
            "bad": {**bad.__dict__, "created_at": "not-a-date"},
        }
        path.write_text(json.dumps(data))
        store = SessionStore(path=path)
        assert store.names() == ["good"]  # no crash, bad evicted, file rewritten
        assert "not-a-date" not in path.read_text()

    def test_null_created_at_evicts_only_that_record(self, tmp_path):
        import json

        path = tmp_path / "sessions.json"
        data = {"broken": {**make_record(name="broken").__dict__, "created_at": None}}
        path.write_text(json.dumps(data))
        assert SessionStore(path=path).names() == []

    def test_unknown_record_fields_are_ignored_not_fatal(self, tmp_path):
        # a newer wabot may add fields; older versions must not quarantine the store
        import json

        path = tmp_path / "sessions.json"
        data = {"scraper1": {**make_record().__dict__, "future_field": 42}}
        path.write_text(json.dumps(data))
        assert SessionStore(path=path).get("scraper1") is not None

    def test_repeat_corruption_with_existing_bad_file_does_not_crash(self, tmp_path):
        path = tmp_path / "sessions.json"
        (tmp_path / "sessions.json.bad").write_text("old corruption")
        path.write_text("{ corrupt again")
        store = SessionStore(path=path)
        assert store.load() == {}
        store.save(make_record())
        assert store.get("scraper1") is not None

    def test_write_is_atomic_no_tmp_leftover(self, tmp_path):
        store = SessionStore(path=tmp_path / "sessions.json")
        store.save(make_record())
        assert not (tmp_path / "sessions.json.tmp").exists()