daanselen/pylibmeshctrl
1
0
Fork 0
forked from Narcissus/pylibmeshctrl
Code Issues Pull Requests Activity

Added ability to download files over http(s)

Browse Source 
Also fixed some tests and a couple other bugs
This commit is contained in:
Josiah Baldwin
2024-12-12 16:06:18 -08:00
parent 4eda4e6c08
commit a3c721318d
12 changed files with 166 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
tests/.gitignore vendored
View File

@@ -1,2 +1,2 @@
data
environment/scripts/meshcentral/users.json
/data
/environment/scripts/meshcentral/users.json

6
tests/environment/__init__.py
View File

@@ -43,10 +43,8 @@ class Agent(object):
return self
def __exit__(self, exc_t, exc_v, exc_tb):
try:
requests.post("{self._clienturl}/remove-agent/{self.nodeid}")
except:
pass
requests.post(f"{self._clienturl}/remove-agent/{self.nodeid}")
class TestEnvironment(object):
def __init__(self):

0
tests/environment/meshcentral/data/config.json → tests/environment/config/meshcentral/data/config.json
View File

2
tests/environment/meshcentral.dockerfile
View File

@@ -3,5 +3,5 @@ RUN apk add curl
RUN apk add python3
WORKDIR /opt/meshcentral/
COPY ./scripts/meshcentral ./scripts
COPY ./meshcentral/data /opt/meshcentral/meshcentral-data
COPY ./config/meshcentral/data /opt/meshcentral/meshcentral-data
CMD ["python3", "/opt/meshcentral/scripts/create_users.py"]

20
tests/test_files.py
View File

@@ -5,6 +5,7 @@ import meshctrl
import requests
import io
import random
import time
async def test_commands(env):
async with meshctrl.Session("wss://" + env.dockerurl, user="admin", password=env.users["admin"], ignore_ssl=True, proxy=env.proxyurl) as admin_session:
@@ -78,7 +79,7 @@ async def test_upload_download(env):
async with admin_session.file_explorer(agent.nodeid) as files:
r = await files.upload(upfilestream, f"{pwd}/test", timeout=5)
print("\ninfo files_upload: {}\n".format(r))
assert r["result"] == "success", "Upload failed"
assert r["result"] == True, "Upload failed"
assert r["size"] == len(randdata), "Uploaded wrong number of bytes"
for f in await files.ls(pwd, timeout=5):
if f["n"] == "test" and f["t"] == meshctrl.constants.FileType.FILE:
@@ -95,10 +96,23 @@ async def test_upload_download(env):
else:
raise Exception("Uploaded file not found")
r = await files.download(f"{pwd}/test", downfilestream, timeout=5)
start = time.perf_counter()
r = await files.download(f"{pwd}/test", downfilestream, skip_ws_attempt=True, timeout=5)
print("\ninfo files_download: {}\n".format(r))
assert r["result"] == "success", "Domnload failed"
assert r["result"] == True, "Domnload failed"
assert r["size"] == len(randdata), "Downloaded wrong number of bytes"
print(f"http download time: {time.perf_counter()-start}")
downfilestream.seek(0)
assert downfilestream.read() == randdata, "Got wrong data back"
downfilestream.seek(0)
start = time.perf_counter()
r = await files.download(f"{pwd}/test", downfilestream, skip_http_attempt=True, timeout=5)
print("\ninfo files_download: {}\n".format(r))
assert r["result"] == True, "Domnload failed"
assert r["size"] == len(randdata), "Downloaded wrong number of bytes"
print(f"ws download time: {time.perf_counter()-start}")
downfilestream.seek(0)
assert downfilestream.read() == randdata, "Got wrong data back"

3
tests/test_sanity.py
View File

@@ -43,7 +43,7 @@ async def test_urlparse():
try:
async with meshctrl.Session("wss://localhost", user="unprivileged", password="Not a real password", ignore_ssl=True) as s:
pass
except* TimeoutError:
except* asyncio.TimeoutError:
#We're not running a server, so timeout is our expected outcome
pass
@@ -52,5 +52,4 @@ async def test_urlparse():
async with meshctrl.Session("https://localhost", user="unprivileged", password="Not a real password", ignore_ssl=True) as s:
pass
except* ValueError:
#We're not running a server, so timeout is our expected outcome
pass

