09-plugin-auth-defer-unpwd-fail.py 1.3 KB

1234567891011121314151617181920212223242526272829303132333435363738394041
  1. #!/usr/bin/env python3
  2. # Test whether a connection fail when using a auth_plugin that defer authentication.
  3. from mosq_test_helper import *
  4. def write_config(filename, port, plugin_ver):
  5. with open(filename, 'w') as f:
  6. f.write("port %d\n" % (port))
  7. f.write("auth_plugin c/auth_plugin_v%d.so\n" % (plugin_ver))
  8. f.write("allow_anonymous false\n")
  9. def do_test(plugin_ver):
  10. port = mosq_test.get_port()
  11. conf_file = os.path.basename(__file__).replace('.py', '.conf')
  12. write_config(conf_file, port, plugin_ver)
  13. rc = 1
  14. keepalive = 10
  15. connect_packet = mosq_test.gen_connect("connect-uname-pwd-test", keepalive=keepalive, username="test-username@v2", password="doesNotMatter")
  16. connack_packet = mosq_test.gen_connack(rc=5)
  17. broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
  18. try:
  19. sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=20, port=port)
  20. rc = 0
  21. sock.close()
  22. except mosq_test.TestError:
  23. pass
  24. finally:
  25. os.remove(conf_file)
  26. broker.terminate()
  27. broker.wait()
  28. (stdo, stde) = broker.communicate()
  29. if rc:
  30. print(stde.decode('utf-8'))
  31. exit(rc)
  32. do_test(4)
  33. do_test(5)