FabInfra-Mirror/api.pyfabapi
1
0
Fork 0
mirror of https://gitlab.com/fabinfra/fabaccess/pyfabapi.git synced 2025-03-12 14:51:42 +01:00
Code Issues Packages Projects Releases Wiki Activity

Update to new API

Browse Source
This commit is contained in:
Nadja Reitzenstein 2022-03-14 00:04:28 +01:00
parent 8189d04d0a
commit 47d22cf973
2 changed files with 9 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
fabapi/connect.py
View File

@ -38,20 +38,17 @@ async def connect(host, port, user, pw):
# Start TwoPartyClient using TwoWayPipe (takes no arguments in this mode) # Start TwoPartyClient using TwoWayPipe (takes no arguments in this mode)
client = capnp.TwoPartyClient() client = capnp.TwoPartyClient()
cap = client.bootstrap().cast_as(connection_capnp.Bootstrap)
# Assemble reader and writer tasks, run in the background # Assemble reader and writer tasks, run in the background
coroutines = [myreader(client, reader), mywriter(client, writer)] coroutines = [myreader(client, reader), mywriter(client, writer)]
asyncio.gather(*coroutines, return_exceptions=True) asyncio.gather(*coroutines, return_exceptions=True)
auth = cap.authenticationSystem().authenticationSystem boot = client.bootstrap().cast_as(connection_capnp.Bootstrap)
req = auth.start_request() auth = await boot.createSession("PLAIN").a_wait()
req.request.mechanism = "PLAIN" p = "\0" + user + "\0" + pw
req.request.initialResponse.initial = "\0" + user + "\0" + pw response = await auth.authentication.step(p).a_wait()
rep_prom = req.send() if response.which() == 'successful':
response = await rep_prom.a_wait() return response.successful.session
if response.response.outcome.result == "successful":
return cap
else: else:
print("Auth failed") print("Authentication failed!")
return None return None

8
test.py
View File

@ -3,12 +3,8 @@ import fabapi
async def main(): async def main():
boot = await fabapi.connect("localhost", 59661, "Testuser", "secret") session = await fabapi.connect("localhost", 59661, "Testuser", "secret")
if boot: print(session)
ms = await boot.machineSystem().a_wait()
info = await ms.machineSystem.info().a_wait()
ma = await info.info.getMachineURN("urn:fabaccess:resource:Another").a_wait()
print(ma.machine)
if __name__ == "__main__": if __name__ == "__main__":
loop = asyncio.get_event_loop() loop = asyncio.get_event_loop()