From 056a41c9850393ad1c8f2fdc68dc58459cc470d7 Mon Sep 17 00:00:00 2001
From: Jarrod Johnson <jjohnson2@lenovo.com>
Date: Tue, 7 May 2024 17:00:25 -0400
Subject: [PATCH] Fix client async invocations

---
 confluent_client/bin/confetty     |  6 +++---
 confluent_client/bin/nodediscover | 22 +++++++++++-----------
 2 files changed, 14 insertions(+), 14 deletions(-)

diff --git a/confluent_client/bin/confetty b/confluent_client/bin/confetty
index 012fbdde..fc6dfb3e 100755
--- a/confluent_client/bin/confetty
+++ b/confluent_client/bin/confetty
@@ -484,7 +484,7 @@ async def do_command(command, server):
         startconsole(nodename)
         return
     elif argv[0] == 'set':
-        setvalues(argv[1:])
+        await setvalues(argv[1:])
     elif argv[0] == 'create':
         await createresource(argv[1:])
     elif argv[0] in ('rm', 'delete', 'remove'):
@@ -555,7 +555,7 @@ async def delresource(resname):
     await makecall(session.delete, (resname,))
 
 
-def setvalues(attribs):
+async def setvalues(attribs):
     global exitcode
     if '=' in attribs[0]:  # going straight to attribute
         resource = attribs[0][:attribs[0].index("=")]
@@ -569,7 +569,7 @@ def setvalues(attribs):
     if not keydata:
         return
     targpath = fullpath_target(resource)
-    for res in session.update(targpath, keydata):
+    async for res in session.update(targpath, keydata):
         if 'error' in res:
             if 'errorcode' in res:
                 exitcode = res['errorcode']
diff --git a/confluent_client/bin/nodediscover b/confluent_client/bin/nodediscover
index 48564195..f6c5a7be 100755
--- a/confluent_client/bin/nodediscover
+++ b/confluent_client/bin/nodediscover
@@ -180,7 +180,7 @@ def datum_to_attrib(datum):
 
 unique_fields = frozenset(['serial', 'mac', 'uuid'])
 
-def import_csv(options, session):
+async def import_csv(options, session):
     nodedata = []
     unique_data = {}
     exitcode = 0
@@ -208,7 +208,7 @@ def import_csv(options, session):
             if not datum_complete(nodedatum):
                 sys.exit(1)
             if not search_record(nodedatum, options, session) and not broken:
-                blocking_scan(session)
+                await blocking_scan(session)
             if not search_record(nodedatum, options, session):
                 sys.stderr.write(
                     "Could not match the following data: " +
@@ -271,7 +271,7 @@ async def list_discovery(options, session):
 
 async def clear_discovery(options, session):
     async for mac in list_matching_macs(options, session):
-        for res in session.delete('/discovery/by-mac/{0}'.format(mac)):
+        async for res in session.delete('/discovery/by-mac/{0}'.format(mac)):
             if 'deleted' in res:
                 print('Cleared info for {0}'.format(res['deleted']))
             else:
@@ -311,7 +311,7 @@ async def list_matching_macs(options, session, node=None, checknode=True):
 async def assign_discovery(options, session, needid=True):
     abort = False
     if options.importfile:
-        return import_csv(options, session)
+        return await import_csv(options, session)
     if not options.node:
         sys.stderr.write("Node (-n) must be specified for assignment\n")
         abort = True
@@ -325,7 +325,7 @@ async def assign_discovery(options, session, needid=True):
     matches = [x async for x in list_matching_macs(options, session, None if needid else options.node, False)]
     if not matches:
         # Do a rescan to catch missing requested data
-        blocking_scan(session)
+        await blocking_scan(session)
         matches = [x async for x in list_matching_macs(options, session, None if needid else options.node, False)]
     if not matches:
         sys.stderr.write("No matching discovery candidates found\n")
@@ -343,11 +343,11 @@ async def assign_discovery(options, session, needid=True):
     if exitcode:
         sys.exit(exitcode)
 
-def blocking_scan(session):
-    list(session.update('/discovery/rescan', {'rescan': 'start'}))
-    while(list(session.read('/discovery/rescan'))[0].get('scanning', False)):
+async def blocking_scan(session):
+    list([x async for x in session.update('/discovery/rescan', {'rescan': 'start'})])
+    while(list([x async for x in session.read('/discovery/rescan')])[0].get('scanning', False)):
         time.sleep(0.5)
-    list(session.update('/networking/macs/rescan', {'rescan': 'start'}))
+    list([x async for x in session.update('/networking/macs/rescan', {'rescan': 'start'})])
 
 
 async def main():
@@ -403,7 +403,7 @@ async def main():
     if args[0] == 'list':
         await list_discovery(options, session)
     if args[0] == 'clear':
-        clear_discovery(options, session)
+        await clear_discovery(options, session)
     if args[0] == 'assign':
         await assign_discovery(options, session)
     if args[0] == 'reassign':
@@ -415,7 +415,7 @@ async def main():
     if args[0] == 'unsubscribe':
         subscribe_discovery(options, session, False, args[1])
     if args[0] == 'rescan':
-        blocking_scan(session)
+        await blocking_scan(session)
         print("Rescan complete")