21
tests/test_session.py
View File

@@ -29,7 +29,7 @@ async def test_admin(env):
assert len(no_sessions.keys()) == 0, "non-admin has admin acess"
assert len(admin_users) == len(env.users.keys()), "Admin cannot see correct number of users"
assert len(admin_sessions) == 2, "Admin cannot see correct number of oser sessions"
assert len(admin_sessions) == 2, "Admin cannot see correct number of user sessions"
async def test_auto_reconnect(env):
async with meshctrl.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True, auto_reconnect=True) as admin_session:
@@ -39,6 +39,7 @@ async def test_auto_reconnect(env):
# As above, but with proxy
async with meshctrl.Session("wss://" + env.dockerurl, user="admin", password=env.users["admin"], ignore_ssl=True, auto_reconnect=True, proxy=env.proxyurl) as admin_session:
env.restart_mesh()
for i in range(3):
try:
@@ -284,14 +285,14 @@ async def test_mesh_device(env):
except:
raise Exception("Failed to run command on device after it was moved to a new mesh while having individual device permissions")
r = await admin_session.remove_users_from_device_group((await privileged_session.user_info())["_id"], mesh.meshid, timeout=10)
print("\ninfo remove_users_from_device_group: {}\n".format(r))
assert (await admin_session.remove_users_from_device(agent.nodeid, (await unprivileged_session.user_info())["_id"], timeout=10)), "Failed to remove user from device"
assert (r[(await privileged_session.user_info())["_id"]]["success"]), "Failed to remove user from devcie group"
r = await admin_session.remove_users_from_device_group((await privileged_session.user_info())["_id"], mesh.meshid, timeout=10)
print("\ninfo remove_users_from_device_group: {}\n".format(r))
assert (r[(await privileged_session.user_info())["_id"]]["success"]), "Failed to remove user from device group"
assert (await admin_session.remove_users_from_device(agent.nodeid, (await unprivileged_session.user_info())["_id"], timeout=10)), "Failed to remove user from device"
assert (await admin_session.remove_device_group(mesh.meshid, timeout=10)), "Failed to remove device group"
assert (await admin_session.remove_device_group(mesh2.name, isname=True, timeout=10)), "Failed to remove device group"
assert (await admin_session.remove_device_group(mesh2.name, isname=True, timeout=10)), "Failed to remove device group by name"
assert not (await admin_session.add_users_to_device_group((await privileged_session.user_info())["_id"], mesh.meshid, rights=meshctrl.constants.MeshRights.fullrights, timeout=5))[(await privileged_session.user_info())["_id"]]["success"], "Added user to device group which doesn't exist?"
async def test_user_groups(env):
@@ -416,12 +417,12 @@ async def test_session_files(env):
r = await admin_session.upload(agent.nodeid, upfilestream, f"{pwd}/test", timeout=5)
print("\ninfo files_upload: {}\n".format(r))
assert r["result"] == "success", "Upload failed"
assert r["result"] == True, "Upload failed"
assert r["size"] == len(randdata), "Uploaded wrong number of bytes"
r = await admin_session.upload_file(agent.nodeid, os.path.join(thisdir, "data", "test"), f"{pwd}/test2", timeout=5)
print("\ninfo files_upload: {}\n".format(r))
assert r["result"] == "success", "Upload failed"
assert r["result"] == True, "Upload failed"
assert r["size"] == len(randdata), "Uploaded wrong number of bytes"
s = await admin_session.download(agent.nodeid, f"{pwd}/test", timeout=5)
@@ -437,7 +438,7 @@ async def test_session_files(env):
r = await admin_session.upload_file(agent.nodeid, os.path.join(thisdir, "data", "test"), f"{pwd}/test2", unique_file_tunnel=True, timeout=5)
assert r["result"] == "success", "Upload failed"
assert r["result"] == True, "Upload failed"
assert r["size"] == len(randdata), "Uploaded wrong number of bytes"
await admin_session.download_file(agent.nodeid, f"{pwd}/test2", os.path.join(thisdir, "data", "test"), unique_file_tunnel=True, timeout=5)