File size: 2,930 Bytes
b30f068 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 | """Unit tests for the HF Dataset sync — fully mocked, no network / huggingface_hub.
Verifies: the local-dev guard (no token => no-op), pull copying the remote file
in, debounced vs forced push, and flush of pending writes.
"""
from __future__ import annotations
import pytest
pytestmark = pytest.mark.unit
class _FakeApi:
def __init__(self):
self.uploads = 0
self.repos = []
def create_repo(self, **kw):
self.repos.append(kw.get("repo_id"))
def upload_file(self, **kw):
self.uploads += 1
def test_disabled_without_token(tmp_path):
from src.accounts.hf_sync import HFSync
s = HFSync(repo_id="user/data", local_db_path=tmp_path / "x.db", token=None)
assert s.enabled is False
# Every op is a safe no-op.
assert s.pull() is False
assert s.push(force=True) is False
assert s.flush() is False
def test_disabled_without_repo(tmp_path):
from src.accounts.hf_sync import HFSync
s = HFSync(repo_id=None, local_db_path=tmp_path / "x.db", token="tok")
assert s.enabled is False
def test_pull_copies_remote_file(tmp_path):
from src.accounts.hf_sync import HFSync
# Simulate a "remote" file returned by hf_hub_download.
remote = tmp_path / "remote.db"
remote.write_bytes(b"SQLITEDATA")
local = tmp_path / "local" / "accounts.db"
def fake_downloader(**kw):
return str(remote)
s = HFSync(
repo_id="user/data",
local_db_path=local,
token="tok",
downloader=fake_downloader,
)
assert s.enabled is True
assert s.pull() is True
assert local.read_bytes() == b"SQLITEDATA"
def test_pull_missing_remote_is_not_error(tmp_path):
from src.accounts.hf_sync import HFSync
def boom(**kw):
raise FileNotFoundError("not there yet")
s = HFSync(
repo_id="user/data",
local_db_path=tmp_path / "a.db",
token="tok",
downloader=boom,
)
assert s.pull() is False # fresh deploy: no file yet, handled gracefully
def test_push_debounce_and_force(tmp_path):
from src.accounts.hf_sync import HFSync
db = tmp_path / "a.db"
db.write_bytes(b"data")
api = _FakeApi()
s = HFSync(
repo_id="user/data",
local_db_path=db,
token="tok",
min_push_interval=1000, # long window so debounce is observable
api=api,
)
assert s.push(force=True) is True # first forced push succeeds
assert api.uploads == 1
assert s.maybe_push() is False # within debounce window -> skipped, marked pending
assert api.uploads == 1
assert s.flush() is True # pending write gets forced out
assert api.uploads == 2
def test_ensure_repo(tmp_path):
from src.accounts.hf_sync import HFSync
api = _FakeApi()
s = HFSync(repo_id="user/data", local_db_path=tmp_path / "a.db", token="tok", api=api)
s.ensure_repo()
assert "user/data" in api.repos
|