From 184ce3ef3e04f656c14a697c30553aa8857a48c3 Mon Sep 17 00:00:00 2001 From: Josiah Baldwin Date: Mon, 2 Dec 2024 13:39:40 -0800 Subject: [PATCH] Added Session to the __init__ file, and changed docs and test accordingly --- README.rst | 9 ++-- src/meshctrl/exceptions.py | 3 -- .../scripts/meshcentral/users.json | 2 +- tests/test_files.py | 4 +- tests/test_sanity.py | 4 +- tests/test_session.py | 44 +++++++++---------- tests/test_shell.py | 4 +- 7 files changed, 33 insertions(+), 37 deletions(-) diff --git a/README.rst b/README.rst index 822ffbf..c04f039 100644 --- a/README.rst +++ b/README.rst @@ -44,8 +44,7 @@ Usage ----- This module is implemented as a primarily asynchronous library -(asyncio), mostly through the ``Session`` class, which is exported as -default. Because the library is asynchronous, you must wait for it to be +(asyncio), mostly through the `Session `__ class. Because the library is asynchronous, you must wait for it to be initialized before interacting with the server. The preferred way to do this is to use the async context manager pattern: @@ -53,12 +52,12 @@ this is to use the async context manager pattern: import meshctrl - async with meshctrl.session.Session(url, **options): + async with meshctrl.Session(url, **options): print(await session.list_users()) ... However, if you prefer to instantiate the object yourself, you can -simply use the ``initialized`` property: +simply use the `initialized `__ property: .. code:: python @@ -66,7 +65,7 @@ simply use the ``initialized`` property: await session.initialized.wait() Note that, in this case, you will be rquired to clean up tho session -using its ``close`` method. +using its `close `__ method. Session Parameters ------------------ diff --git a/src/meshctrl/exceptions.py b/src/meshctrl/exceptions.py index eff0dfe..43ae378 100644 --- a/src/meshctrl/exceptions.py +++ b/src/meshctrl/exceptions.py @@ -22,9 +22,6 @@ class FileTransferError(MeshCtrlError): Attributes: stats (dict): {"result" (str): Human readable result, "size" (int): number of bytes successfully transferred} - initialized (asyncio.Event): Event marking if the Session initialization has finished. Wait on this to wait for a connection. - alive (bool): Whether the session connection is currently alive - closed (asyncio.Event): Event that occurs when the session closes permanently """ def __init__(self, message, stats): self.stats = stats diff --git a/tests/environment/scripts/meshcentral/users.json b/tests/environment/scripts/meshcentral/users.json index b05a61d..dabcd61 100644 --- a/tests/environment/scripts/meshcentral/users.json +++ b/tests/environment/scripts/meshcentral/users.json @@ -1 +1 @@ -{"admin": "3U6zP4iIes5ISH15XxjYLjJcCdw9jU0m", "privileged": "aiIO0zLMGsU7++FYVDNxhlpYlZ1andRB", "unprivileged": "Cz9OMV1wkVd9pXdWi4lkBAAu6TMt43MA"} \ No newline at end of file +{"admin": "vt9BbctCg59vcuxKPh8v2tbDjudjwyeX", "privileged": "BWl1vhSe0j0lBfoAkx1JLXLBOwIWc0st", "unprivileged": "vCuatfTQGq8bL2pxdvrNzF+Dc4xBq+5Z"} \ No newline at end of file diff --git a/tests/test_files.py b/tests/test_files.py index edfe605..6649573 100644 --- a/tests/test_files.py +++ b/tests/test_files.py @@ -7,7 +7,7 @@ import io import random async def test_commands(env): - async with meshctrl.session.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session: + async with meshctrl.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session: mesh = await admin_session.add_device_group("test", description="This is a test group", amtonly=False, features=0, consent=0, timeout=10) try: with env.create_agent(mesh.short_meshid) as agent: @@ -53,7 +53,7 @@ async def test_commands(env): assert (await admin_session.remove_device_group(mesh.meshid, timeout=10)), "Failed to remove device group" async def test_upload_download(env): - async with meshctrl.session.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session: + async with meshctrl.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session: mesh = await admin_session.add_device_group("test", description="This is a test group", amtonly=False, features=0, consent=0, timeout=10) try: with env.create_agent(mesh.short_meshid) as agent: diff --git a/tests/test_sanity.py b/tests/test_sanity.py index da50652..32482ff 100644 --- a/tests/test_sanity.py +++ b/tests/test_sanity.py @@ -8,14 +8,14 @@ import ssl import requests async def test_sanity(env): - async with meshctrl.session.Session(env.mcurl, user="unprivileged", password=env.users["unprivileged"], ignore_ssl=True) as s: + async with meshctrl.Session(env.mcurl, user="unprivileged", password=env.users["unprivileged"], ignore_ssl=True) as s: print("\ninfo user_info: {}\n".format(await s.user_info())) print("\ninfo server_info: {}\n".format(await s.server_info())) pass async def test_ssl(env): try: - async with meshctrl.session.Session(env.mcurl, user="unprivileged", password=env.users["unprivileged"], ignore_ssl=False) as s: + async with meshctrl.Session(env.mcurl, user="unprivileged", password=env.users["unprivileged"], ignore_ssl=False) as s: pass except* ssl.SSLCertVerificationError: pass diff --git a/tests/test_session.py b/tests/test_session.py index 8ce1a97..3cefa6b 100644 --- a/tests/test_session.py +++ b/tests/test_session.py @@ -8,8 +8,8 @@ import io thisdir = os.path.dirname(os.path.realpath(__file__)) async def test_admin(env): - async with meshctrl.session.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session,\ - meshctrl.session.Session(env.mcurl, user="privileged", password=env.users["privileged"], ignore_ssl=True) as privileged_session: + async with meshctrl.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session,\ + meshctrl.Session(env.mcurl, user="privileged", password=env.users["privileged"], ignore_ssl=True) as privileged_session: admin_users = await admin_session.list_users(timeout=10) print("\ninfo list_users: {}\n".format(admin_users)) try: @@ -34,22 +34,22 @@ async def test_admin(env): async def test_users(env): try: - async with meshctrl.session.Session(env.mcurl[3:], user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session: + async with meshctrl.Session(env.mcurl[3:], user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session: pass except* ValueError: pass else: raise Exception("Connected with bad URL") try: - async with meshctrl.session.Session(env.mcurl, user="admin", ignore_ssl=True) as admin_session: + async with meshctrl.Session(env.mcurl, user="admin", ignore_ssl=True) as admin_session: pass except* meshctrl.exceptions.MeshCtrlError: pass else: raise Exception("Connected with no password") - async with meshctrl.session.Session(env.mcurl+"/", user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session,\ - meshctrl.session.Session(env.mcurl, user="privileged", password=env.users["privileged"], ignore_ssl=True) as privileged_session,\ - meshctrl.session.Session(env.mcurl, user="unprivileged", password=env.users["unprivileged"], ignore_ssl=True) as unprivileged_session: + async with meshctrl.Session(env.mcurl+"/", user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session,\ + meshctrl.Session(env.mcurl, user="privileged", password=env.users["privileged"], ignore_ssl=True) as privileged_session,\ + meshctrl.Session(env.mcurl, user="unprivileged", password=env.users["unprivileged"], ignore_ssl=True) as unprivileged_session: assert len(await admin_session.list_users(timeout=10)) == 3, "Wrong number of users" @@ -74,17 +74,17 @@ async def test_users(env): assert len(await admin_session.list_users(timeout=10)) == 3, "Failed to remove user" async def test_login_token(env): - async with meshctrl.session.Session(env.mcurl, user="unprivileged", password=env.users["unprivileged"], ignore_ssl=True) as s: + async with meshctrl.Session(env.mcurl, user="unprivileged", password=env.users["unprivileged"], ignore_ssl=True) as s: token = await s.add_login_token("test", expire=1, timeout=10) print("\ninfo add_login_token: {}\n".format(token)) - async with meshctrl.session.Session(env.mcurl, user=token["tokenUser"], password=token["tokenPass"], ignore_ssl=True) as s2: + async with meshctrl.Session(env.mcurl, user=token["tokenUser"], password=token["tokenPass"], ignore_ssl=True) as s2: assert (await s2.user_info())["_id"] == (await s.user_info())["_id"], "Login token logged into wrong account" # Wait for the login token to expire await asyncio.sleep(65) try: - async with meshctrl.session.Session(env.mcurl, user=token["tokenUser"], password=token["tokenPass"], ignore_ssl=True) as s2: + async with meshctrl.Session(env.mcurl, user=token["tokenUser"], password=token["tokenPass"], ignore_ssl=True) as s2: pass except: pass @@ -94,7 +94,7 @@ async def test_login_token(env): token = await s.add_login_token("test2", timeout=10) token2 = await s.add_login_token("test3", timeout=10) print("\ninfo add_login_token_no_expire: {}\n".format(token)) - async with meshctrl.session.Session(env.mcurl, user=token["tokenUser"], password=token["tokenPass"], ignore_ssl=True) as s2: + async with meshctrl.Session(env.mcurl, user=token["tokenUser"], password=token["tokenPass"], ignore_ssl=True) as s2: assert (await s2.user_info())["_id"] == (await s.user_info())["_id"], "Login token logged into wrong account" r = await s.list_login_tokens(timeout=10) @@ -107,9 +107,9 @@ async def test_login_token(env): assert len(await s.remove_login_token([token2["name"]], timeout=10)) == 0, "Residual login tokens" async def test_mesh_device(env): - async with meshctrl.session.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session,\ - meshctrl.session.Session(env.mcurl, user="privileged", password=env.users["privileged"], ignore_ssl=True) as privileged_session,\ - meshctrl.session.Session(env.mcurl, user="unprivileged", password=env.users["unprivileged"], ignore_ssl=True) as unprivileged_session: + async with meshctrl.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session,\ + meshctrl.Session(env.mcurl, user="privileged", password=env.users["privileged"], ignore_ssl=True) as privileged_session,\ + meshctrl.Session(env.mcurl, user="unprivileged", password=env.users["unprivileged"], ignore_ssl=True) as unprivileged_session: # Test creating a mesh mesh = await admin_session.add_device_group("test", description="This is a test group", amtonly=False, features=0, consent=0, timeout=10) print("\ninfo add_device_group: {}\n".format(mesh)) @@ -266,9 +266,9 @@ async def test_mesh_device(env): 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): - async with meshctrl.session.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session,\ - meshctrl.session.Session(env.mcurl, user="privileged", password=env.users["privileged"], ignore_ssl=True) as privileged_session,\ - meshctrl.session.Session(env.mcurl, user="unprivileged", password=env.users["unprivileged"], ignore_ssl=True) as unprivileged_session: + async with meshctrl.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session,\ + meshctrl.Session(env.mcurl, user="privileged", password=env.users["privileged"], ignore_ssl=True) as privileged_session,\ + meshctrl.Session(env.mcurl, user="unprivileged", password=env.users["unprivileged"], ignore_ssl=True) as unprivileged_session: user_group = await admin_session.add_user_group("test", description="aoeu") print("\ninfo add_user_group: {}\n".format(user_group)) @@ -294,7 +294,7 @@ async def test_user_groups(env): assert await admin_session.remove_user_group(user_group2.id.split("/")[-1]) async def test_events(env): - async with meshctrl.session.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session: + async with meshctrl.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session: await admin_session.list_events() mesh = await admin_session.add_device_group("test", description="This is a test group", amtonly=False, features=0, consent=0, timeout=10) try: @@ -310,7 +310,7 @@ async def test_events(env): await asyncio.sleep(1) else: break - async with meshctrl.session.Session(env.mcurl, user="privileged", password=env.users["privileged"], ignore_ssl=True) as privileged_session: + async with meshctrl.Session(env.mcurl, user="privileged", password=env.users["privileged"], ignore_ssl=True) as privileged_session: # assert len(await privileged_session.list_events()) == 0, "non-admin user has access to admin events" @@ -337,8 +337,8 @@ async def test_events(env): assert (await admin_session.remove_device_group(mesh.meshid, timeout=10)), "Failed to remove device group" async def test_interuser(env): - async with meshctrl.session.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session,\ - meshctrl.session.Session(env.mcurl, user="privileged", password=env.users["privileged"], ignore_ssl=True) as privileged_session: + async with meshctrl.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session,\ + meshctrl.Session(env.mcurl, user="privileged", password=env.users["privileged"], ignore_ssl=True) as privileged_session: got_message = asyncio.Event() async def _(): async for message in admin_session.events({"action": "interuser"}): @@ -361,7 +361,7 @@ async def test_interuser(env): tg.create_task(asyncio.wait_for(got_message.wait(), 5)) async def test_session_files(env): - async with meshctrl.session.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session: + async with meshctrl.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session: mesh = await admin_session.add_device_group("test", description="This is a test group", amtonly=False, features=0, consent=0, timeout=10) try: with env.create_agent(mesh.short_meshid) as agent: diff --git a/tests/test_shell.py b/tests/test_shell.py index ba919ce..48729fb 100644 --- a/tests/test_shell.py +++ b/tests/test_shell.py @@ -5,7 +5,7 @@ import meshctrl import requests async def test_shell(env): - async with meshctrl.session.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session: + async with meshctrl.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session: mesh = await admin_session.add_device_group("test", description="This is a test group", amtonly=False, features=0, consent=0, timeout=10) try: with env.create_agent(mesh.short_meshid) as agent: @@ -40,7 +40,7 @@ async def test_shell(env): async def test_smart_shell(env): - async with meshctrl.session.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session: + async with meshctrl.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session: mesh = await admin_session.add_device_group("test", description="This is a test group", amtonly=False, features=0, consent=0, timeout=10) try: with env.create_agent(mesh.short_meshid) as agent: