01-connect-uname-password-success-no-tls.py 1.3 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647
  1. #!/usr/bin/env python3
  2. # Test whether a connection is denied if it provides a correct username but
  3. # incorrect password.
  4. from mosq_test_helper import *
  5. def write_config(filename, port):
  6. with open(filename, 'w') as f:
  7. f.write("port %d\n" % (port))
  8. f.write("password_file %s\n" % (filename.replace('.conf', '.pwfile')))
  9. f.write("allow_anonymous false\n")
  10. def do_test(proto_ver):
  11. port = mosq_test.get_port()
  12. conf_file = os.path.basename(__file__).replace('.py', '.conf')
  13. write_config(conf_file, port)
  14. rc = 1
  15. keepalive = 10
  16. connect_packet = mosq_test.gen_connect("connect-uname-pwd-test", keepalive=keepalive, username="user", password="password", proto_ver=proto_ver)
  17. connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
  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, port=port)
  21. sock.close()
  22. rc = 0
  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. print("proto_ver=%d" % (proto_ver))
  33. exit(rc)
  34. do_test(proto_ver=4)
  35. do_test(proto_ver=5)
  36. exit(0)