60 lines
1.6 KiB
Python
60 lines
1.6 KiB
Python
import pytest
|
|
from target_simulator.utils.tftp_client import TFTPClient, TFTPError
|
|
import io
|
|
|
|
|
|
def test_tftp_client_init():
|
|
client = TFTPClient(server_ip="127.0.0.1", server_port=69)
|
|
assert client.server_ip == "127.0.0.1"
|
|
assert client.server_port == 69
|
|
|
|
|
|
def test_tftp_client_validate_params():
|
|
client = TFTPClient(server_ip="127.0.0.1")
|
|
# filename valido
|
|
client._validate_params("file.txt", "octet")
|
|
# filename non valido
|
|
with pytest.raises(ValueError):
|
|
client._validate_params("", "octet")
|
|
# mode non valido
|
|
with pytest.raises(ValueError):
|
|
client._validate_params("file.txt", "invalid_mode")
|
|
|
|
|
|
def test_tftp_error():
|
|
err = TFTPError(1, "fail")
|
|
assert str(err) == "TFTP Error 1: fail"
|
|
assert err.code == 1
|
|
assert err.message == "fail"
|
|
|
|
|
|
def test_tftp_client_upload_mock(monkeypatch):
|
|
client = TFTPClient(server_ip="127.0.0.1")
|
|
|
|
# Mock socket and fileobj
|
|
class DummySocket:
|
|
def __init__(self):
|
|
self.calls = 0
|
|
|
|
def settimeout(self, t):
|
|
pass
|
|
|
|
def sendto(self, data, addr):
|
|
pass
|
|
|
|
def recvfrom(self, bufsize):
|
|
self.calls += 1
|
|
if self.calls == 1:
|
|
return (b"\x00\x04", ("127.0.0.1", 69)) # ACK(0) corto
|
|
else:
|
|
# ACK valido per blocchi successivi
|
|
return (b"\x00\x04\x00\x01", ("127.0.0.1", 69))
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
monkeypatch.setattr("socket.socket", lambda *a, **kw: DummySocket())
|
|
fileobj = io.BytesIO(b"testdata")
|
|
# Should not raise
|
|
client.upload("file.txt", fileobj, mode="octet")
|