09-plugin-auth-unpwd-success.py 1.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142
  1. #!/usr/bin/env python3
  2. # Test whether a connection is successful with correct username and password
  3. # when using a simple auth_plugin.
  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("allow_anonymous false\n")
  10. def do_test(plugin_ver):
  11. port = mosq_test.get_port()
  12. conf_file = os.path.basename(__file__).replace('.py', '.conf')
  13. write_config(conf_file, port, plugin_ver)
  14. rc = 1
  15. keepalive = 10
  16. connect_packet = mosq_test.gen_connect("connect-uname-pwd-test", keepalive=keepalive, username="test-username", password="cnwTICONIURW")
  17. connack_packet = mosq_test.gen_connack(rc=0)
  18. broker = mosq_test.start_broker(filename=os.path.basename(__file__), use_conf=True, port=port)
  19. try:
  20. sock = mosq_test.do_client_connect(connect_packet, connack_packet, timeout=20, port=port)
  21. rc = 0
  22. sock.close()
  23. except mosq_test.TestError:
  24. pass
  25. finally:
  26. os.remove(conf_file)
  27. broker.terminate()
  28. broker.wait()
  29. (stdo, stde) = broker.communicate()
  30. if rc:
  31. print(stde.decode('utf-8'))
  32. exit(rc)
  33. do_test(4)
  34. do_test(5)