09-plugin-auth-defer-unpwd-success.py 1.4 KB

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