Initial upload: snippet-manager with CI/CD workflow
This commit is contained in:
45
tests/test_sync.py
Normal file
45
tests/test_sync.py
Normal file
@@ -0,0 +1,45 @@
|
||||
"""Tests for P2P sync functionality."""
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
import pytest
|
||||
|
||||
from snip.db.database import Database
|
||||
from snip.sync.protocol import SyncProtocol
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def db():
|
||||
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
|
||||
db_path = f.name
|
||||
database = Database(db_path)
|
||||
database.init_db()
|
||||
yield database
|
||||
os.unlink(db_path)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sync_protocol(db):
|
||||
return SyncProtocol(db, port=18765)
|
||||
|
||||
|
||||
def test_sync_protocol_init(sync_protocol):
|
||||
"""Test sync protocol initialization."""
|
||||
assert sync_protocol.port == 18765
|
||||
assert sync_protocol.server is None
|
||||
|
||||
|
||||
def test_start_stop_server(sync_protocol):
|
||||
"""Test starting and stopping the sync server."""
|
||||
sync_protocol.start_server()
|
||||
assert sync_protocol.server is not None
|
||||
|
||||
sync_protocol.stop_server()
|
||||
assert sync_protocol.server is None
|
||||
|
||||
|
||||
def test_sync_with_peer_no_connection(sync_protocol, db):
|
||||
"""Test sync with unreachable peer."""
|
||||
synced = sync_protocol.sync_with_peer("127.0.0.1", 9999)
|
||||
assert synced == 0
|
||||
Reference in New Issue
Block a user